From 68bc67c95783bfc49006e8f10d3a707e91642fa7 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 2 Sep 2026 17:53:18 +0900 Subject: [PATCH 1/2] feat(server): make batches, transcription, audio-output chat SDK-reachable Four operations the stock OpenAI SDK (api_key + base_url only) could not reach, all fixed at the server's own request/response shape rather than by inventing a CO-specific client: - Add POST/GET /v1/batches and POST /v1/batches/{id}/cancel (client.batches.create/retrieve/list/cancel()), wrapping the existing submit_batch/LocalBatchBackend/PgLlmBatchBackend machinery. Input comes from a file already uploaded through /v1/files with purpose=batch; a completed batch's real, already-computed results are reshaped into an OpenAI-shaped output file and registered as a downloadable gateway file (FileRegistry.register_local) so client.files.content() also works for it. Only endpoint=/v1/chat/completions is wired; other endpoints fail closed with invalid_endpoint. - Give POST /v1/audio/transcriptions the same multipart-detection treatment /v1/files already had, translating the SDK's always-multipart audio.transcriptions.create() upload into the existing input_audio JSON shape instead of 415ing. - Let /v1/chat/completions accept modalities=["text","audio"] + audio:{voice,format} -- the only SDK-native way to request audio-output chat completions -- by routing to the same single-agent capability="audio" passthrough /v1/audio/generations already uses. - Document (not implement) why GET /v1/videos (list) stays out of scope: unlike Files/Batches this gateway keeps no locally cached video-job status, so a real list needs per-job provider fan-out, a materially different feature than retrieve. Also fixed the mock file-upload transport (ModelClient.proxy_upload) to return the created_at/status fields the real OpenAI FileObject shape requires -- it was silently unusable by a genuine SDK client, which blocked writing a real test for /v1/files in the first place. tests/test_openai_sdk_compat.py instantiates the actual openai package's client against a live gateway server and drives every fixed operation through the stock SDK's own methods (never hand-rolled JSON), including a minimal test-double file transport that makes upload/download round-trip real bytes so the batch flow is exercised end to end. PR #1012 (chat<->responses shape translation) is a separate gap and is not duplicated here. Co-Authored-By: Claude Sonnet 5 --- .../openai-sdk-batches-audio-modalities.md | 1 + README.md | 8 + contextual_orchestrator/file_registry.py | 61 ++ contextual_orchestrator/orchestrator.py | 5 + contextual_orchestrator/server.py | 703 +++++++++++++++++- pyproject.toml | 3 + tests/test_chat_modalities_http_honesty.py | 41 +- ...modalities_prediction_noop_http_honesty.py | 9 +- tests/test_files_api.py | 25 + tests/test_openai_sdk_compat.py | 302 ++++++++ uv.lock | 175 +++++ 11 files changed, 1286 insertions(+), 47 deletions(-) create mode 100644 CHANGELOG.d/openai-sdk-batches-audio-modalities.md create mode 100644 tests/test_openai_sdk_compat.py diff --git a/CHANGELOG.d/openai-sdk-batches-audio-modalities.md b/CHANGELOG.d/openai-sdk-batches-audio-modalities.md new file mode 100644 index 000000000..9f04ed642 --- /dev/null +++ b/CHANGELOG.d/openai-sdk-batches-audio-modalities.md @@ -0,0 +1 @@ +Made four operations reachable with the unmodified stock OpenAI SDK (`api_key` + `base_url` only). Added a standard `POST/GET /v1/batches` and `POST /v1/batches/{id}/cancel` surface (`client.batches.create/retrieve/list/cancel()`) that wraps the existing `submit_batch`/`LocalBatchBackend`/`PgLlmBatchBackend` machinery, sourced from a file already uploaded through `/v1/files` with `purpose=batch`; a completed batch's real, already-computed results are reshaped into an OpenAI-shaped output file downloadable through the same `client.files.content()` call the SDK already supports. Gave `POST /v1/audio/transcriptions` the same multipart-detection treatment `/v1/files` already had, so `client.audio.transcriptions.create(file=..., model=...)` no longer 415s. Let Chat Completions accept `modalities: ["text","audio"]` + `audio: {voice, format}` (the SDK-native, and only SDK-reachable, way to request spoken-audio chat output) by routing it to the same single-agent `capability="audio"` passthrough `/v1/audio/generations` already used, instead of failing closed on every non-text modality. `GET /v1/videos` (list) stays intentionally unimplemented — see the code comment above `GET /v1/videos/{id}` for why a genuine list needs a different (per-job, provider-fanned) implementation than retrieve. diff --git a/README.md b/README.md index 6eedf5755..0aab8c528 100644 --- a/README.md +++ b/README.md @@ -272,6 +272,14 @@ is read from a **KV config store**, never `os.getenv`. Submit via `POST /api/v1/batch_routing_jobs`, poll `GET /api/v1/batch_routing_jobs/{id}`, retrieve `POST /api/v1/batch_routing_jobs/{id}/results` (which records usage + cost). +- **Standard OpenAI Batch API.** `POST /v1/batches`, `GET /v1/batches/{id}`, + `GET /v1/batches`, and `POST /v1/batches/{id}/cancel` wrap the same batch + machinery above in the real `client.batches.create/retrieve/list/cancel()` + shape the stock OpenAI SDK expects: `input_file_id` is a file already + uploaded through `POST /v1/files` with `purpose=batch` (an OpenAI Batch + input-line JSONL), and a completed batch's `output_file_id` downloads + through the same `client.files.content()` call. Only + `endpoint: "/v1/chat/completions"` is wired today. - **Batch embeddings.** Bulk, latency-tolerant embedding work (e.g. naruon's email-import backfill) submits to `POST /v1/batch/embeddings` (`{model, input|inputs:[...], endpoint, metadata|attribution}`) and polls diff --git a/contextual_orchestrator/file_registry.py b/contextual_orchestrator/file_registry.py index 81f1c0321..d8944c5a5 100644 --- a/contextual_orchestrator/file_registry.py +++ b/contextual_orchestrator/file_registry.py @@ -3,11 +3,18 @@ from __future__ import annotations from dataclasses import dataclass, replace +import base64 import hashlib import json +import time from typing import Any import uuid +# Sentinel ``agent_id``/``provider_file_id`` for a gateway-generated file (for +# example batch output JSONL) that has no upstream provider replica at all — +# its bytes are stored locally instead. See ``register_local``/``is_local``. +_LOCAL_FILE_AGENT_ID = "__local__" + class FileContractError(RuntimeError): """Raised when a provider response cannot become a gateway file resource.""" @@ -49,6 +56,9 @@ def __init__(self, job_registry: Any) -> None: self._owners = job_registry.mapping( "file_owners", decode=lambda raw: FileOwner(**raw) ) + # base64 text (JSON-mapping values must be JSON-serializable), keyed + # by gateway file id. Only populated for locally registered files. + self._content = job_registry.mapping("local_file_content") def register( self, @@ -100,6 +110,56 @@ def register_replicas( self._owners[gateway_file_id] = owner return self.public_response(provider_result, owner) + def register_local( + self, + content: bytes, + *, + filename: str, + purpose: str, + owner_id: str, + ) -> dict[str, Any]: + """Register gateway-computed content (no upstream provider replica). + + Used for artifacts the gateway builds itself — batch-output JSONL is + the first caller — rather than receiving from an upload. There is no + provider file to bind, so ``agent_id``/``provider_file_id`` are the + ``_LOCAL_FILE_AGENT_ID`` sentinel and the bytes are served straight + back out of ``self._content`` (see ``is_local``/``local_content``). + """ + if not owner_id: + raise FileContractError("file owner is unavailable") + gateway_file_id = f"file_{uuid.uuid4().hex}" + document = { + "id": gateway_file_id, + "object": "file", + "bytes": len(content), + "created_at": int(time.time()), + "filename": filename, + "purpose": purpose, + "status": "processed", + } + owner = FileOwner( + gateway_file_id=gateway_file_id, + provider_file_id=gateway_file_id, + agent_id=_LOCAL_FILE_AGENT_ID, + owner_id=owner_id, + agent_affinity_key=_LOCAL_FILE_AGENT_ID, + document=document, + replicas={}, + ) + self._owners[gateway_file_id] = owner + self._content[gateway_file_id] = base64.b64encode(content).decode("ascii") + return dict(document) + + @staticmethod + def is_local(owner: FileOwner) -> bool: + """True when ``owner`` is gateway-generated content, not provider-backed.""" + return owner.agent_id == _LOCAL_FILE_AGENT_ID + + def local_content(self, gateway_file_id: str) -> bytes: + """Return the raw bytes registered for a locally generated file.""" + return base64.b64decode(self._content[gateway_file_id]) + @staticmethod def public_response(document: dict[str, Any], owner: FileOwner) -> dict[str, Any]: """Return a provider document with only the gateway identity exposed.""" @@ -129,6 +189,7 @@ def delete(self, gateway_file_id: str, owner_id: str) -> FileOwner: """Remove and return a principal-owned binding after provider deletion.""" owner = self.owner(gateway_file_id, owner_id) del self._owners[gateway_file_id] + self._content.pop(gateway_file_id, None) return owner def retain_replicas( diff --git a/contextual_orchestrator/orchestrator.py b/contextual_orchestrator/orchestrator.py index a09c4e8f0..3dddb2b0b 100644 --- a/contextual_orchestrator/orchestrator.py +++ b/contextual_orchestrator/orchestrator.py @@ -2591,12 +2591,17 @@ def proxy_upload( ) -> dict[str, Any]: """Stream a seekable multipart body to one validated provider account.""" if agent.base_url.startswith("mock://"): + # created_at/status are required by the real OpenAI FileObject + # shape (client.files.create() return type) -- omitting them left + # the mock transport unusable by a genuine SDK client. return { "id": f"provider_file_{uuid.uuid4().hex}", "object": "file", "bytes": content_length, + "created_at": int(time.time()), "purpose": "user_data", "filename": "upload", + "status": "processed", } api_key = _provider_credential(agent) # pragma: no cover headers = { # pragma: no cover diff --git a/contextual_orchestrator/server.py b/contextual_orchestrator/server.py index eb6a77519..04a23b53b 100644 --- a/contextual_orchestrator/server.py +++ b/contextual_orchestrator/server.py @@ -4,6 +4,7 @@ from dataclasses import dataclass, field from email.message import Message +from email.parser import BytesParser from http.cookies import CookieError, SimpleCookie from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import base64 @@ -97,6 +98,13 @@ MAX_FILE_UPLOAD_BYTES + MAX_FILE_MULTIPART_OVERHEAD_BYTES ) MAX_BATCH_FILE_BYTES = 200 * 1024 * 1024 +# OpenAI's real Batch API caps a batch input file at 50,000 requests. +MAX_BATCH_REQUESTS_PER_FILE = 50_000 +# OpenAI's real Audio API caps a transcription upload at 25 MB. +MAX_AUDIO_UPLOAD_BYTES = 25 * 1024 * 1024 +MAX_AUDIO_UPLOAD_REQUEST_BYTES = ( + MAX_AUDIO_UPLOAD_BYTES + MAX_FILE_MULTIPART_OVERHEAD_BYTES +) def _multipart_upload_metadata(body: Any, content_type: str) -> tuple[str, str, int]: @@ -156,6 +164,48 @@ def _multipart_upload_metadata(body: Any, content_type: str) -> tuple[str, str, return purpose, filename, file_size +def _read_multipart_form(raw: bytes, content_type: str) -> dict[str, Any]: + """Parse a small, fully-buffered ``multipart/form-data`` body into fields. + + A text field decodes to ``str``; a file field (one with a ``filename``) + decodes to a ``(filename, bytes)`` tuple. Built for small request bodies + that are already read into memory in full (Audio uploads, capped at + ``MAX_AUDIO_UPLOAD_BYTES`` by the caller) — the streaming, disk-backed + ``_multipart_upload_metadata`` above stays the path for the much larger + Files uploads. + """ + probe = Message() + probe["content-type"] = content_type + boundary = probe.get_param("boundary", header="content-type") + if not isinstance(boundary, str) or not boundary or len(boundary) > 70: + raise RequestError(400, "invalid_file", "multipart boundary is invalid") + header = f"Content-Type: {content_type}\r\nMIME-Version: 1.0\r\n\r\n".encode("latin-1") + message = BytesParser().parsebytes(header + raw) + if not message.is_multipart(): + raise RequestError(400, "invalid_file", "multipart body is malformed") + fields: dict[str, Any] = {} + for part in message.get_payload(): + field_name = part.get_param("name", header="content-disposition") + if not isinstance(field_name, str) or not field_name: + continue + filename = part.get_filename() + payload = part.get_payload(decode=True) or b"" + if isinstance(filename, str) and filename: + fields[field_name] = (filename, payload) + else: + charset = part.get_content_charset("utf-8") + fields[field_name] = payload.decode(charset, errors="replace").strip() + return fields + + +def _is_multipart_content_type(content_type: str) -> bool: + """True when a request's content-type declares a ``multipart/form-data`` body.""" + return ( + content_type.split(";", 1)[0].strip().lower() == "multipart/form-data" + and "boundary=" in content_type + ) + + class ResponsiveThreadingHTTPServer(ThreadingHTTPServer): """Serve slow upstream calls concurrently without a five-connection backlog. @@ -236,6 +286,15 @@ def server_close(self) -> None: "previous_response_id", "conversation", "truncation", "include", "text", } | OPENAI_PASSTHROUGH_PARAM_KEYS ALLOWED_BATCH_KEYS = {"requests", "attribution", "routing", "model", "zdr_only"} +# OpenAI Batch API create body (client.batches.create()). Only the chat +# endpoint is wired to a request-shaped batch backend today (see +# _parse_batch_input_jsonl); other endpoints fail closed with invalid_endpoint. +ALLOWED_BATCHES_CREATE_KEYS = {"input_file_id", "endpoint", "completion_window", "metadata", "zdr_only"} +BATCH_CREATE_ENDPOINTS = {"/v1/chat/completions"} +_OPENAI_BATCH_TERMINAL_STATUSES = {"completed", "failed", "expired", "cancelled"} +_OPENAI_BATCH_STATUSES = _OPENAI_BATCH_TERMINAL_STATUSES | { + "validating", "in_progress", "finalizing", "cancelling", +} ALLOWED_EMBEDDINGS_BATCH_KEYS = {"model", "input", "inputs", "endpoint", "metadata", "attribution", "user", "encoding_format", "dimensions", "routing", "zdr_only"} ALLOWED_EMBEDDINGS_KEYS = { "model", "input", "encoding_format", "dimensions", "user", "metadata", "attribution", "routing", "zdr_only", @@ -3313,6 +3372,176 @@ def _validate_batch_requests( return batch +def _parse_batch_input_jsonl(raw: bytes, *, zdr_only: bool) -> list[BatchRequest]: + """Parse an OpenAI Batch API input file (JSONL) into internal batch requests. + + Each line must be shaped like a real Batch input line: ``{"custom_id", + "method": "POST", "url": "/v1/chat/completions", "body": {"model", + "messages", ...}}`` -- the same shape ``BatchRequest.to_jsonl_line`` + already writes, so a file this gateway wrote round-trips. + """ + try: + text = raw.decode("utf-8") + except UnicodeDecodeError as exc: + raise RequestError(400, "invalid_file", "batch input file must be UTF-8 JSONL") from exc + lines = [line for line in text.splitlines() if line.strip()] + if not lines: + raise RequestError(400, "invalid_file", "batch input file has no requests") + if len(lines) > MAX_BATCH_REQUESTS_PER_FILE: + raise RequestError( + 400, + "invalid_file", + f"batch input file exceeds {MAX_BATCH_REQUESTS_PER_FILE} requests", + ) + requests: list[BatchRequest] = [] + seen_custom_ids: set[str] = set() + for line_number, line in enumerate(lines, start=1): + try: + entry = json.loads(line) + except json.JSONDecodeError as exc: + raise RequestError(400, "invalid_file", f"line {line_number} is not valid JSON") from exc + if not isinstance(entry, dict): + raise RequestError(400, "invalid_file", f"line {line_number} must be a JSON object") + custom_id = entry.get("custom_id") + if not isinstance(custom_id, str) or not custom_id.strip(): + raise RequestError(400, "invalid_file", f"line {line_number} is missing custom_id") + if len(custom_id) > 64: + raise RequestError( + 400, "invalid_file", f"line {line_number} custom_id must be at most 64 characters" + ) + if custom_id in seen_custom_ids: + raise RequestError(400, "invalid_file", "custom_id values must be unique within a batch") + seen_custom_ids.add(custom_id) + if entry.get("method") != "POST": + raise RequestError(400, "invalid_file", f"line {line_number} method must be POST") + body_obj = entry.get("body") + if not isinstance(body_obj, dict): + raise RequestError(400, "invalid_file", f"line {line_number} body must be an object") + messages = _validate_messages(body_obj.get("messages")) + model = body_obj.get("model", TaskOrchestrator.GATEWAY_DEFAULT_MODEL) + if not isinstance(model, str) or not model.strip(): + raise RequestError(400, "invalid_file", f"line {line_number} model must be a non-empty string") + requests.append( + BatchRequest( + messages=messages, + model=model.strip(), + custom_id=custom_id, + attribution={}, + mode="auto", + zdr_only=zdr_only, + ) + ) + return requests + + +def _resolve_file_download_agent( + orchestrator: Any, files: FileRegistry, gateway_file_id: str, principal_id: str, +) -> tuple[Any, str]: + """Resolve the provider agent + provider-side id backing an owned gateway file. + + Shared by ``GET /v1/files/{id}/content`` and batch input-file ingestion so + both surfaces apply the same replica/affinity matching. + """ + try: + owner = files.owner(gateway_file_id, principal_id) + except KeyError: + raise RequestError(404, "file_not_found", "file was not found") from None + replicas = owner.replicas or { + owner.agent_id: { + "provider_file_id": owner.provider_file_id, + "agent_affinity_key": owner.agent_affinity_key, + } + } + selected = next( + ( + (item, replica["provider_file_id"]) + for item in orchestrator.agents + if (replica := replicas.get(item.id)) is not None + and replica.get("agent_affinity_key") == file_agent_affinity_key(item) + ), + None, + ) + if selected is None: + raise RequestError(503, "file_provider_unavailable", "the file provider is unavailable") + return selected + + +def _co_batch_status_to_openai(raw_status: Any, is_complete: Any) -> str: + """Map one batch backend's poll status to an OpenAI Batch API status string.""" + if isinstance(raw_status, str) and raw_status in _OPENAI_BATCH_STATUSES: + return raw_status + return "completed" if is_complete else "in_progress" + + +def _openai_batch_object(job_id: str, tracked: dict[str, Any], status: str) -> dict[str, Any]: + """Render one tracked batch as an OpenAI Batch API object.""" + created_at = tracked.get("created_at", 0) + request_count = tracked.get("request_count", 0) + return { + "id": job_id, + "object": "batch", + "endpoint": tracked.get("endpoint"), + "errors": None, + "input_file_id": tracked.get("input_file_id"), + "completion_window": tracked.get("completion_window"), + "status": status, + "output_file_id": tracked.get("output_file_id"), + "error_file_id": None, + "created_at": created_at, + "in_progress_at": created_at if status != "validating" else None, + "expires_at": None, + "finalizing_at": None, + "completed_at": tracked.get("completed_at"), + "failed_at": tracked.get("failed_at"), + "expired_at": None, + "cancelling_at": tracked.get("cancelling_at"), + "cancelled_at": tracked.get("cancelled_at"), + "request_counts": { + "total": request_count, + "completed": request_count if status == "completed" else 0, + "failed": request_count if status == "failed" else 0, + }, + "metadata": tracked.get("metadata"), + } + + +def _batch_output_jsonl_line(item: dict[str, Any], *, model: str) -> str: + """Render one ``retrieve_batch`` result row as an OpenAI Batch output line. + + ``item`` carries real, already-computed answers/usage from + ``CostRoutingCoordinator.retrieve_batch`` -- this only reshapes them into + the OpenAI Batch output-file line format, it invents nothing. + """ + prompt_tokens = item.get("prompt_tokens") or 0 + completion_tokens = item.get("completion_tokens") or 0 + body = { + "id": _new_chat_completion_id(), + "object": "chat.completion", + "created": int(time.time()), + "model": model, + "choices": [ + { + "index": 0, + "message": {"role": "assistant", "content": item.get("answer", "")}, + "finish_reason": "stop", + } + ], + "usage": { + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, + "total_tokens": prompt_tokens + completion_tokens, + }, + } + return json.dumps( + { + "id": f"batch_req_{uuid.uuid4().hex}", + "custom_id": item.get("custom_id"), + "response": {"status_code": 200, "request_id": _new_chat_completion_id(), "body": body}, + "error": None, + } + ) + + def _is_token_id_shaped(value: Any) -> bool: """True for numeric token-id shapes (int / whole float), including negatives and bools. @@ -4174,12 +4403,46 @@ def _validate_responses_prediction(body: dict[str, Any]) -> None: ) +def _validate_chat_audio_output_request(body: dict[str, Any]) -> bool: + """True when Chat Completions requests OpenAI's audio-output modalities shape. + + ``modalities: ["text", "audio"]`` plus an ``audio: {voice, format}`` object + is the SDK-native (and only SDK-reachable) way to request spoken-audio chat + output — there is no separate ``client.audio.generations`` method. The + caller routes this case to single-agent passthrough instead of the + multi-agent orchestration path (opaque audio output cannot be merged or + re-verified across agents). Any other ``modalities`` value returns + ``False`` and is left for ``_validate_chat_modalities`` below to + reject/normalize exactly as before. + """ + modalities = body.get("modalities") + if not isinstance(modalities, list): + return False + normalized = {item.strip().lower() for item in modalities if isinstance(item, str)} + if normalized != {"text", "audio"} or len(modalities) != len(normalized): + return False + audio = body.get("audio") + if not isinstance(audio, dict) or not all( + isinstance(audio.get(field), str) and audio[field].strip() + for field in ("voice", "format") + ): + raise RequestError( + 400, + "invalid_audio", + 'audio.voice and audio.format are required when modalities includes "audio"', + ) + body["modalities"] = ["text", "audio"] + return True + + def _validate_chat_modalities(body: dict[str, Any]) -> list[str] | None: - """Chat Completions ``modalities`` — omit or ``["text"]`` only. + """Chat Completions ``modalities`` — omit, ``["text"]``, or audio-output only. - OpenAI selects output types (text/audio) via modalities. This gateway is - text-only; non-text modalities fail closed so clients cannot silently - believe audio (or other) output was applied. + OpenAI selects output types (text/audio) via modalities. This gateway has + no multi-agent verification story for non-text output, so every shape but + plain text or the audio-output pair (handled earlier by + ``_validate_chat_audio_output_request``) fails closed rather than + silently believing an unsupported modality was applied. """ if "modalities" not in body: return None @@ -5487,6 +5750,12 @@ def build_server( coordinator = coordinator or CostRoutingCoordinator(orchestrator) video_jobs = VideoJobRegistry(coordinator.job_registry) files = FileRegistry(coordinator.job_registry) + # OpenAI Batch API (/v1/batches) tracking: job_id -> {owner_id, + # input_file_id, endpoint, completion_window, metadata, created_at, + # request_count, output_file_id, models_by_custom_id, ...}. The batch's + # own lifecycle (submit/poll/retrieve) stays in coordinator._batch_jobs; + # this only carries the OpenAI-object fields that has no other home. + openai_batches = coordinator.job_registry.mapping("openai_batches") configure_telemetry(config=coordinator.config) if clearfolio_url is not None: parsed_viewer = urllib.parse.urlparse(clearfolio_url) @@ -5755,25 +6024,15 @@ def do_GET(self) -> None: # noqa: N802 owner = files.owner(gateway_file_id, principal_id) except KeyError: raise RequestError(404, "file_not_found", "file was not found") from None - replicas = owner.replicas or { - owner.agent_id: { - "provider_file_id": owner.provider_file_id, - "agent_affinity_key": owner.agent_affinity_key, - } - } - selected = next( - ( - (item, replica["provider_file_id"]) - for item in orchestrator.agents - if (replica := replicas.get(item.id)) is not None - and replica.get("agent_affinity_key") == file_agent_affinity_key(item) - ), - None, - ) - if selected is None: - raise RequestError(503, "file_provider_unavailable", "the file provider is unavailable") - agent, provider_file_id = selected + if content_request and files.is_local(owner): + # Gateway-generated content (batch output JSONL) has no + # upstream provider replica at all. + self._send_bytes(files.local_content(gateway_file_id), "application/jsonl") + return if content_request: + agent, provider_file_id = _resolve_file_download_agent( + orchestrator, files, gateway_file_id, principal_id + ) try: raw, content_type = self._run( lambda: orchestrator.client.proxy_get_bytes( @@ -5796,6 +6055,52 @@ def do_GET(self) -> None: # noqa: N802 else: self._send(files.public_response(owner.document, owner)) return + if path == "/v1/batches" or path.startswith("/v1/batches/"): + self._authorize("inference") + principal_id = security.principal_id(self.headers) + if path == "/v1/batches": + items = sorted( + ( + (job_id, tracked) + for job_id, tracked in openai_batches.items() + if tracked.get("owner_id") == principal_id + ), + key=lambda pair: pair[1].get("created_at", 0), + reverse=True, + ) + after = (query.get("after") or [None])[0] + if after is not None: + cursor_index = next( + (index for index, (job_id, _tracked) in enumerate(items) if job_id == after), + None, + ) + items = items[cursor_index + 1 :] if cursor_index is not None else [] + limit = self._parse_positive_int( + (query.get("limit") or [None])[0], "limit", 20, max_value=100 + ) + page = items[:limit] + data = [ + self._batch_document(job_id, principal_id) for job_id, _tracked in page + ] + self._send( + { + "object": "list", + "data": data, + "first_id": data[0]["id"] if data else None, + "last_id": data[-1]["id"] if data else None, + "has_more": len(items) > limit, + } + ) + return + batch_id = path[len("/v1/batches/") :] + if not batch_id or "/" in batch_id: + raise RequestError(400, "invalid_batch", "batch id must be one path segment") + tracked = openai_batches.get(batch_id) + if tracked is None or tracked.get("owner_id") != principal_id: + self._send_error(404, "batch_not_found", f"batch {batch_id} not found") + return + self._send(self._batch_document(batch_id, principal_id)) + return if path.startswith("/v1/batch/embeddings/"): # Embeddings batch polling is an inference-scope surface, so # it is authorized here before the admin gate below. @@ -5810,6 +6115,16 @@ def do_GET(self) -> None: # noqa: N802 except KeyError: self._send_error(404, "embeddings_batch_not_found", f"embeddings batch {batch_id} not found") return + # ``GET /v1/videos`` (client.videos.list(), no id) is + # intentionally not implemented: unlike Files/Batches, this + # gateway keeps no locally cached video-job status document -- + # GET /v1/videos/{id} below always re-fetches live status from + # the owning provider. A genuine list would need per-job + # provider round-trips for every owned job on every call, a + # materially different (and much more expensive) feature than + # single-job retrieve/content; nothing in this product's + # consumers calls it today. Add VideoJobRegistry.list() + + # N-way live status fan-out here if that changes. if path.startswith("/v1/videos/"): self._authorize("inference") principal_id = security.principal_id(self.headers) @@ -6356,6 +6671,12 @@ def do_DELETE(self) -> None: # noqa: N802 owner = files.owner(gateway_file_id, principal_id) except KeyError: raise RequestError(404, "file_not_found", "file was not found") from None + if files.is_local(owner): + # Gateway-generated content has no provider replica to + # delete upstream. + files.delete(gateway_file_id, principal_id) + self._send({"id": gateway_file_id, "object": "file", "deleted": True}) + return replicas = owner.replicas or { owner.agent_id: { "provider_file_id": owner.provider_file_id, @@ -6546,26 +6867,36 @@ def do_POST(self) -> None: # noqa: N802 if saw_failure and every_failure_was_request_too_large: raise RequestError(413, "request_too_large", "request body exceeds every eligible provider limit") raise RequestError(503, "file_provider_unavailable", "no eligible file provider accepted the upload") - scope = ( - "admin" - if path in {"/admin/simulate", "/api/v1/evaluation_runs"} - or path.startswith(("/api/v1/agent_pools/", "/api/v1/model_groups")) - else "inference" - ) - self._authorize(scope, state_changing=True) - large_inference_json = path in { - "/v1/chat/completions", - "/v1/responses", - "/v1/images/generations", - "/v1/videos", - } or path.startswith("/v1/audio/") - body = self._read_json( - max_body_bytes=( - min(security.max_body_bytes, MAX_MULTIMODAL_JSON_BODY_BYTES) - if large_inference_json - else security.max_body_bytes + if path == "/v1/audio/transcriptions" and _is_multipart_content_type( + self.headers.get("content-type", "") + ): + # The stock SDK's audio.transcriptions.create() always sends + # multipart/form-data -- give it the same ingress treatment + # /v1/files gets, then rejoin the shared body/zdr_only/ + # request_policy pipeline below exactly like a JSON body would. + self._authorize("inference", state_changing=True) + body = self._read_multipart_transcription_body() + else: + scope = ( + "admin" + if path in {"/admin/simulate", "/api/v1/evaluation_runs"} + or path.startswith(("/api/v1/agent_pools/", "/api/v1/model_groups")) + else "inference" + ) + self._authorize(scope, state_changing=True) + large_inference_json = path in { + "/v1/chat/completions", + "/v1/responses", + "/v1/images/generations", + "/v1/videos", + } or path.startswith("/v1/audio/") + body = self._read_json( + max_body_bytes=( + min(security.max_body_bytes, MAX_MULTIMODAL_JSON_BODY_BYTES) + if large_inference_json + else security.max_body_bytes + ) ) - ) zdr_only = _validate_zdr_only(body) request_policy = orchestrator.request_policy(zdr_only) request_policy.__enter__() @@ -6816,6 +7147,57 @@ def register_video_job(agent: ModelAgent, provider_result: dict[str, Any]) -> di return if path == "/v1/chat/completions": _reject_unknown_keys(body, ALLOWED_CHAT_KEYS) + if _validate_chat_audio_output_request(body): + # Single-agent OpenAI-shaped passthrough -- the same + # capability="audio" plumbing /v1/audio/generations + # already uses. Opaque audio bytes have no multi-agent + # verification story, so this never enters coordinator + # routing/conduct. + _validate_messages(body.get("messages")) + _validate_chat_model(body) + if body.get("stream"): + # ponytail: audio-output chat completions are + # non-streaming only -- proxy_capability has no + # SSE-audio passthrough. Add one if a caller needs + # streamed audio deltas. + raise RequestError( + 400, + "invalid_stream", + "stream is not supported with audio-output modalities on /v1/chat/completions", + ) + audio_started_at = time.perf_counter() + try: + audio_result = self._run( + lambda: orchestrator.proxy_capability( + body, + capability="audio", + endpoint="chat/completions", + binary=False, + ) + ) + except ValueError as exc: + raise RequestError(400, "invalid_model", str(exc)) from exc + except ProviderRequestTooLargeError as exc: + raise RequestError(413, "request_too_large", str(exc)) from exc + except RuntimeError as exc: + raise RequestError( + 503, + "capability_unavailable", + "no enabled audio-capable model group member is available", + ) from exc + orchestrator.record_analytics_event( + "chat_completion_audio_output", + { + "endpoint_path": "/v1/chat/completions", + "actor_scope": "inference", + "status_code": 200, + "duration_ms": round( + (time.perf_counter() - audio_started_at) * 1000, 2 + ), + }, + ) + self._send(audio_result) + return _validate_chat_audio_web_search_surface(body) _validate_openai_sdk_control_fields(body, endpoint_path="/v1/chat/completions") _validate_tool_resources(body, endpoint_path="/v1/chat/completions") @@ -7491,6 +7873,115 @@ def register_video_job(agent: ModelAgent, provider_result: dict[str, Any]) -> di self._audit_trace_disclosure("/api/v1/batch_routing_jobs/{job_id}/results") self._send(_response_payload(retrieved, include_trace=True)) return + if path == "/v1/batches": + # client.batches.create(input_file_id=..., endpoint=..., + # completion_window=...) -- the stock SDK's only entry + # point for the Batch API. Wraps the same submit_batch + # machinery /api/v1/batch_routing_jobs uses, sourced from + # a file already uploaded through the SDK-compatible + # /v1/files endpoint with purpose=batch. + _reject_unknown_keys(body, ALLOWED_BATCHES_CREATE_KEYS) + input_file_id = body.get("input_file_id") + if not isinstance(input_file_id, str) or not input_file_id.strip(): + raise RequestError( + 400, "invalid_input_file_id", "input_file_id must be a non-empty string" + ) + endpoint = body.get("endpoint") + if endpoint not in BATCH_CREATE_ENDPOINTS: + raise RequestError( + 400, + "invalid_endpoint", + f"endpoint must be one of {sorted(BATCH_CREATE_ENDPOINTS)}", + ) + completion_window = body.get("completion_window", "24h") + if not isinstance(completion_window, str) or not completion_window.strip(): + raise RequestError( + 400, + "invalid_completion_window", + "completion_window must be a non-empty string", + ) + metadata = _validate_openai_metadata(body) + principal_id = security.principal_id(self.headers) + agent, provider_file_id = _resolve_file_download_agent( + orchestrator, files, input_file_id, principal_id + ) + try: + raw, _content_type = self._run( + lambda: orchestrator.client.proxy_get_bytes( + agent, + f"files/{urllib.parse.quote(provider_file_id, safe='')}/content", + max_response_bytes=MAX_BATCH_FILE_BYTES, + ) + ) + except urllib.error.HTTPError as exc: + if exc.code == 404: + raise RequestError( + 404, "file_not_found", "input file content was not found" + ) from exc + raise RequestError( + 503, "file_provider_unavailable", "the file provider is unavailable" + ) from exc + batch_requests = _parse_batch_input_jsonl(raw, zdr_only=zdr_only) + models_by_custom_id = { + request.custom_id: request.model for request in batch_requests + } + try: + job = self._run( + lambda: coordinator.submit_batch( + batch_requests, + metadata={"actor_scope": "inference"}, + owner_id=principal_id, + ) + ) + except InvalidBatchModelError as exc: + raise RequestError(400, "invalid_model", str(exc)) from exc + openai_batches[job.job_id] = { + "owner_id": principal_id, + "input_file_id": input_file_id, + "endpoint": endpoint, + "completion_window": completion_window, + "metadata": metadata, + "created_at": int(time.time()), + "request_count": job.request_count, + "output_file_id": None, + "models_by_custom_id": models_by_custom_id, + } + orchestrator.record_analytics_event( + "openai_batch_created", + { + "endpoint_path": "/v1/batches", + "actor_scope": "inference", + "status_code": 200, + "batch_job_id": job.job_id, + "batch_backend": job.backend, + "request_count": job.request_count, + }, + ) + self._send(self._batch_document(job.job_id, principal_id)) + return + if path.startswith("/v1/batches/") and path.endswith("/cancel"): + # client.batches.cancel(batch_id). + batch_id = path[len("/v1/batches/") : -len("/cancel")] + principal_id = security.principal_id(self.headers) + tracked = openai_batches.get(batch_id) + if tracked is None or tracked.get("owner_id") != principal_id: + self._send_error(404, "batch_not_found", f"batch {batch_id} not found") + return + poll_result = self._run( + lambda: coordinator.poll_batch(batch_id, owner_id=principal_id) + ) + if not poll_result.get("is_complete") and tracked.get("cancelling_at") is None: + # ponytail: neither LocalBatchBackend nor PgLlmBatchBackend + # expose a cancel primitive for chat batches (only the + # provider embeddings backend does) -- record the + # cancellation request locally and surface "cancelling" + # rather than fabricate provider-side cancellation. + # Upgrade path: wire BatchBackend.cancel() through + # pg-llm-batch once that primitive exists. + tracked["cancelling_at"] = int(time.time()) + openai_batches[batch_id] = tracked + self._send(self._batch_document(batch_id, principal_id)) + return if path == "/v1/responses": # The Responses API has no chat-completions verifier equivalent, # so every request is proxied to one agent verbatim. @@ -8135,6 +8626,73 @@ def _run(self, callback: Any) -> dict[str, Any]: finally: security.release_run_slot() + def _materialize_batch_output( + self, job_id: str, tracked: dict[str, Any], owner_id: str + ) -> str: + """Build + register the OpenAI-shaped output file for one completed batch. + + Reshapes ``coordinator.retrieve_batch``'s real, already-computed + results into Batch API output lines and registers them as a + downloadable gateway file (see ``FileRegistry.register_local``). + May raise ``BatchDownloadError`` when the backend's download + explicitly failed -- the caller treats that as batch status + "failed", not as an unrelated request error. + """ + retrieved = self._run( + lambda: coordinator.retrieve_batch(job_id, owner_id=owner_id) + ) + models_by_custom_id = tracked.get("models_by_custom_id") or {} + lines = [ + _batch_output_jsonl_line( + item, + model=models_by_custom_id.get(item.get("custom_id"), "contextual-orchestrator"), + ) + for item in retrieved.get("results", []) + ] + content = ("\n".join(lines) + "\n").encode("utf-8") if lines else b"" + document = files.register_local( + content, + filename=f"{job_id}_output.jsonl", + purpose="batch_output", + owner_id=owner_id, + ) + return document["id"] + + def _batch_document(self, job_id: str, principal_id: str) -> dict[str, Any]: + """Render one tracked batch as a fresh OpenAI Batch object. + + Polls the underlying batch job for live status and, the first time + it observes "completed", materializes the output file. Cancellation + and download-failure state recorded on ``tracked`` win over a raw + poll status. + """ + tracked = openai_batches[job_id] + poll_result = self._run( + lambda: coordinator.poll_batch(job_id, owner_id=principal_id) + ) + status = _co_batch_status_to_openai( + poll_result.get("status"), poll_result.get("is_complete") + ) + if tracked.get("cancelling_at") is not None: + if poll_result.get("is_complete"): + status = "cancelled" + tracked.setdefault("cancelled_at", int(time.time())) + else: + status = "cancelling" + elif tracked.get("failed_at") is not None: + status = "failed" + elif status == "completed" and not tracked.get("output_file_id"): + try: + output_file_id = self._materialize_batch_output(job_id, tracked, principal_id) + except BatchDownloadError: + tracked["failed_at"] = int(time.time()) + status = "failed" + else: + tracked["output_file_id"] = output_file_id + tracked["completed_at"] = int(time.time()) + openai_batches[job_id] = tracked + return _openai_batch_object(job_id, tracked, status) + def _parse_positive_int(self, raw: str | None, field_name: str, default: int, max_value: int | None = None) -> int: value = default if raw is None else int(raw) if value < 1: @@ -8182,6 +8740,69 @@ def _read_json(self, *, max_body_bytes: int | None = None) -> dict[str, Any]: self._request_body_consumed = True return _coerce_json(raw) if raw else {} + def _read_multipart_transcription_body(self) -> dict[str, Any]: + """Translate a multipart Audio transcription upload to the internal shape. + + The stock SDK's ``audio.transcriptions.create()`` always sends a + ``file``/``model``/``language``/``response_format``/``prompt``/ + ``temperature`` multipart form -- there is no JSON-body form of this + call. This reads that upload and builds the ``input_audio: {data, + format}`` body ``_validate_capability_request`` and + ``proxy_capability`` already route, so nothing downstream of ingress + changes. + """ + content_type = self.headers.get("content-type", "") + try: + body_size = _request_body_size(self.headers, MAX_AUDIO_UPLOAD_REQUEST_BYTES) + except RequestError: + self.close_connection = True + raise + raw = self.rfile.read(body_size) + if len(raw) != body_size: + self.close_connection = True + raise RequestError( + 400, + "invalid_request_framing", + "request body ended before content-length", + ) + self._request_body_consumed = True + fields = _read_multipart_form(raw, content_type) + file_field = fields.get("file") + if not isinstance(file_field, tuple): + raise RequestError(400, "invalid_file", "multipart upload requires a file field") + filename, file_bytes = file_field + if not file_bytes: + raise RequestError(400, "invalid_file", "multipart upload requires a non-empty file") + if len(file_bytes) > MAX_AUDIO_UPLOAD_BYTES: + raise RequestError(413, "request_too_large", "audio files may not exceed 25 MB") + audio_format = filename.rsplit(".", 1)[-1].lower() if "." in filename else "" + if not audio_format: + raise RequestError( + 400, "invalid_file", "file field must have a filename with an extension" + ) + request_body: dict[str, Any] = { + "input_audio": { + "data": base64.b64encode(file_bytes).decode("ascii"), + "format": audio_format, + }, + } + model = fields.get("model") + if isinstance(model, str) and model.strip(): + request_body["model"] = model.strip() + for passthrough_field in ("language", "response_format", "prompt"): + value = fields.get(passthrough_field) + if isinstance(value, str) and value: + request_body[passthrough_field] = value + temperature = fields.get("temperature") + if isinstance(temperature, str) and temperature: + try: + request_body["temperature"] = float(temperature) + except ValueError: + raise RequestError( + 400, "invalid_temperature", "temperature must be a number" + ) from None + return request_body + def log_message(self, format: str, *args: object) -> None: """Suppress default request logging to keep service output structured.""" return diff --git a/pyproject.toml b/pyproject.toml index de9f90de1..d62a13a79 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,6 +38,9 @@ queue = [ dev = [ "hypothesis>=6.100", "pytest>=8.0", + # Real openai SDK client, used only to prove genuine SDK compatibility in + # tests (see test_openai_sdk_compat.py) -- never imported by the server. + "openai>=2.0", ] [tool.contextual_orchestrator] diff --git a/tests/test_chat_modalities_http_honesty.py b/tests/test_chat_modalities_http_honesty.py index ee7f19ffc..7a6540a6c 100644 --- a/tests/test_chat_modalities_http_honesty.py +++ b/tests/test_chat_modalities_http_honesty.py @@ -1,4 +1,10 @@ -"""Chat Completions modalities honesty over HTTP (text-only gateway).""" +"""Chat Completions modalities honesty over HTTP. + +Text-only unless a caller opts into OpenAI's audio-output shape +(``modalities: ["text", "audio"]`` + ``audio: {voice, format}``), which +routes to single-agent passthrough exactly like ``/v1/audio/generations`` +(see ``test_openai_sdk_compat.py`` for a real SDK round trip through that +path).""" from __future__ import annotations @@ -87,7 +93,8 @@ def test_http_chat_rejects_modalities_audio() -> None: thread.join(timeout=5) -def test_http_chat_rejects_modalities_text_and_audio() -> None: +def test_http_chat_rejects_modalities_text_and_audio_without_audio_object() -> None: + """Opting into audio output still requires audio.voice and audio.format.""" server, thread, port = _server() try: status, body = _post( @@ -99,7 +106,32 @@ def test_http_chat_rejects_modalities_text_and_audio() -> None: }, ) assert status == 400, body - assert "invalid_modalities" in json.dumps(body) + assert "invalid_audio" in json.dumps(body) + finally: + server.shutdown() + thread.join(timeout=5) + + +def test_http_chat_audio_output_modalities_reach_the_audio_capability_route() -> None: + """A well-shaped audio-output request is not rejected outright. + + modalities=["text","audio"] + audio{voice,format} fails closed on + capability availability (no audio-tagged agent here), not on + modalities/audio shape. + """ + server, thread, port = _server() + try: + status, body = _post( + port, + { + "model": "mock-planner", + "messages": [{"role": "user", "content": "speak"}], + "modalities": ["text", "audio"], + "audio": {"voice": "alloy", "format": "wav"}, + }, + ) + assert status == 503, body + assert "capability_unavailable" in json.dumps(body) finally: server.shutdown() thread.join(timeout=5) @@ -160,7 +192,8 @@ def test_http_chat_accepts_modalities_omitted() -> None: if __name__ == "__main__": test_http_chat_accepts_modalities_text_only() test_http_chat_rejects_modalities_audio() - test_http_chat_rejects_modalities_text_and_audio() + test_http_chat_rejects_modalities_text_and_audio_without_audio_object() + test_http_chat_audio_output_modalities_reach_the_audio_capability_route() test_http_chat_accepts_empty_modalities_as_omit() test_http_chat_rejects_modalities_non_array() test_http_chat_accepts_modalities_omitted() diff --git a/tests/test_empty_modalities_prediction_noop_http_honesty.py b/tests/test_empty_modalities_prediction_noop_http_honesty.py index 8d175a963..7bff2d9f7 100644 --- a/tests/test_empty_modalities_prediction_noop_http_honesty.py +++ b/tests/test_empty_modalities_prediction_noop_http_honesty.py @@ -131,7 +131,12 @@ def test_http_completions_accepts_empty_modalities_and_prediction() -> None: thread.join(timeout=5) -def test_http_chat_still_rejects_audio_modalities() -> None: +def test_http_chat_still_rejects_audio_modalities_without_audio_object() -> None: + """Opting into audio output via modalities still needs audio.voice/format. + + See test_chat_modalities_http_honesty.py for the full audio-output + contract. + """ server, thread, port = _server() try: status, body = _post( @@ -144,7 +149,7 @@ def test_http_chat_still_rejects_audio_modalities() -> None: }, ) assert status == 400, body - assert "invalid_modalities" in json.dumps(body) + assert "invalid_audio" in json.dumps(body) finally: server.shutdown() thread.join(timeout=5) diff --git a/tests/test_files_api.py b/tests/test_files_api.py index bd05e288a..e219c5fcf 100644 --- a/tests/test_files_api.py +++ b/tests/test_files_api.py @@ -125,6 +125,31 @@ def test_file_registry_hides_provider_identity_and_enforces_principal_ownership( assert bindings[public["id"]]["agent-a"]["provider_file_id"] == "provider-secret-id" +def test_file_registry_register_local_has_no_provider_replica() -> None: + """A gateway-generated file (batch output) is served from local content.""" + registry = FileRegistry(JobRegistryFactory()) + document = registry.register_local( + b'{"custom_id": "a"}\n', + filename="batch_output.jsonl", + purpose="batch_output", + owner_id="principal-a", + ) + assert document["id"].startswith("file_") + assert document["purpose"] == "batch_output" + assert document["bytes"] == len(b'{"custom_id": "a"}\n') + + owner = registry.owner(document["id"], "principal-a") + assert registry.is_local(owner) + assert registry.local_content(document["id"]) == b'{"custom_id": "a"}\n' + assert registry.public_response(owner.document, owner)["id"] == document["id"] + + registry.delete(document["id"], "principal-a") + with pytest.raises(KeyError): + registry.owner(document["id"], "principal-a") + with pytest.raises(KeyError): + registry.local_content(document["id"]) + + def test_files_http_upload_list_retrieve_content_and_delete() -> None: """The public Files lifecycle retains opaque provider affinity end to end.""" server = build_server( diff --git a/tests/test_openai_sdk_compat.py b/tests/test_openai_sdk_compat.py new file mode 100644 index 000000000..39cad1f21 --- /dev/null +++ b/tests/test_openai_sdk_compat.py @@ -0,0 +1,302 @@ +"""Genuine OpenAI SDK compatibility for the org gap-review backlog (item 33). + +Every test here instantiates the real ``openai`` package's client +(``openai.OpenAI(api_key=..., base_url=...)``) against a live gateway HTTP +server and drives it through the stock SDK's own methods -- never hand-rolled +JSON -- proving the fixed operations are actually reachable with an +unmodified SDK: + +* ``client.batches.create/retrieve/list/cancel()`` (POST/GET ``/v1/batches``) +* ``client.audio.transcriptions.create(file=..., model=...)`` (always + multipart/form-data -- there is no JSON-body form of this SDK call) +* ``client.chat.completions.create(modalities=["text","audio"], audio=...)`` + (the only SDK-native way to request spoken-audio chat output) + +PR #1012 (chat<->responses shape translation) is a separate, unrelated gap +and is not touched here. +""" + +from __future__ import annotations + +import base64 +import io +import json +import threading + +import openai +import pytest + +from contextual_orchestrator import ModelAgent, TaskOrchestrator +from contextual_orchestrator.server import SecurityConfig, _read_multipart_form, build_server + +_TOKEN = "openai_sdk_compat_token" # noqa: S105 + + +class _FakeFileTransport: + """Minimal in-memory file provider double -- upload really persists bytes. + + The gateway's own mock transport (``ModelClient._mock_raw``/``proxy_upload``) + intentionally returns canned content on download regardless of what was + uploaded (see test_files_api.py), which is fine for shape tests but makes + a batch input file's real JSONL unrecoverable. This test double closes + that loop: upload really stores what was sent, download really returns + it -- exercising the real request/response shapes end to end, per this + task's own "real or a test double at the transport layer" allowance. + """ + + def __init__(self) -> None: + self._store: dict[str, bytes] = {} + + def proxy_upload(self, agent, endpoint, body, *, content_type, content_length, max_response_bytes): + raw = body.read() + fields = _read_multipart_form(raw, content_type) + filename, content = fields["file"] + provider_id = f"provider_file_{len(self._store)}" + self._store[provider_id] = content + return { + "id": provider_id, + "object": "file", + "bytes": len(content), + "created_at": 0, + "purpose": fields.get("purpose", "batch"), + "filename": filename, + "status": "processed", + } + + def proxy_get_bytes(self, agent, endpoint, *, max_response_bytes): + provider_id = endpoint.split("/")[1] + return self._store[provider_id], "application/jsonl" + + +def _start(orchestrator: TaskOrchestrator): + server = build_server(orchestrator, port=0, security=SecurityConfig(auth_token=_TOKEN)) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + return server, thread + + +def _client(port: int) -> openai.OpenAI: + return openai.OpenAI( + api_key=_TOKEN, base_url=f"http://127.0.0.1:{port}/v1", max_retries=0 + ) + + +def _batch_input_jsonl(custom_ids: tuple[str, ...], *, model: str) -> bytes: + lines = [ + json.dumps( + { + "custom_id": custom_id, + "method": "POST", + "url": "/v1/chat/completions", + "body": { + "model": model, + "messages": [{"role": "user", "content": f"question for {custom_id}"}], + }, + } + ) + for custom_id in custom_ids + ] + return ("\n".join(lines) + "\n").encode("utf-8") + + +def test_sdk_batches_create_retrieve_list_cancel_round_trip() -> None: + """client.batches.create/retrieve/list/cancel() against a real HTTP gateway. + + The default coordinator backs chat batches with LocalBatchBackend, which + runs eagerly, so this batch is already "completed" -- with a real, + downloadable output file -- by the time create() returns. + """ + agent = ModelAgent("batch_worker", "mock-batch-model", tags=("reasoning", "files")) + orchestrator = TaskOrchestrator([agent]) + transport = _FakeFileTransport() + orchestrator.client.proxy_upload = transport.proxy_upload # type: ignore[method-assign] + orchestrator.client.proxy_get_bytes = transport.proxy_get_bytes # type: ignore[method-assign] + server, thread = _start(orchestrator) + try: + client = _client(server.server_address[1]) + uploaded = client.files.create( + file=( + "batch_input.jsonl", + io.BytesIO(_batch_input_jsonl(("task_a", "task_b"), model="mock-batch-model")), + "application/jsonl", + ), + purpose="batch", + ) + assert uploaded.id.startswith("file_") + assert uploaded.purpose is not None + + batch = client.batches.create( + input_file_id=uploaded.id, + endpoint="/v1/chat/completions", + completion_window="24h", + ) + assert batch.object == "batch" + assert batch.input_file_id == uploaded.id + assert batch.endpoint == "/v1/chat/completions" + assert batch.status == "completed" + assert batch.request_counts is not None + assert batch.request_counts.total == 2 + assert batch.request_counts.completed == 2 + assert batch.output_file_id is not None + + retrieved = client.batches.retrieve(batch.id) + assert retrieved.id == batch.id + assert retrieved.status == "completed" + assert retrieved.output_file_id == batch.output_file_id + + output_bytes = client.files.content(retrieved.output_file_id).read() + output_lines = [ + json.loads(line) + for line in output_bytes.decode("utf-8").splitlines() + if line.strip() + ] + assert {line["custom_id"] for line in output_lines} == {"task_a", "task_b"} + for line in output_lines: + assert line["response"]["status_code"] == 200 + assert line["response"]["body"]["object"] == "chat.completion" + assert line["response"]["body"]["choices"][0]["message"]["content"] + + listed = client.batches.list(limit=10) + assert batch.id in [item.id for item in listed.data] + + # Cancelling an already-terminal batch is an idempotent no-op, same + # as the real API -- it must not error or fabricate a status change. + cancelled = client.batches.cancel(batch.id) + assert cancelled.id == batch.id + assert cancelled.status == "completed" + finally: + server.shutdown() + thread.join(timeout=5) + + +def test_sdk_batches_create_rejects_unsupported_endpoint() -> None: + """A non-chat batch endpoint fails closed rather than being mishandled.""" + server, thread = _start(TaskOrchestrator([ModelAgent("worker_agent", "mock-model", tags=("files",))])) + try: + client = _client(server.server_address[1]) + uploaded = client.files.create( + file=( + "in.jsonl", + io.BytesIO( + b'{"custom_id":"a","method":"POST","url":"/v1/embeddings","body":{}}\n' + ), + "application/jsonl", + ), + purpose="batch", + ) + with pytest.raises(openai.BadRequestError) as excinfo: + client.batches.create( + input_file_id=uploaded.id, + endpoint="/v1/embeddings", + completion_window="24h", + ) + assert excinfo.value.response.status_code == 400 + finally: + server.shutdown() + thread.join(timeout=5) + + +def test_sdk_batches_retrieve_unknown_id_is_a_real_404() -> None: + server, thread = _start(TaskOrchestrator([ModelAgent("worker_agent", "mock-model", tags=("files",))])) + try: + client = _client(server.server_address[1]) + with pytest.raises(openai.NotFoundError): + client.batches.retrieve("batch_does_not_exist") + finally: + server.shutdown() + thread.join(timeout=5) + + +def test_sdk_audio_transcriptions_create_multipart_upload() -> None: + """client.audio.transcriptions.create() always sends multipart/form-data. + + A test-double transport (matching this repo's own convention for + exercising capability passthrough, see test_multimodal_model_group_http.py) + returns a real Transcription shape and captures the routed payload, so + this proves both genuine SDK parsing and that the multipart upload was + translated into the internal input_audio shape faithfully. + """ + agent = ModelAgent("transcribe_worker", "mock-transcribe", tags=("transcription",)) + orchestrator = TaskOrchestrator([agent]) + captured: dict = {} + + def fake_proxy_send(agent: ModelAgent, endpoint: str, payload: dict) -> dict: + captured["endpoint"] = endpoint + captured["payload"] = payload + return {"text": "mock transcript text"} + + orchestrator.client.proxy_send = fake_proxy_send # type: ignore[method-assign] + server, thread = _start(orchestrator) + try: + client = _client(server.server_address[1]) + audio_bytes = b"RIFF....WAVEfmt fake-but-nonempty audio payload" + transcript = client.audio.transcriptions.create( + file=("clip.wav", io.BytesIO(audio_bytes), "audio/wav"), + model="mock-transcribe", + language="en", + temperature=0.2, + ) + assert transcript.text == "mock transcript text" + assert captured["endpoint"] == "audio/transcriptions" + payload = captured["payload"] + assert payload["model"] == "mock-transcribe" + assert payload["input_audio"]["format"] == "wav" + assert base64.b64decode(payload["input_audio"]["data"]) == audio_bytes + assert payload["language"] == "en" + assert payload["temperature"] == pytest.approx(0.2) + finally: + server.shutdown() + thread.join(timeout=5) + + +def test_sdk_chat_completions_with_audio_output_modalities() -> None: + """chat.completions.create(modalities=["text","audio"], audio={...}). + + The only SDK-native way to request spoken-audio chat output -- there is + no separate client.audio.generations method. Routes to the same + capability="audio" passthrough /v1/audio/generations already uses. + """ + agent = ModelAgent("audio_worker", "mock-audio-preview", tags=("audio",)) + server, thread = _start(TaskOrchestrator([agent])) + try: + client = _client(server.server_address[1]) + completion = client.chat.completions.create( + model="mock-audio-preview", + messages=[{"role": "user", "content": "say hello out loud"}], + modalities=["text", "audio"], + audio={"voice": "alloy", "format": "wav"}, + ) + assert completion.object == "chat.completion" + assert completion.model == "mock-audio-preview" + assert completion.choices[0].message.content + finally: + server.shutdown() + thread.join(timeout=5) + + +def test_sdk_chat_completions_audio_output_requires_no_audio_capable_agent_fails_closed() -> None: + """Without an audio-capable agent, the SDK sees a real 503 -- not silence.""" + server, thread = _start( + TaskOrchestrator([ModelAgent("text_only", "mock-planner", tags=("reasoning",))]) + ) + try: + client = _client(server.server_address[1]) + with pytest.raises(openai.APIStatusError) as excinfo: + client.chat.completions.create( + model="mock-planner", + messages=[{"role": "user", "content": "speak"}], + modalities=["text", "audio"], + audio={"voice": "alloy", "format": "wav"}, + ) + assert excinfo.value.response.status_code == 503 + finally: + server.shutdown() + thread.join(timeout=5) + + +if __name__ == "__main__": + for name, fn in sorted(globals().items()): + if name.startswith("test_") and callable(fn): + fn() + print(f"ok {name}") + print("ok") diff --git a/uv.lock b/uv.lock index 7a11636f2..34a1d5641 100644 --- a/uv.lock +++ b/uv.lock @@ -422,6 +422,7 @@ test = [ [package.dev-dependencies] dev = [ { name = "hypothesis" }, + { name = "openai" }, { name = "pytest" }, ] @@ -448,6 +449,7 @@ provides-extras = ["test", "api", "db", "fuzz", "queue"] [package.metadata.requires-dev] dev = [ { name = "hypothesis", specifier = ">=6.100" }, + { name = "openai", specifier = ">=2.0" }, { name = "pytest", specifier = ">=8.0" }, ] @@ -662,6 +664,45 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/04/4b/29cac41a4d98d144bf5f6d33995617b185d14b22401f75ca86f384e87ff1/h11-0.16.0-py3-none-any.whl", hash = "sha256:63cf8bbe7522de3bf65932fda1d9c2772064ffb3dae62d55932da54b31cb6c86", size = 37515, upload-time = "2025-04-24T03:35:24.344Z" }, ] +[[package]] +name = "httpcore2" +version = "2.12.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "h11" }, + { name = "truststore" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/be/ad/f4f0e57345f1870f3e8cb624e058d7eca6e5a27d33bcc3311d9b618734cd/httpcore2-2.12.0.tar.gz", hash = "sha256:9293522bba0aa7c4c8e9e3f040c16575bd8868e155a77fa30c7a9085a5eae648", size = 67548, upload-time = "2026-08-18T13:22:08.211Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d2/74/d370e55600d9bcfa0d9794b0166126d49291a3d2b20c268fc98c453a4948/httpcore2-2.12.0-py3-none-any.whl", hash = "sha256:7e04258ce01013d7d615e5b910a3b27fac937d7a95038227e79652b4ba3b4ceb", size = 83074, upload-time = "2026-08-18T13:22:05.854Z" }, +] + +[[package]] +name = "httpx2" +version = "2.12.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "anyio", marker = "sys_platform != 'emscripten'" }, + { name = "httpcore2", marker = "sys_platform != 'emscripten'" }, + { name = "httpx2-jsfetch", marker = "python_full_version >= '3.12' and sys_platform == 'emscripten'" }, + { name = "idna" }, + { name = "truststore", marker = "sys_platform != 'emscripten'" }, + { name = "typing-extensions", marker = "python_full_version < '3.13'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/7f/f8/579a8b51e42e38ee32647df9f08aa25643ae788e275cc625b199829c4671/httpx2-2.12.0.tar.gz", hash = "sha256:7631fe9887a8a2275f4a2540e053aa670fcc50742864a9ae7c66e609fdcf12cf", size = 100040, upload-time = "2026-08-18T13:22:09.086Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c8/95/411ba65569158e862368917aaf56597f3e5fa3b91b0502919638465a08f3/httpx2-2.12.0-py3-none-any.whl", hash = "sha256:cc8b6eecb8661c146b8f89a60e97456ee086e91a784ed31ac450c3a9e613dd36", size = 95427, upload-time = "2026-08-18T13:22:06.834Z" }, +] + +[[package]] +name = "httpx2-jsfetch" +version = "1.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/cd/c4/0e5636363151a2a1795e0a77617168b9ca438e1748ec05fc9b5687f93d64/httpx2_jsfetch-1.0.tar.gz", hash = "sha256:70a0e3eabfef7cce5ad9c629f7d01ca05e418f586646f4ddf14782e4c1454c60", size = 6872, upload-time = "2026-08-07T00:13:07.492Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/9b/43/832f631d32e4f1211caa2ba368317739fe71f0b8530e4c9d15dc454bac2a/httpx2_jsfetch-1.0-py3-none-any.whl", hash = "sha256:cb916b707601e69a07721aabc8f3f6659be3a6893bc1ff5c6f9e02241df2da32", size = 6382, upload-time = "2026-08-07T00:13:06.567Z" }, +] + [[package]] name = "hypothesis" version = "6.165.10" @@ -772,6 +813,105 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" }, ] +[[package]] +name = "jiter" +version = "0.16.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/1d/1f/10936e16d8860c70698a1aa939a46aa0224813b782bce4e000e637da0b2d/jiter-0.16.0.tar.gz", hash = "sha256:7b24c3492c5f4f84a37946ad9cf504910cf6a782d6a4e0689b6673c5894b4a1c", size = 176431, upload-time = "2026-06-29T13:05:13.657Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/76/d8/b959609e44012a42b1f3e5ba98ea3b33c7e41e6d4b77cd8f00fd19b1d3ad/jiter-0.16.0-cp310-cp310-macosx_10_12_x86_64.whl", hash = "sha256:c5fc4f8def331036a7b8e981b4347ebe409981edbc8308a5ea842b8c3614fa6c", size = 310082, upload-time = "2026-06-29T13:02:31.356Z" }, + { url = "https://files.pythonhosted.org/packages/c6/3d/4d7f5667ea0e0548534ba880b84bb3d12924fd133aa83ad6c6c80fca3d76/jiter-0.16.0-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:5a71d0d2014c3275043e1170bf3d4e771493cb0dcf07be54c567155f4d8ee64b", size = 315643, upload-time = "2026-06-29T13:02:33.204Z" }, + { url = "https://files.pythonhosted.org/packages/9b/83/bed2dcb5c9f3e1ccfcbc67dda48265fe7d5ad0c9cadda5fe95f6e3b87f94/jiter-0.16.0-cp310-cp310-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:741eed508c233a76313a1c7b001f8f21b82f14327e9196ae8bd29a2cc164ae84", size = 341363, upload-time = "2026-06-29T13:02:34.853Z" }, + { url = "https://files.pythonhosted.org/packages/f4/2f/6bb3c3dda668ebc0445689c81a2b0f26a82b10843d67ed9c9b2c3edc177f/jiter-0.16.0-cp310-cp310-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:3fb7bc819187b56dc48aa5c833aaf92257da8e07efdb9306156667bd2eeb491c", size = 365483, upload-time = "2026-06-29T13:02:36.295Z" }, + { url = "https://files.pythonhosted.org/packages/92/35/8a045ccb39164e70dcdae696413b661771f148b68b12b175c3a04d901937/jiter-0.16.0-cp310-cp310-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:7c9610fd25ebccb43fca584136f5c2fbb26802447eccd430dfdbab95a0fd5126", size = 461219, upload-time = "2026-06-29T13:02:38.116Z" }, + { url = "https://files.pythonhosted.org/packages/e7/99/22292dbbf0ed0c610cfe5ddc7f3bd67237a412f121318f865196e62a07bd/jiter-0.16.0-cp310-cp310-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:4a1d68ff7ca1d3b5dee20a97a3decda7d5f15003823bf6d140c81f8561d3bc5c", size = 374905, upload-time = "2026-06-29T13:02:40.357Z" }, + { url = "https://files.pythonhosted.org/packages/29/ac/2f55ccb1f0eeafa6d89d24caf52f6f0944a59290ee199e9ade62177dca42/jiter-0.16.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:fb08c276dd02dac3a284acdd02cacc630d2e3cd6572a4b85519f35cbd133c3de", size = 348320, upload-time = "2026-06-29T13:02:41.923Z" }, + { url = "https://files.pythonhosted.org/packages/50/e3/7d88b9174c40064fabc07c84a9b62e6b10f5644562ec0e0a29392edbe978/jiter-0.16.0-cp310-cp310-manylinux_2_31_riscv64.whl", hash = "sha256:8fc4d94713c4697347e38faf7d6ef91547c142219bdcfc7220c4870879974244", size = 356519, upload-time = "2026-06-29T13:02:43.436Z" }, + { url = "https://files.pythonhosted.org/packages/27/57/c4a33aeef513a9d5e26e31534e0bcc752d6ea0e54c94ddb7b68bade669c2/jiter-0.16.0-cp310-cp310-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:1a0f05e229edb29e68cdd0ccb83cea13b64263416120cf943767a6fd72e6787f", size = 394204, upload-time = "2026-06-29T13:02:44.987Z" }, + { url = "https://files.pythonhosted.org/packages/9d/70/c6c23e76ebb3766b111bc399437bbc9f870a76e2a92e10b2a5f561d57372/jiter-0.16.0-cp310-cp310-musllinux_1_1_aarch64.whl", hash = "sha256:2c842cbf374a8daf50b2c04212995bee34ca2ac2cdc29a901b4cdb072c9c4131", size = 521477, upload-time = "2026-06-29T13:02:46.724Z" }, + { url = "https://files.pythonhosted.org/packages/2a/d3/0001c8c0c5976af2625bb1cfb1895e8ec693b6589fe4574b8e6fc2c85501/jiter-0.16.0-cp310-cp310-musllinux_1_1_x86_64.whl", hash = "sha256:5ed466aee31294d7cdcd4d37dfe5c42c97bc29d9a5f00eacf24504358309cb9b", size = 552187, upload-time = "2026-06-29T13:02:48.144Z" }, + { url = "https://files.pythonhosted.org/packages/f6/76/311b718e07e85740e48619c0632b36f7e0b8d113984499e436452ed13a9a/jiter-0.16.0-cp310-cp310-win32.whl", hash = "sha256:b42e9ff5376819c053da25809a8d4b6fa6e473b4856ebe42e298ac958be3d7f9", size = 206513, upload-time = "2026-06-29T13:02:49.515Z" }, + { url = "https://files.pythonhosted.org/packages/db/7f/ac680eeb0777dc0eb7dc824800ba27880d7f6bc712e362d34ad8ee559f36/jiter-0.16.0-cp310-cp310-win_amd64.whl", hash = "sha256:10438939205546132189c8e74a2d536a707841f3a25cd7c74ee91fe503407a26", size = 199505, upload-time = "2026-06-29T13:02:50.829Z" }, + { url = "https://files.pythonhosted.org/packages/4e/3f/fae6cc967d120ec89e31c5418a51176d8278b3087fbb384a9176754f353c/jiter-0.16.0-cp311-cp311-macosx_10_12_x86_64.whl", hash = "sha256:67fddeda1688f0cce2d2ae83ccf8a80f79936f2d2997d6cc2261f82fdb54a4d3", size = 309289, upload-time = "2026-06-29T13:02:52.301Z" }, + { url = "https://files.pythonhosted.org/packages/c8/e3/97c6c3562c077f6247d6e6ce5c82562500b6316c0d928e97e106b7a1321a/jiter-0.16.0-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:c90c0f63df322be920eda6ce622e3083d8906ba267f8220fe7873213b8b4430e", size = 315181, upload-time = "2026-06-29T13:02:53.964Z" }, + { url = "https://files.pythonhosted.org/packages/7b/89/d8d073f8aa2667e46c6c0873f86fe4a512bba4293cc730f626a076211a62/jiter-0.16.0-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:64c0203212098470032aabcde9356fc168f377aade3e43def61dfe17e92f2037", size = 340939, upload-time = "2026-06-29T13:02:55.412Z" }, + { url = "https://files.pythonhosted.org/packages/87/c9/db4fda3ed73fb864139305e935e5b8b38a5a24692a5a9dd356c22f1b9c8d/jiter-0.16.0-cp311-cp311-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:12288303c9844e61e1651d02a9a6f6633e47d39f897d6991d1427161ce6b746e", size = 364932, upload-time = "2026-06-29T13:02:57.28Z" }, + { url = "https://files.pythonhosted.org/packages/a2/74/52b5e86241057f52ddd7c9a580f90effb51f9d06239f6fc612279b91a838/jiter-0.16.0-cp311-cp311-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:5cf109d010b4b05a105afb3d43be36a21322d345ad3111e13d15f680afef0e5b", size = 461132, upload-time = "2026-06-29T13:02:58.994Z" }, + { url = "https://files.pythonhosted.org/packages/a9/87/544a700f7447c1f31c5d7833821a4daa5683165c2d5a094fbf5b5800c3dc/jiter-0.16.0-cp311-cp311-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:62c1b7fe1f77925acf5af68b6140b8810fa87dfd4dc0a9c8568ec2fa2a10429c", size = 374857, upload-time = "2026-06-29T13:03:00.455Z" }, + { url = "https://files.pythonhosted.org/packages/40/cd/0fcc3f7d39183674d5bfa9ec640faaeb506c60be7c8f94625dfba366e37c/jiter-0.16.0-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:8597d23c87f59294f83bcb6229b9ed1fccee13dbba967b46930d2f1759466fee", size = 347053, upload-time = "2026-06-29T13:03:02.045Z" }, + { url = "https://files.pythonhosted.org/packages/5c/ae/c7e64e7932ad597fa395b61440b249ada6366716e25c6e08dd2afbd021e6/jiter-0.16.0-cp311-cp311-manylinux_2_31_riscv64.whl", hash = "sha256:3126a5dbad56401989ac769aca0cb56005bfb3e2366eea0ca99d1a91c3c1ee03", size = 356153, upload-time = "2026-06-29T13:03:03.706Z" }, + { url = "https://files.pythonhosted.org/packages/d4/1c/1c719044f14da814e1a060191ab19b96f3e99207bc5b4bfc6d6be34b3f80/jiter-0.16.0-cp311-cp311-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:c4b4717bdb35ae456f831a6b08d01880fff399887a6bbc526a583a406e484eea", size = 393956, upload-time = "2026-06-29T13:03:05.165Z" }, + { url = "https://files.pythonhosted.org/packages/3b/dc/7b2f303a2847207e265503853a2d964a55354cffd62a5f2936c155486798/jiter-0.16.0-cp311-cp311-musllinux_1_1_aarch64.whl", hash = "sha256:adff21bc78edfe086c15eb495b900306076de378dc2337c132401fc39bd79c91", size = 521081, upload-time = "2026-06-29T13:03:06.886Z" }, + { url = "https://files.pythonhosted.org/packages/c2/5f/501cf6e1e09caeb420195179ffc6f62aca603f1220ec53fd80d0d70b3e56/jiter-0.16.0-cp311-cp311-musllinux_1_1_x86_64.whl", hash = "sha256:dab907db06fc593645e73109acf4581ba5b548897d28b9348dc41ddc8343b2d3", size = 552085, upload-time = "2026-06-29T13:03:08.339Z" }, + { url = "https://files.pythonhosted.org/packages/79/54/aa5be86520113b79455c3877f3d1f07a348098df4083ba3688e9537e52dd/jiter-0.16.0-cp311-cp311-win32.whl", hash = "sha256:560b2cf3fb03240cd34f27409a238547488708f05b7c3924f571a60422251ec7", size = 206755, upload-time = "2026-06-29T13:03:09.653Z" }, + { url = "https://files.pythonhosted.org/packages/64/ec/2feb893eb330bd69b413866f4d5daada33c3962f1c6f270c91ca2d87fdf9/jiter-0.16.0-cp311-cp311-win_amd64.whl", hash = "sha256:e431cfc9caf44c1d5459ff77d4e64cbf85fddb6a35dad836a15c6a9ec23087c1", size = 199155, upload-time = "2026-06-29T13:03:10.979Z" }, + { url = "https://files.pythonhosted.org/packages/b9/9c/ca040d94415048a3666fc237774df8151c96f8d2b661cbe3b184acc95876/jiter-0.16.0-cp311-cp311-win_arm64.whl", hash = "sha256:2a8e9e39cf083016137aa5cadafe3188adc2ba6ba1fbf1e5d18889ad3e9ad056", size = 194403, upload-time = "2026-06-29T13:03:12.341Z" }, + { url = "https://files.pythonhosted.org/packages/83/2b/52ace16ed031354f0539749a49e4bf33797d82bea5137910835fa4b09793/jiter-0.16.0-cp312-cp312-macosx_10_12_x86_64.whl", hash = "sha256:67c3bc1760f8c99d805dcab4e644027142a53b1d5d861f18780ebdbd5d40b72a", size = 306943, upload-time = "2026-06-29T13:03:14.035Z" }, + { url = "https://files.pythonhosted.org/packages/94/2e/34957c2c1b661c252ba9bcc60ae0bddc27e0f7202c6073326a13c5390eec/jiter-0.16.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:5af7780e4a26bd7d0d989592bf9ef12ebf806b74ab709223ecca37c749872ea9", size = 307779, upload-time = "2026-06-29T13:03:15.418Z" }, + { url = "https://files.pythonhosted.org/packages/88/6c/59bd309cab4460c54cf1079f3eb7fe7af6a4c895c5c957a53378693bad2b/jiter-0.16.0-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:d5bf78d0e05e45cfdd66558893938d59afe3d1b1a824a202039b20e607d25a72", size = 335826, upload-time = "2026-06-29T13:03:17.11Z" }, + { url = "https://files.pythonhosted.org/packages/3b/8c/f5ef7b65f0df47afa16596969defb281ebb86e96df346d62be6fd853d620/jiter-0.16.0-cp312-cp312-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:f4444a83f946605990c98f625cdd3d2725bfb818158760c5748c653170a20e0e", size = 362573, upload-time = "2026-06-29T13:03:18.781Z" }, + { url = "https://files.pythonhosted.org/packages/2b/0b/ace4354da061ee38844a0c27dc2c21eecd27aea119e8da324bea987522d0/jiter-0.16.0-cp312-cp312-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:3a23f0e4f957e1be65752d2dfac9a5a06b1917af8dc85deb639c3b9d02e31290", size = 457979, upload-time = "2026-06-29T13:03:20.293Z" }, + { url = "https://files.pythonhosted.org/packages/55/40/c0253d3772eb9dcd8e6606ee9b2d53ec8e5b814589c47f140aa585f21eaa/jiter-0.16.0-cp312-cp312-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:c22a488f7b9218e245a0025a9ba6b100e2e54700831cf4cf16833a27fba3ad01", size = 372302, upload-time = "2026-06-29T13:03:21.739Z" }, + { url = "https://files.pythonhosted.org/packages/a8/d2/4839422241aa12860ce597b20068727094ba0bc480723c74924ca5bad483/jiter-0.16.0-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:46add52f4ad47a08bfb1219f3e673da972191489a33016edefdb5ea55bfa8c48", size = 343805, upload-time = "2026-06-29T13:03:23.384Z" }, + { url = "https://files.pythonhosted.org/packages/e2/59/e196888a05befdda7dbe299b722d56f2f6eec65402bc34c0a3306d595feb/jiter-0.16.0-cp312-cp312-manylinux_2_31_riscv64.whl", hash = "sha256:9c8a956fd72c2cf1e730d01ea080341f13aa0a97a4a33b51abebe725b7ae9ca9", size = 351107, upload-time = "2026-06-29T13:03:24.815Z" }, + { url = "https://files.pythonhosted.org/packages/ec/74/4cd9e0fca65232136400354b630fbfcd2de634e22ccbb96567725981b548/jiter-0.16.0-cp312-cp312-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:561926e0573ffe4a32498420a76d64b16c513e1ab413b9d28158a8764ac701e5", size = 388441, upload-time = "2026-06-29T13:03:26.266Z" }, + { url = "https://files.pythonhosted.org/packages/d9/8c/554691e48bc711299c0a293dd8a6179e24b2d66a54dc295421fcf64569c0/jiter-0.16.0-cp312-cp312-musllinux_1_1_aarch64.whl", hash = "sha256:44d019fa8cdaf89bf29c71b39e3712143fdd0ac76725c6ef954f9957a5ea8730", size = 516354, upload-time = "2026-06-29T13:03:28.02Z" }, + { url = "https://files.pythonhosted.org/packages/a4/cb/01e9d69dc2cc6759d4f91e230b34489c4fdb2518992650633f9e20bece89/jiter-0.16.0-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:0df91907609837f33341b8e6fe73b95991fdaa57caf1a0fbd343dffe826f386f", size = 547880, upload-time = "2026-06-29T13:03:29.534Z" }, + { url = "https://files.pythonhosted.org/packages/79/70/2953195f1c6ad00f49fa67e13df7e60acb3dd4f387101bc15abccddd905e/jiter-0.16.0-cp312-cp312-win32.whl", hash = "sha256:51d7b836acb0108d7c77df1742332cac2a1fa04a74d6dacec46e7091f0e91274", size = 203473, upload-time = "2026-06-29T13:03:31.025Z" }, + { url = "https://files.pythonhosted.org/packages/2d/05/2909a8b10699a4d560f8c502b6b2c5f3991b682b1922c1eedda242b225bd/jiter-0.16.0-cp312-cp312-win_amd64.whl", hash = "sha256:1878349266f8ee36ecb1375cc5ba2f115f35fd9f0a1a4119e725e379126647f7", size = 196905, upload-time = "2026-06-29T13:03:32.472Z" }, + { url = "https://files.pythonhosted.org/packages/e9/a9/6b82bb1c8d7790d602489b967b982a909e5d092875a6c2ade96444c8dfc5/jiter-0.16.0-cp312-cp312-win_arm64.whl", hash = "sha256:2ed5738ae4af18271a51a528b8811b0cbfa4a1858de9d83359e4169855d6a331", size = 190618, upload-time = "2026-06-29T13:03:34.672Z" }, + { url = "https://files.pythonhosted.org/packages/91/c0/555fc60473d30d66894ba825e63615e3be7524fac23858356afa7a38906c/jiter-0.16.0-cp313-cp313-macosx_10_12_x86_64.whl", hash = "sha256:41977aa5654023948c2dae2a81cbf9c43343954bef1cd59a154dd15a4d84c195", size = 306203, upload-time = "2026-06-29T13:03:36.243Z" }, + { url = "https://files.pythonhosted.org/packages/d0/2b/c3eaf16f5d7c9bad66ea32f40a95bd169b29a91217fcc7f081375157e99c/jiter-0.16.0-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:d28bb3c26762358dadf3e5bf0bccd29ae987d65e6988d2e6f49829c76b003c09", size = 306489, upload-time = "2026-06-29T13:03:37.846Z" }, + { url = "https://files.pythonhosted.org/packages/96/3f/02fdfc6705cad96127d883af5c34e4867f554f29ec7705ec1a46156400a9/jiter-0.16.0-cp313-cp313-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:0542a7189c26920778658fc8fcf2af8bae05bae9924577f71804acef37996536", size = 335453, upload-time = "2026-06-29T13:03:39.221Z" }, + { url = "https://files.pythonhosted.org/packages/b2/a6/e4bda5920d4b0d7c5dfb7174ce4a6b2e4d3e11c9162c452ef0eab4cdbdbd/jiter-0.16.0-cp313-cp313-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:8fb8de1e23a0cb2a7f53c335049c7b72b6db41aa6227cdcc0972a1de5cb39450", size = 361625, upload-time = "2026-06-29T13:03:40.597Z" }, + { url = "https://files.pythonhosted.org/packages/b7/97/4e6b59b2c6e55cbb3e183595f81ad65dcfb21c915fee5e19e335df21bc55/jiter-0.16.0-cp313-cp313-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:b72d0b2990ca754a9102779ac98d8597b7cb31678958562214a007f909eab78e", size = 456958, upload-time = "2026-06-29T13:03:42.074Z" }, + { url = "https://files.pythonhosted.org/packages/15/e0/97e9557686d2f94f4b93786eccb7eed28e9228ad132ea8237f44727314a7/jiter-0.16.0-cp313-cp313-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:d5f91b1c27fc22a57993d5a5cb8a627cb8ed4b10502716fac1ffbfe1d19d84e8", size = 372017, upload-time = "2026-06-29T13:03:43.658Z" }, + { url = "https://files.pythonhosted.org/packages/0f/94/db768b6938e0df35c86beeba3dfbbb025c9ee5c19e1aa271f2396e50864d/jiter-0.16.0-cp313-cp313-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:c682bea068a90b764577bdb78a60a4c1d1606daf9cd4c893832a37c7cc9d9026", size = 343320, upload-time = "2026-06-29T13:03:45.226Z" }, + { url = "https://files.pythonhosted.org/packages/c1/d6/5a59d938244a30735fe62d9433fd325f9021ea29d89780ea4596ea93bc89/jiter-0.16.0-cp313-cp313-manylinux_2_31_riscv64.whl", hash = "sha256:8d031aabecc4f1b6276adfb42e3aabb77c89d468bf616600e8d3a11328929053", size = 350520, upload-time = "2026-06-29T13:03:46.671Z" }, + { url = "https://files.pythonhosted.org/packages/67/f8/c4a857f49c9af125f6bbcac7e3eee7f7978ed89682833062e2dbf62576b1/jiter-0.16.0-cp313-cp313-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:eab2cd170150e70153de16896a1774e3a1dca80154c56b54d7a812c479a7165e", size = 387550, upload-time = "2026-06-29T13:03:48.361Z" }, + { url = "https://files.pythonhosted.org/packages/8b/d6/5fbc2f7d6b67b754caa61a993a2e626e815dec47ffc2f9e35f01adfebec7/jiter-0.16.0-cp313-cp313-musllinux_1_1_aarch64.whl", hash = "sha256:6edb63a46e65a82c26800a868e49b2cac30dd5a4218b88d74bc2c848c8ad60bb", size = 515424, upload-time = "2026-06-29T13:03:49.881Z" }, + { url = "https://files.pythonhosted.org/packages/ed/54/284f0164b64a5fed915fea6ba7e9ba9b3d8d37c67d59cf2e3bb99d45cdfe/jiter-0.16.0-cp313-cp313-musllinux_1_1_x86_64.whl", hash = "sha256:659039cc50b5addcc35fcc87ae2c1833b7c0a8e5326ef631a75e4478447bcf84", size = 546981, upload-time = "2026-06-29T13:03:51.363Z" }, + { url = "https://files.pythonhosted.org/packages/13/c5/2a467585a576594384e1d2c43e1224deaafc085f24e243529cf98beef8e1/jiter-0.16.0-cp313-cp313-win32.whl", hash = "sha256:c9c53be232c2e206ef9cdbad81a48bfa74c3d3f08bcf8124630a8a748aad993e", size = 202853, upload-time = "2026-06-29T13:03:53.015Z" }, + { url = "https://files.pythonhosted.org/packages/88/6a/de61d04b9eec69c71719968d2f716532a3bc121170c44a39e14979c6be81/jiter-0.16.0-cp313-cp313-win_amd64.whl", hash = "sha256:baad945ed47f163ad833314f8e3288c396118934f94e7bbb9e243ce4b341a4fd", size = 196160, upload-time = "2026-06-29T13:03:54.447Z" }, + { url = "https://files.pythonhosted.org/packages/19/4b/b390ed59bafb3f31d008d1218578f10327714484b334439947f7e5b11e7f/jiter-0.16.0-cp313-cp313-win_arm64.whl", hash = "sha256:3c1fd2dbe1b0af19e987f03fe66c5f5bd105a2229c1aff4ab14890b24f41d21a", size = 189862, upload-time = "2026-06-29T13:03:55.754Z" }, + { url = "https://files.pythonhosted.org/packages/a7/89/bc4f1b57d5da938fd344a466396541e586d161320d70bffd929aaafcd8f4/jiter-0.16.0-cp314-cp314-macosx_10_12_x86_64.whl", hash = "sha256:b2c61484666ad42726029af0c00ef4541f0f3b5cdc550221f56c2343208018ee", size = 308239, upload-time = "2026-06-29T13:03:57.205Z" }, + { url = "https://files.pythonhosted.org/packages/65/7a/c415453e5213001bf3b411ff65dec3d303b0e76a4a2cfea9768cd4960994/jiter-0.16.0-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:63efadc657488f45db1c676d81e704cac2abf3fdb892def1faea61db053127e2", size = 308928, upload-time = "2026-06-29T13:03:58.643Z" }, + { url = "https://files.pythonhosted.org/packages/11/fc/1f4fb7ebf9a724c7741994f4aae18fba1e2f3133df14521a79194952c34a/jiter-0.16.0-cp314-cp314-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:cf0d73f50e7b6935677854f6e8e31d499ca7064dd24734f703e060f5b237d883", size = 336998, upload-time = "2026-06-29T13:04:00.071Z" }, + { url = "https://files.pythonhosted.org/packages/a0/8d/72cadaac05ccfa7cc3a0a2232862e6c72443ca40cf300ba8b57f9f18b69b/jiter-0.16.0-cp314-cp314-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:bf3ea07d9bc8e7d03a9fbc051295462e6dbc295b894fd72457c3136e3e43d898", size = 362112, upload-time = "2026-06-29T13:04:01.52Z" }, + { url = "https://files.pythonhosted.org/packages/58/4a/c4b0d5f651fda90a24ffce9f8d56cde462a2e09d31ae3de3c68cef34c04e/jiter-0.16.0-cp314-cp314-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:26798522707abb47d767db536e4148ceac1b14446bf028ee85e579a2e043cfe5", size = 459807, upload-time = "2026-06-29T13:04:03.214Z" }, + { url = "https://files.pythonhosted.org/packages/80/58/ef77879ea9aa56b50824edc5a445e226422c7a8d211f3fd2a56bcb9493cf/jiter-0.16.0-cp314-cp314-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:bc837c1b9631be10abfe0191537fe8009838204cec7e44827401ace390ddb567", size = 373181, upload-time = "2026-06-29T13:04:04.629Z" }, + { url = "https://files.pythonhosted.org/packages/49/2e/ffbc3f254e4d8a66da3062c624a7df4b7c2b2cf9e1fe43cf394b3e104041/jiter-0.16.0-cp314-cp314-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:49060fd70737fad59d33ba9dcc0d83247dc9e77187de26053a19c16c9f32bd69", size = 344927, upload-time = "2026-06-29T13:04:06.067Z" }, + { url = "https://files.pythonhosted.org/packages/9a/f6/0be5dc6d64a89f80aa8fec984f94dedb2973e251edcae55841d60786d578/jiter-0.16.0-cp314-cp314-manylinux_2_31_riscv64.whl", hash = "sha256:adbb8edeadd431bc4477879d5d371ece7cb1334486584e0f252656dd7ffada29", size = 352754, upload-time = "2026-06-29T13:04:07.477Z" }, + { url = "https://files.pythonhosted.org/packages/da/6e/7d31243b3b91cd261dd19e9d3557fc3251a80883d3d8049c86174e7ab7af/jiter-0.16.0-cp314-cp314-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:31aaee5b80f672c1dc21272bcfb9cbdcfc1ea04ff50f00ed5af500b80c44fa93", size = 390553, upload-time = "2026-06-29T13:04:08.92Z" }, + { url = "https://files.pythonhosted.org/packages/25/33/51ae371fde3c88897520f62b4d5f8b27ad7103e2bb10812ff52195609853/jiter-0.16.0-cp314-cp314-musllinux_1_1_aarch64.whl", hash = "sha256:6722bcef4ffc86c835574b1b2fac6b33b9fb4a889c781e67950e891591f3c55a", size = 516900, upload-time = "2026-06-29T13:04:10.407Z" }, + { url = "https://files.pythonhosted.org/packages/a0/45/6449b3d123ea439ba79507c657288f461d55049e7bcbdc2cf8eb8210f491/jiter-0.16.0-cp314-cp314-musllinux_1_1_x86_64.whl", hash = "sha256:5ab4f50ff971b611d656554ea10b75f80097392c827bc32923c6eeb6386c8b00", size = 548754, upload-time = "2026-06-29T13:04:12.046Z" }, + { url = "https://files.pythonhosted.org/packages/9b/e7/fd2fb11ae3e2649333da3aa170d04d7b3000bbdc3b270f6513382fdf4e04/jiter-0.16.0-cp314-cp314-pyemscripten_2026_0_wasm32.whl", hash = "sha256:710cc51d4ebdcd3c1f70b232c1db1ea1344a075770422bbd4bede5708335acbe", size = 122381, upload-time = "2026-06-29T13:04:13.413Z" }, + { url = "https://files.pythonhosted.org/packages/26/80/f0b147a62c315a164ed2168908286ca302310824c218d3aae52b06c0c9a9/jiter-0.16.0-cp314-cp314-win32.whl", hash = "sha256:57b37fc887a32d44798e4d8ebfa7c9683ff3da1d5bf38f08d1bb3573ccb39106", size = 204578, upload-time = "2026-06-29T13:04:14.813Z" }, + { url = "https://files.pythonhosted.org/packages/5e/e6/4758a14304b4523a6f5adb2419340086aa3593bd4327c2b25b5948a90548/jiter-0.16.0-cp314-cp314-win_amd64.whl", hash = "sha256:cbd18dd5e2df96b580487b5745adf57ef64ad89ba2d9662fc3c19386acce7db8", size = 198154, upload-time = "2026-06-29T13:04:16.272Z" }, + { url = "https://files.pythonhosted.org/packages/26/be/41fa54a2e7ea41d6c99f1dc5b1f0fd4cb474680304b5d268dd518e81da3a/jiter-0.16.0-cp314-cp314-win_arm64.whl", hash = "sha256:a32d2027a9fa67f109ff245a3252ece3ccc32cc56703e1deab6cc846a59e0585", size = 191458, upload-time = "2026-06-29T13:04:17.707Z" }, + { url = "https://files.pythonhosted.org/packages/81/6b/59127338b86d9fe4d99418f5a15118bea778103ee0fe9d9dd7e0af174e95/jiter-0.16.0-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:2577196f4474ef3fc4779a088a23b0897bbf86f9ea3679c372d45b8383b43207", size = 316739, upload-time = "2026-06-29T13:04:19.663Z" }, + { url = "https://files.pythonhosted.org/packages/2d/95/49461034d5388196d3dabf98748935f017b7785d8f3f5349f834bcc4ed0d/jiter-0.16.0-cp314-cp314t-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:616e89e008a93c01104161c75b4988e58716b01d62307ebfe161e52a56d2a818", size = 340911, upload-time = "2026-06-29T13:04:21.257Z" }, + { url = "https://files.pythonhosted.org/packages/cd/97/a4369f2fb82cb3dda13b98622f31249b2e014b223fe64ee534413ad72294/jiter-0.16.0-cp314-cp314t-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:0e2e9efbe042210df657bade597f66d6d75723e3d8f45a12ea6d8167ff8bbce3", size = 361747, upload-time = "2026-06-29T13:04:22.677Z" }, + { url = "https://files.pythonhosted.org/packages/28/51/49b6ed456261646e1906016a6760367a28aacd3c24805e4e5fe64116c1db/jiter-0.16.0-cp314-cp314t-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:3f4d9e473a5ce7d27fef8b848df4dc16e283893d3f53b4a585e72c9595f3c284", size = 460225, upload-time = "2026-06-29T13:04:24.441Z" }, + { url = "https://files.pythonhosted.org/packages/33/b5/5689aff4f66c5b60be63106e591dbfcba2190df97d2c9c7cf052361ddb98/jiter-0.16.0-cp314-cp314t-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:8d30a4a1c87713060c8d1cc59a7b6c8fb6b8ef0a6900368014c76c87922a2929", size = 373169, upload-time = "2026-06-29T13:04:25.884Z" }, + { url = "https://files.pythonhosted.org/packages/a2/96/3ae1b85ee0d6d6cab254fb7f8da018272b932bbf2d69b07e98aa2a96c746/jiter-0.16.0-cp314-cp314t-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:bae96332410f866e5900d809298b1ed82735932986c672495f9701daacd80620", size = 350332, upload-time = "2026-06-29T13:04:27.302Z" }, + { url = "https://files.pythonhosted.org/packages/15/32/c99d7bafd78986556c95bf60ce84c6cc98786eac56066c12d7f828bb6747/jiter-0.16.0-cp314-cp314t-manylinux_2_31_riscv64.whl", hash = "sha256:da3d7ec75dc83bb18bca888b5edfae0656a26849056c59e05a7728badd17e7af", size = 353377, upload-time = "2026-06-29T13:04:28.731Z" }, + { url = "https://files.pythonhosted.org/packages/0e/4b/f99a8e571287c3dec766bcc18528bbe8e8fb5365522ab5e6d64c93e87066/jiter-0.16.0-cp314-cp314t-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:ee6162b77d49a9939229df666dfa8af3e656b6701b54c4c84966d740e189264e", size = 387746, upload-time = "2026-06-29T13:04:30.319Z" }, + { url = "https://files.pythonhosted.org/packages/75/69/c78a5b3f71040e34eb5917df26fb7ae9a2174cad1ccbf277512507c53a6e/jiter-0.16.0-cp314-cp314t-musllinux_1_1_aarch64.whl", hash = "sha256:63ffdbdae7d4499f4cda14eadc12ddcabef0fc0c081191bdc2247489cb698077", size = 517292, upload-time = "2026-06-29T13:04:31.709Z" }, + { url = "https://files.pythonhosted.org/packages/c2/f7/095b38eda4c70d03651c403f29a5590f16d12ddc5d544aac9f9cddf72277/jiter-0.16.0-cp314-cp314t-musllinux_1_1_x86_64.whl", hash = "sha256:a111256a7193bea0759267b10385e5870949c239ed7b6ddbaaf57573edb38734", size = 549259, upload-time = "2026-06-29T13:04:33.721Z" }, + { url = "https://files.pythonhosted.org/packages/2e/c5/6a0207d90e5f656d95af98ebd0934f382d37674416f215aeda2ff8063e51/jiter-0.16.0-cp314-cp314t-win32.whl", hash = "sha256:de5ba8763e56b793561f43bed197c9ea55776daa5e9a6b91eed68a909bc9cdbf", size = 206523, upload-time = "2026-06-29T13:04:35.068Z" }, + { url = "https://files.pythonhosted.org/packages/a5/31/c757d5f30a8980fd945ce7b98be10be9e4ff59c7c42f5fd86804c2e87db8/jiter-0.16.0-cp314-cp314t-win_amd64.whl", hash = "sha256:b8a3f9a6008048fe9def7bf465180564a6e458047d2ce499149cfbe73c3ae9db", size = 200366, upload-time = "2026-06-29T13:04:36.61Z" }, + { url = "https://files.pythonhosted.org/packages/7c/a2/d88de6d313d734a544a7901353ad5db67cb38dcfcd91713b7979dafc345d/jiter-0.16.0-cp314-cp314t-win_arm64.whl", hash = "sha256:0fa25b09b13075c46f5bc174f2690525a925a4fc2f7c82969a2bbabff22386ce", size = 190516, upload-time = "2026-06-29T13:04:38.004Z" }, + { url = "https://files.pythonhosted.org/packages/06/d3/8e278946d43eeca2585b4dd0834a887cd71136329b837f3a16ed86a8b4b0/jiter-0.16.0-graalpy311-graalpy242_311_native-macosx_10_12_x86_64.whl", hash = "sha256:850ccb1d7eedb4200f4014b1c0e8a577de114fc3cd88faad646dcc9bc4bb12ad", size = 304518, upload-time = "2026-06-29T13:05:00.172Z" }, + { url = "https://files.pythonhosted.org/packages/72/43/28d4ef495028bf0506a413d4db3f4eb3e7288a382e0f065f306a17bbeb5e/jiter-0.16.0-graalpy311-graalpy242_311_native-macosx_11_0_arm64.whl", hash = "sha256:e34e97bda77eb63242a410243c071e28ac7e0d8c0948c5ee658498690a4b2f2f", size = 310207, upload-time = "2026-06-29T13:05:02.123Z" }, + { url = "https://files.pythonhosted.org/packages/e0/ca/c366b1012da1d640de975d9683acd44e4d150d9068845d0ca2610435253f/jiter-0.16.0-graalpy311-graalpy242_311_native-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:b7dc85ea77d4abbae8bad0d3538678aedee75bceec4e2f6c8dfb1c74772e5aa5", size = 342771, upload-time = "2026-06-29T13:05:03.55Z" }, + { url = "https://files.pythonhosted.org/packages/16/52/50cc4056fc1ae02e7154704e7ecc89df0afb8300222cfe8a52d3f67e4730/jiter-0.16.0-graalpy311-graalpy242_311_native-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:17ca7fae79f6d99cd9a042b75f917eaada7b895cfc7dd2ee3a16089dcaec7a85", size = 346468, upload-time = "2026-06-29T13:05:05.452Z" }, + { url = "https://files.pythonhosted.org/packages/98/ab/664fd8c4be028b2bedd3d2ff08769c4ede23d0dbc87a77c62384a0515b5d/jiter-0.16.0-graalpy312-graalpy250_312_native-macosx_10_12_x86_64.whl", hash = "sha256:f17d61a28b4b3e0e3e2ba98490c70501403b4d196f78732439160e7fd3678127", size = 303106, upload-time = "2026-06-29T13:05:07.118Z" }, + { url = "https://files.pythonhosted.org/packages/1a/07/421f1d5b65493a76e16027b848aba6a7d28073ae75944fa4289cc914d39f/jiter-0.16.0-graalpy312-graalpy250_312_native-macosx_11_0_arm64.whl", hash = "sha256:96e38eea538c8ddf853a35727c7be0741c76c13f04148ac5c116222f50ece3b3", size = 304658, upload-time = "2026-06-29T13:05:08.708Z" }, + { url = "https://files.pythonhosted.org/packages/0a/db/bba1155f01a01c3c37a89425d571da751bbedf5c54247b831a04cb971798/jiter-0.16.0-graalpy312-graalpy250_312_native-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:d284fb8d94d5855d60c44fefcab4bf966f1da6fada73992b01f6f0c9bc0c6702", size = 339719, upload-time = "2026-06-29T13:05:10.41Z" }, + { url = "https://files.pythonhosted.org/packages/78/f7/18a1afcd64f35314b68c1f23afcd9994d0bc13e65cc77517afff4e83986d/jiter-0.16.0-graalpy312-graalpy250_312_native-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:64d613743df53199b1aa256a7d328340da6d7078aac7705a7db9d7a791e9cfd2", size = 343885, upload-time = "2026-06-29T13:05:12.087Z" }, +] + [[package]] name = "jsonschema" version = "4.26.0" @@ -970,6 +1110,23 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/b4/07/458c344f0f0c178f4481dad5cca790626ffe4c34eabf9467069d06ee4999/numpy-2.5.2-cp315-cp315t-win_arm64.whl", hash = "sha256:5f8e00be2ec6f45f4e8a41a527f68d44a7d96fee92a650e4d8b1326f77f61e6e", size = 10748103, upload-time = "2026-08-09T13:48:24.21Z" }, ] +[[package]] +name = "openai" +version = "3.7.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "anyio" }, + { name = "httpx2" }, + { name = "jiter" }, + { name = "pydantic" }, + { name = "sniffio" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/ad/ba/6d46da4232f80cb5842280e024242e6fed163418ab81bebc1c83f693bb0b/openai-3.7.0.tar.gz", hash = "sha256:e836eb7effee89df802cd0c7d1bad8de8c993976cf238c44d5b5b844f5aefd38", size = 1455615, upload-time = "2026-09-02T01:30:54.555Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d3/9d/77e0f43f04ae51a3d21bb08bd372a752856b86ee22648b1458c93a623927/openai-3.7.0-py3-none-any.whl", hash = "sha256:008fa33e0a01dc71039c27355ef8f476eed690e8e48bec18183878dcd32fb841", size = 1699594, upload-time = "2026-09-02T01:30:52.445Z" }, +] + [[package]] name = "opentelemetry-api" version = "1.44.0" @@ -1625,6 +1782,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/10/85/48f0abdcef5cce4e034c7a5b0ceeceba0b01bf0d942824f4bb720afe2dec/rpds_py-2026.6.3-pp311-pypy311_pp73-musllinux_1_2_x86_64.whl", hash = "sha256:8e65860d238379ed982fd9ba690579b5e95af2f4840f99c772816dbe573cb826", size = 586486, upload-time = "2026-06-30T07:17:51.141Z" }, ] +[[package]] +name = "sniffio" +version = "1.3.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/a2/87/a6771e1546d97e7e041b6ae58d80074f81b7d5121207425c964ddf5cfdbd/sniffio-1.3.1.tar.gz", hash = "sha256:f4324edc670a0f49750a81b895f35c3adb843cca46f0530f79fc1babb23789dc", size = 20372, upload-time = "2024-02-25T23:20:04.057Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e9/44/75a9c9421471a6c4805dbf2356f7c181a29c1879239abab1ea2cc8f38b40/sniffio-1.3.1-py3-none-any.whl", hash = "sha256:2f6da418d1f1e0fddd844478f41680e794e6051915791a034ff65e5f100525a2", size = 10235, upload-time = "2024-02-25T23:20:01.196Z" }, +] + [[package]] name = "sortedcontainers" version = "2.4.0" @@ -1750,6 +1916,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/7b/61/cceae43728b7de99d9b847560c262873a1f6c98202171fd5ed62640b494b/tomli-2.4.1-py3-none-any.whl", hash = "sha256:0d85819802132122da43cb86656f8d1f8c6587d54ae7dcaf30e90533028b49fe", size = 14583, upload-time = "2026-03-25T20:22:03.012Z" }, ] +[[package]] +name = "truststore" +version = "0.10.4" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/53/a3/1585216310e344e8102c22482f6060c7a6ea0322b63e026372e6dcefcfd6/truststore-0.10.4.tar.gz", hash = "sha256:9d91bd436463ad5e4ee4aba766628dd6cd7010cf3e2461756b3303710eebc301", size = 26169, upload-time = "2025-08-12T18:49:02.73Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/19/97/56608b2249fe206a67cd573bc93cd9896e1efb9e98bce9c163bcdc704b88/truststore-0.10.4-py3-none-any.whl", hash = "sha256:adaeaecf1cbb5f4de3b1959b42d41f6fab57b2b1666adb59e89cb0b53361d981", size = 18660, upload-time = "2025-08-12T18:49:01.46Z" }, +] + [[package]] name = "typing-extensions" version = "4.16.0" From 14f9f0ed3b6e12f9a5d0a536b2ec289646fe7116 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Fri, 4 Sep 2026 17:14:42 +0900 Subject: [PATCH 2/2] fix(api): repair OpenAI batch and transcription contracts --- contextual_orchestrator/batch_routing.py | 45 ++++- contextual_orchestrator/cost_router.py | 31 +++- contextual_orchestrator/file_registry.py | 12 ++ contextual_orchestrator/server.py | 200 ++++++++++++++++------ docs/library_research.md | 1 + fuzz/targets.py | 15 ++ tests/test_openai_batch_review_repairs.py | 83 +++++++++ tests/test_openai_sdk_compat.py | 184 +++++++++++++++++++- 8 files changed, 506 insertions(+), 65 deletions(-) create mode 100644 tests/test_openai_batch_review_repairs.py diff --git a/contextual_orchestrator/batch_routing.py b/contextual_orchestrator/batch_routing.py index 5650e441d..9e5c1ddbc 100644 --- a/contextual_orchestrator/batch_routing.py +++ b/contextual_orchestrator/batch_routing.py @@ -190,6 +190,7 @@ class BatchRequest: attribution: Dict[str, Any] = field(default_factory=dict) mode: str = "auto" zdr_only: bool = False + parameters: Dict[str, Any] = field(default_factory=dict) def to_jsonl_line(self, endpoint: str = "/v1/chat/completions") -> Dict[str, Any]: """Render this request as an OpenAI Batch API JSONL line.""" @@ -199,7 +200,11 @@ def to_jsonl_line(self, endpoint: str = "/v1/chat/completions") -> Dict[str, Any "url": endpoint, # ``zdr_only`` is a contextual-orchestrator selection policy, not # an upstream provider request field. - "body": {"model": self.model, "messages": self.messages}, + "body": { + **self.parameters, + "model": self.model, + "messages": self.messages, + }, } @@ -324,13 +329,17 @@ def retrieve(self, job: BatchJob) -> List[BatchResultItem]: """Retrieve completed results for a batch job.""" ... + def cancel(self, job: BatchJob) -> Dict[str, Any]: + """Request backend cancellation without inventing a terminal state.""" + ... + class LocalBatchBackend: """In-process batch backend that runs each request via an injected runner. Preserves the mock/local path: no external service, no Postgres. The runner - is any callable ``(messages, mode, model) -> {"answer": str, "mode": str}`` — the - orchestrator's own ``complete`` fits directly. Results are computed eagerly + is any callable ``(messages, mode, model[, parameters]) -> result``; the fourth + argument is used only when an SDK batch line carries provider options. Results are computed eagerly on submit and returned verbatim on retrieve, so the batch lifecycle is fully observable in tests. """ @@ -368,7 +377,12 @@ def run(request: BatchRequest) -> BatchResultItem: else nullcontext() ) with context: - result = self._runner(request.messages, request.mode, request.model) + if request.parameters: + result = self._runner( + request.messages, request.mode, request.model, request.parameters + ) + else: + result = self._runner(request.messages, request.mode, request.model) answer = result.get("answer", "") # The runner (typically orchestrator.complete()) has no top-level # "usage" key -- real provider usage is nested per-step inside @@ -430,6 +444,10 @@ def retrieve(self, job: BatchJob) -> List[BatchResultItem]: """Return the results computed at submit time.""" return self._results.get(job.job_id, []) + def cancel(self, job: BatchJob) -> Dict[str, Any]: + """Report the already-terminal local job; no cancellation is fabricated.""" + return {"accepted": False, **self.poll(job)} + class PgLlmBatchBackend: """Batch backend that submits to **pg-llm-batch** and retrieves results. @@ -581,6 +599,25 @@ async def _download() -> Dict[str, Any]: ) return items + def cancel(self, job: BatchJob) -> Dict[str, Any]: + """Ask pg-llm-batch to cancel and return its authoritative status.""" + cancel_batch_job = getattr(self._client, "cancel_batch_job", None) + if not callable(cancel_batch_job): + return {"accepted": False, **self.poll(job)} + + async def _cancel() -> Dict[str, Any]: + return await cancel_batch_job(job.job_id, self._endpoint_alias) + + result = self._run(_cancel()) + status = result.get("status") + accepted = result.get("accepted") is True or status in {"cancelling", "cancelled"} + return { + "accepted": accepted, + "job_id": job.job_id, + "status": status, + "is_complete": status in {"completed", "failed", "expired", "cancelled"}, + } + def _extract_answer(body: Dict[str, Any]) -> str: choices = body.get("choices") or [] diff --git a/contextual_orchestrator/cost_router.py b/contextual_orchestrator/cost_router.py index 751c9e366..10acb606a 100644 --- a/contextual_orchestrator/cost_router.py +++ b/contextual_orchestrator/cost_router.py @@ -176,7 +176,11 @@ def __init__( start_embedding_job(recovered_job) def _run_local_batch( - self, messages: List[Dict[str, str]], mode: str, model: str + self, + messages: List[Dict[str, str]], + mode: str, + model: str, + parameters: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """Run one local item while retaining completed endpoint-race usage.""" context = { @@ -186,7 +190,21 @@ def _run_local_batch( } token = self._race_usage_context.set(context) try: - result = self.orchestrator.complete(messages, mode=mode, model_name=model) + if parameters: + proxied = self.orchestrator.proxy_completion( + {**parameters, "model": model, "messages": messages}, + endpoint="chat/completions", + single_agent=True, + ) + choices = proxied.get("choices") or [] + message = choices[0].get("message", {}) if choices else {} + result = { + "answer": message.get("content", ""), + "mode": "route", + "trace": [{"usage": proxied.get("usage", {})}], + } + else: + result = self.orchestrator.complete(messages, mode=mode, model_name=model) finally: self._race_usage_context.reset(token) race_usage = [] @@ -1030,6 +1048,11 @@ def poll_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict[str job = self._require_job(job_id, owner_id=owner_id) return self.batch_backend.poll(job) + def cancel_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict[str, Any]: + """Request cancellation from the backend that owns the batch.""" + job = self._require_job(job_id, owner_id=owner_id) + return self.batch_backend.cancel(job) + def retrieve_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict[str, Any]: """Retrieve results for a batch owned by ``owner_id`` and record usage. @@ -1241,7 +1264,9 @@ def _batch_usage_record_id(job_id: str, custom_id: str, kind: str, index: int) - def _batch_item_usage_valid(item: BatchResultItem) -> bool: """True when a batch result item's provider-reported usage is trustworthy.""" return ( - item.prompt_tokens >= 0 + type(item.prompt_tokens) is int + and item.prompt_tokens >= 0 + and type(item.completion_tokens) is int and item.completion_tokens >= 0 and ( item.usage_valid is True diff --git a/contextual_orchestrator/file_registry.py b/contextual_orchestrator/file_registry.py index d8944c5a5..0835b4fa8 100644 --- a/contextual_orchestrator/file_registry.py +++ b/contextual_orchestrator/file_registry.py @@ -59,6 +59,16 @@ def __init__(self, job_registry: Any) -> None: # base64 text (JSON-mapping values must be JSON-serializable), keyed # by gateway file id. Only populated for locally registered files. self._content = job_registry.mapping("local_file_content") + self._retention_seconds = job_registry.retention_seconds + + def _purge_expired_local_content(self) -> None: + """Remove orphaned or expired gateway-generated file bodies.""" + cutoff = int(time.time()) - self._retention_seconds + for gateway_file_id in list(self._content): + owner = self._owners.get(gateway_file_id) + created_at = owner.document.get("created_at") if owner is not None else None + if owner is None or type(created_at) is not int or created_at < cutoff: + self._content.pop(gateway_file_id, None) def register( self, @@ -126,6 +136,7 @@ def register_local( ``_LOCAL_FILE_AGENT_ID`` sentinel and the bytes are served straight back out of ``self._content`` (see ``is_local``/``local_content``). """ + self._purge_expired_local_content() if not owner_id: raise FileContractError("file owner is unavailable") gateway_file_id = f"file_{uuid.uuid4().hex}" @@ -158,6 +169,7 @@ def is_local(owner: FileOwner) -> bool: def local_content(self, gateway_file_id: str) -> bytes: """Return the raw bytes registered for a locally generated file.""" + self._purge_expired_local_content() return base64.b64decode(self._content[gateway_file_id]) @staticmethod diff --git a/contextual_orchestrator/server.py b/contextual_orchestrator/server.py index 04a23b53b..36f6bb355 100644 --- a/contextual_orchestrator/server.py +++ b/contextual_orchestrator/server.py @@ -12,6 +12,7 @@ import ipaddress import json import logging +import math import mmap import secrets import socket @@ -1565,8 +1566,12 @@ def _validate_completions_temperature(body: dict[str, Any]) -> float | None: if temperature is None: return None value = float(temperature) - if value < 0 or value > 2: - raise RequestError(400, "invalid_temperature", "temperature must be a number in [0, 2]") + if not math.isfinite(value) or value < 0 or value > 2: + raise RequestError( + 400, + "invalid_temperature", + "temperature must be a finite number in [0, 2]", + ) body["temperature"] = value return value @@ -2415,6 +2420,17 @@ def _validate_capability_request(path: str, body: dict[str, Any]) -> None: isinstance(audio.get(field), str) and audio[field] for field in ("data", "format") ): raise RequestError(400, "invalid_input_audio", "input_audio.data and input_audio.format are required") + response_format = body.get("response_format") + if response_format is not None and response_format not in { + "diarized_json", "json", "text", "srt", "verbose_json", "vtt" + }: + raise RequestError( + 400, + "invalid_response_format", + "response_format must be diarized_json, json, text, srt, verbose_json, or vtt", + ) + if "temperature" in body: + _validate_completions_temperature(body) if path == "/v1/rerank": documents = body.get("documents") if not isinstance(documents, list) or not documents: @@ -3372,7 +3388,9 @@ def _validate_batch_requests( return batch -def _parse_batch_input_jsonl(raw: bytes, *, zdr_only: bool) -> list[BatchRequest]: +def _parse_batch_input_jsonl( + raw: bytes, *, endpoint: str, zdr_only: bool +) -> list[BatchRequest]: """Parse an OpenAI Batch API input file (JSONL) into internal batch requests. Each line must be shaped like a real Batch input line: ``{"custom_id", @@ -3414,6 +3432,10 @@ def _parse_batch_input_jsonl(raw: bytes, *, zdr_only: bool) -> list[BatchRequest seen_custom_ids.add(custom_id) if entry.get("method") != "POST": raise RequestError(400, "invalid_file", f"line {line_number} method must be POST") + if entry.get("url") != endpoint: + raise RequestError( + 400, "invalid_file", f"line {line_number} url must match the batch endpoint" + ) body_obj = entry.get("body") if not isinstance(body_obj, dict): raise RequestError(400, "invalid_file", f"line {line_number} body must be an object") @@ -3421,6 +3443,16 @@ def _parse_batch_input_jsonl(raw: bytes, *, zdr_only: bool) -> list[BatchRequest model = body_obj.get("model", TaskOrchestrator.GATEWAY_DEFAULT_MODEL) if not isinstance(model, str) or not model.strip(): raise RequestError(400, "invalid_file", f"line {line_number} model must be a non-empty string") + _reject_unknown_keys(body_obj, {"model", "messages"} | OPENAI_PASSTHROUGH_PARAM_KEYS) + if body_obj.get("stream") not in (None, False): + raise RequestError(400, "invalid_file", f"line {line_number} stream is not supported in batches") + _validate_chat_sampling_and_control_fields(body_obj, stream=False) + if "response_format" in body_obj: + _validate_chat_response_format(body_obj) + if "tools" in body_obj: + _validate_chat_tools(body_obj) + if "tool_choice" in body_obj: + _validate_chat_tool_choice(body_obj) requests.append( BatchRequest( messages=messages, @@ -3429,6 +3461,11 @@ def _parse_batch_input_jsonl(raw: bytes, *, zdr_only: bool) -> list[BatchRequest attribution={}, mode="auto", zdr_only=zdr_only, + parameters={ + key: value + for key, value in body_obj.items() + if key not in {"model", "messages"} + }, ) ) return requests @@ -3512,8 +3549,8 @@ def _batch_output_jsonl_line(item: dict[str, Any], *, model: str) -> str: ``CostRoutingCoordinator.retrieve_batch`` -- this only reshapes them into the OpenAI Batch output-file line format, it invents nothing. """ - prompt_tokens = item.get("prompt_tokens") or 0 - completion_tokens = item.get("completion_tokens") or 0 + prompt_tokens = item.get("prompt_tokens") + completion_tokens = item.get("completion_tokens") body = { "id": _new_chat_completion_id(), "object": "chat.completion", @@ -3526,11 +3563,15 @@ def _batch_output_jsonl_line(item: dict[str, Any], *, model: str) -> str: "finish_reason": "stop", } ], - "usage": { - "prompt_tokens": prompt_tokens, - "completion_tokens": completion_tokens, - "total_tokens": prompt_tokens + completion_tokens, - }, + "usage": ( + { + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, + "total_tokens": prompt_tokens + completion_tokens, + } + if type(prompt_tokens) is int and type(completion_tokens) is int + else None + ), } return json.dumps( { @@ -6062,7 +6103,7 @@ def do_GET(self) -> None: # noqa: N802 items = sorted( ( (job_id, tracked) - for job_id, tracked in openai_batches.items() + for job_id, tracked in list(openai_batches.items()) if tracked.get("owner_id") == principal_id ), key=lambda pair: pair[1].get("created_at", 0), @@ -6080,7 +6121,10 @@ def do_GET(self) -> None: # noqa: N802 ) page = items[:limit] data = [ - self._batch_document(job_id, principal_id) for job_id, _tracked in page + _openai_batch_object( + job_id, tracked, tracked.get("status", "validating") + ) + for job_id, tracked in page ] self._send( { @@ -7026,7 +7070,24 @@ def register_video_job(agent: ModelAgent, provider_result: dict[str, Any]) -> di "capability_unavailable", f"no enabled {capability}-capable model group member is available", ) from exc - if binary: + transcription_format = body.get("response_format") + if capability == "transcription" and transcription_format in { + "text", "srt", "vtt" + }: + text_result = result if isinstance(result, str) else result.get("text") + if not isinstance(text_result, str): + raise RequestError( + 502, + "invalid_transcription_response", + "transcription provider did not return text", + ) + content_type = { + "text": "text/plain; charset=utf-8", + "srt": "application/x-subrip; charset=utf-8", + "vtt": "text/vtt; charset=utf-8", + }[transcription_format] + self._send_bytes(text_result.encode("utf-8"), content_type) + elif binary: raw, content_type = result self._send_bytes(raw, content_type) else: @@ -7894,14 +7955,22 @@ def register_video_job(agent: ModelAgent, provider_result: dict[str, Any]) -> di f"endpoint must be one of {sorted(BATCH_CREATE_ENDPOINTS)}", ) completion_window = body.get("completion_window", "24h") - if not isinstance(completion_window, str) or not completion_window.strip(): + if completion_window != "24h": raise RequestError( 400, "invalid_completion_window", - "completion_window must be a non-empty string", + "completion_window must be 24h", ) metadata = _validate_openai_metadata(body) principal_id = security.principal_id(self.headers) + try: + input_owner = files.owner(input_file_id, principal_id) + except KeyError: + raise RequestError(404, "file_not_found", "file was not found") from None + if input_owner.document.get("purpose") != "batch": + raise RequestError( + 400, "invalid_file", "input_file_id must refer to a batch-purpose file" + ) agent, provider_file_id = _resolve_file_download_agent( orchestrator, files, input_file_id, principal_id ) @@ -7921,7 +7990,9 @@ def register_video_job(agent: ModelAgent, provider_result: dict[str, Any]) -> di raise RequestError( 503, "file_provider_unavailable", "the file provider is unavailable" ) from exc - batch_requests = _parse_batch_input_jsonl(raw, zdr_only=zdr_only) + batch_requests = _parse_batch_input_jsonl( + raw, endpoint=endpoint, zdr_only=zdr_only + ) models_by_custom_id = { request.custom_id: request.model for request in batch_requests } @@ -7945,6 +8016,7 @@ def register_video_job(agent: ModelAgent, provider_result: dict[str, Any]) -> di "request_count": job.request_count, "output_file_id": None, "models_by_custom_id": models_by_custom_id, + "status": job.status, } orchestrator.record_analytics_event( "openai_batch_created", @@ -7967,19 +8039,24 @@ def register_video_job(agent: ModelAgent, provider_result: dict[str, Any]) -> di if tracked is None or tracked.get("owner_id") != principal_id: self._send_error(404, "batch_not_found", f"batch {batch_id} not found") return - poll_result = self._run( - lambda: coordinator.poll_batch(batch_id, owner_id=principal_id) - ) - if not poll_result.get("is_complete") and tracked.get("cancelling_at") is None: - # ponytail: neither LocalBatchBackend nor PgLlmBatchBackend - # expose a cancel primitive for chat batches (only the - # provider embeddings backend does) -- record the - # cancellation request locally and surface "cancelling" - # rather than fabricate provider-side cancellation. - # Upgrade path: wire BatchBackend.cancel() through - # pg-llm-batch once that primitive exists. - tracked["cancelling_at"] = int(time.time()) - openai_batches[batch_id] = tracked + with coordinator.job_registry.lock( + "openai_batch", batch_id, lease_seconds=30 + ): + cancellation = self._run( + lambda: coordinator.cancel_batch( + batch_id, owner_id=principal_id + ) + ) + tracked = openai_batches[batch_id] + if cancellation.get("accepted"): + status = cancellation.get("status") + tracked["status"] = ( + status if status in {"cancelling", "cancelled"} else "cancelling" + ) + tracked["cancelling_at"] = int(time.time()) + if tracked["status"] == "cancelled": + tracked["cancelled_at"] = int(time.time()) + openai_batches[batch_id] = tracked self._send(self._batch_document(batch_id, principal_id)) return if path == "/v1/responses": @@ -8666,32 +8743,38 @@ def _batch_document(self, job_id: str, principal_id: str) -> dict[str, Any]: and download-failure state recorded on ``tracked`` win over a raw poll status. """ - tracked = openai_batches[job_id] - poll_result = self._run( - lambda: coordinator.poll_batch(job_id, owner_id=principal_id) - ) - status = _co_batch_status_to_openai( - poll_result.get("status"), poll_result.get("is_complete") - ) - if tracked.get("cancelling_at") is not None: - if poll_result.get("is_complete"): - status = "cancelled" - tracked.setdefault("cancelled_at", int(time.time())) - else: - status = "cancelling" - elif tracked.get("failed_at") is not None: - status = "failed" - elif status == "completed" and not tracked.get("output_file_id"): - try: - output_file_id = self._materialize_batch_output(job_id, tracked, principal_id) - except BatchDownloadError: - tracked["failed_at"] = int(time.time()) - status = "failed" - else: - tracked["output_file_id"] = output_file_id - tracked["completed_at"] = int(time.time()) - openai_batches[job_id] = tracked - return _openai_batch_object(job_id, tracked, status) + with coordinator.job_registry.lock( + "openai_batch", + job_id, + lease_seconds=30, + renew_until_epoch=time.time() + 1800, + ): + tracked = openai_batches[job_id] + poll_result = self._run( + lambda: coordinator.poll_batch(job_id, owner_id=principal_id) + ) + status = _co_batch_status_to_openai( + poll_result.get("status"), poll_result.get("is_complete") + ) + if tracked.get("cancelling_at") is not None and status not in { + "completed", "failed", "expired" + }: + status = "cancelled" if status == "cancelled" else "cancelling" + if status == "cancelled": + tracked.setdefault("cancelled_at", int(time.time())) + if status == "completed" and not tracked.get("output_file_id"): + try: + output_file_id = self._materialize_batch_output( + job_id, tracked, principal_id + ) + except BatchDownloadError: + pass + else: + tracked["output_file_id"] = output_file_id + tracked["completed_at"] = int(time.time()) + tracked["status"] = status + openai_batches[job_id] = tracked + return _openai_batch_object(job_id, tracked, status) def _parse_positive_int(self, raw: str | None, field_name: str, default: int, max_value: int | None = None) -> int: value = default if raw is None else int(raw) @@ -8796,11 +8879,16 @@ def _read_multipart_transcription_body(self) -> dict[str, Any]: temperature = fields.get("temperature") if isinstance(temperature, str) and temperature: try: - request_body["temperature"] = float(temperature) + parsed_temperature = float(temperature) except ValueError: raise RequestError( 400, "invalid_temperature", "temperature must be a number" ) from None + if not math.isfinite(parsed_temperature): + raise RequestError( + 400, "invalid_temperature", "temperature must be a finite number" + ) + request_body["temperature"] = parsed_temperature return request_body def log_message(self, format: str, *args: object) -> None: diff --git a/docs/library_research.md b/docs/library_research.md index 3cee44389..1dd7c1c82 100644 --- a/docs/library_research.md +++ b/docs/library_research.md @@ -24,6 +24,7 @@ primitives use maintained libraries when the enterprise target requires them. | Provider-embedding claim ownership | Existing `redis-py` lock token plus one Valkey Lua transaction | Propagate renewal loss, compare the live execution token, and atomically write terminal state with result/usage/error. A live worker retries claim acquisition until terminal state or deadline; provider-side exactly-once execution is not claimed. | Redis's official distributed-lock guidance requires ownership-safe release/extension and recommends fencing when correctness depends on exclusive work. Reused the existing registry and skipped a new coordination dependency, forced cancellation of synchronous provider I/O, and an unsupported provider-idempotency claim. | | Provider-embedding token accounting | Existing PyO3 + `tiktoken-rs` extension, with configured `pg_tiktoken` first | Load the packaged Rust extension in the production embedding path for the exact OpenAI-published cl100k embedding model IDs. Missing/failing native code and unknown tokenizers are explicitly unavailable; splitting, provider dispatch, usage, and cost fail closed instead of estimating. | OpenAI's public encoding table maps `text-embedding-ada-002`, `text-embedding-3-small`, and `text-embedding-3-large` to cl100k; PyO3 publishes the existing module in-package. Skipped tokenizer-name inference, a second tokenizer implementation, a provider SDK, and heuristic fallback. ADR 0006 now governs chat accounting separately. | | Chat token accounting | Existing provider usage fields plus the packaged PyO3 + `tiktoken-rs` extension | Treat valid provider usage as authoritative. Use Rust only for raw textual output from exact model IDs declared by ADR 0006; prompt framing, tools, multimodal input, unknown models, missing native code, and missing stream usage are explicitly unavailable. Enabled budgets fail closed and token-threshold routing remains synchronous when the required count is unavailable. | OpenAI's Chat Completions contract carries provider usage and notes streamed usage can be absent; OpenAI's public tiktoken model table separates exact mappings from unsafe prefix matching. Reused the existing extension and storage status seam. Skipped a provider SDK, prompt-serialization reimplementation, prefix/name inference, heuristic estimates, and fabricated zero-cost reporting. | +| OpenAI SDK batch and transcription compatibility | Existing stdlib HTTP server, `BatchRequest`/`BatchBackend`, `JobRegistryFactory`, and test-only `openai>=2.0` client | Keep production dependency-free: preserve each validated Chat Completions option in the existing batch request, require the SDK's `purpose=batch`, `/v1/chat/completions`, and `completion_window=24h` contracts, and reuse the durable per-job lock for output publication. Text, SRT, and VTT transcription formats use bounded text responses; unknown usage remains null. | OpenAI Python v2.11 documents `batches.create(input_file_id, endpoint, completion_window, metadata)`, `files.create(..., purpose=...)`, cancellation, and transcription `response_format`; the existing batch papers in `docs/papers/README.md` ground throughput. No runtime SDK, parser package, lock service, or duplicate batch engine was added. | ## Ponytail Decision diff --git a/fuzz/targets.py b/fuzz/targets.py index 415dad9f9..1585b5734 100644 --- a/fuzz/targets.py +++ b/fuzz/targets.py @@ -136,6 +136,21 @@ def exercise_request_body(raw: bytes) -> None: Mirrors what ``Handler._read_json`` does after size/content-type checks: decode + JSON-parse, then run the field validators the POST routes use. """ + try: + server._parse_batch_input_jsonl( + raw, endpoint="/v1/chat/completions", zdr_only=False + ) + except RequestError: + pass + try: + fields = server._read_multipart_form( + raw, "multipart/form-data; boundary=fuzz-boundary" + ) + except RequestError: + pass + else: + assert isinstance(fields, dict) + try: body = server._coerce_json(raw) except _EXPECTED_BODY_EXC: diff --git a/tests/test_openai_batch_review_repairs.py b/tests/test_openai_batch_review_repairs.py new file mode 100644 index 000000000..8259cdd99 --- /dev/null +++ b/tests/test_openai_batch_review_repairs.py @@ -0,0 +1,83 @@ +"""Regression tests for the OpenAI Batch and transcription review repairs.""" + +from __future__ import annotations + +import json +import math + +import pytest + +from contextual_orchestrator.batch_routing import BatchRequest, LocalBatchBackend +from contextual_orchestrator.server import RequestError, _batch_output_jsonl_line, _parse_batch_input_jsonl + + +def _line(body: dict, *, url: str = "/v1/chat/completions") -> bytes: + return (json.dumps({"custom_id": "item", "method": "POST", "url": url, "body": body}) + "\n").encode() + + +def test_batch_parser_requires_matching_endpoint_and_preserves_options() -> None: + body = { + "model": "mock-model", + "messages": [{"role": "user", "content": "hello"}], + "temperature": 0.25, + "max_completion_tokens": 32, + "response_format": {"type": "json_object"}, + } + request = _parse_batch_input_jsonl( + _line(body), endpoint="/v1/chat/completions", zdr_only=False + )[0] + assert request.parameters == { + "temperature": 0.25, + "max_completion_tokens": 32, + "response_format": {"type": "json_object"}, + } + assert request.to_jsonl_line()["body"] == body + + with pytest.raises(RequestError, match="url must match"): + _parse_batch_input_jsonl( + _line(body, url="/v1/embeddings"), + endpoint="/v1/chat/completions", + zdr_only=False, + ) + + +def test_local_batch_applies_preserved_options() -> None: + captured: dict = {} + + def runner(messages, mode, model, parameters): + captured.update(parameters) + return {"answer": "ok", "mode": mode} + + backend = LocalBatchBackend(runner) + job = backend.submit([ + BatchRequest( + messages=[{"role": "user", "content": "hello"}], + parameters={"temperature": 0.2}, + ) + ]) + assert backend.retrieve(job)[0].answer == "ok" + assert captured == {"temperature": 0.2} + + +def test_unknown_batch_usage_stays_unknown() -> None: + line = json.loads( + _batch_output_jsonl_line( + {"custom_id": "item", "answer": "ok", "prompt_tokens": None, "completion_tokens": None}, + model="mock-model", + ) + ) + assert line["response"]["body"]["usage"] is None + + +@pytest.mark.parametrize("value", [math.nan, math.inf, -math.inf]) +def test_batch_request_rejects_non_finite_temperature(value: float) -> None: + with pytest.raises(RequestError, match="finite"): + _parse_batch_input_jsonl( + _line({ + "model": "mock-model", + "messages": [{"role": "user", "content": "hello"}], + "temperature": value, + }), + endpoint="/v1/chat/completions", + zdr_only=False, + ) diff --git a/tests/test_openai_sdk_compat.py b/tests/test_openai_sdk_compat.py index 39cad1f21..6f4d9b69a 100644 --- a/tests/test_openai_sdk_compat.py +++ b/tests/test_openai_sdk_compat.py @@ -22,11 +22,18 @@ import io import json import threading +import time import openai import pytest from contextual_orchestrator import ModelAgent, TaskOrchestrator +from contextual_orchestrator.batch_routing import ( + BatchDownloadError, + BatchJob, + BatchResultItem, +) +from contextual_orchestrator.cost_router import CostRoutingCoordinator from contextual_orchestrator.server import SecurityConfig, _read_multipart_form, build_server _TOKEN = "openai_sdk_compat_token" # noqa: S105 @@ -68,8 +75,13 @@ def proxy_get_bytes(self, agent, endpoint, *, max_response_bytes): return self._store[provider_id], "application/jsonl" -def _start(orchestrator: TaskOrchestrator): - server = build_server(orchestrator, port=0, security=SecurityConfig(auth_token=_TOKEN)) +def _start(orchestrator: TaskOrchestrator, *, coordinator=None): + server = build_server( + orchestrator, + port=0, + security=SecurityConfig(auth_token=_TOKEN), + coordinator=coordinator, + ) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() return server, thread @@ -99,6 +111,48 @@ def _batch_input_jsonl(custom_ids: tuple[str, ...], *, model: str) -> bytes: return ("\n".join(lines) + "\n").encode("utf-8") +class _ControllableBatchBackend: + """Small backend double for cancellation and retrieval races.""" + + name = "controlled" + + def __init__(self, *, status: str = "in_progress", transient_download: bool = False): + self.status = status + self.transient_download = transient_download + self.retrieve_count = 0 + + def submit(self, requests, metadata=None): + self.requests = requests + return BatchJob( + job_id="controlled_batch", + backend=self.name, + status=self.status, + request_count=len(requests), + ) + + def poll(self, job): + return { + "job_id": job.job_id, + "status": self.status, + "is_complete": self.status in {"completed", "cancelled"}, + } + + def retrieve(self, job): + self.retrieve_count += 1 + if self.transient_download: + self.transient_download = False + raise BatchDownloadError(job.job_id, "temporary") + time.sleep(0.02) + return [ + BatchResultItem(custom_id=request.custom_id, answer="done") + for request in self.requests + ] + + def cancel(self, job): + self.status = "cancelling" + return {"accepted": True, "status": self.status, "is_complete": False} + + def test_sdk_batches_create_retrieve_list_cancel_round_trip() -> None: """client.batches.create/retrieve/list/cancel() against a real HTTP gateway. @@ -169,6 +223,79 @@ def test_sdk_batches_create_retrieve_list_cancel_round_trip() -> None: thread.join(timeout=5) +def test_sdk_batch_cancel_reports_only_backend_accepted_state() -> None: + """Cancellation state comes from the backend, not a local timestamp.""" + agent = ModelAgent("batch_worker", "mock-batch-model", tags=("files",)) + orchestrator = TaskOrchestrator([agent]) + transport = _FakeFileTransport() + orchestrator.client.proxy_upload = transport.proxy_upload # type: ignore[method-assign] + orchestrator.client.proxy_get_bytes = transport.proxy_get_bytes # type: ignore[method-assign] + backend = _ControllableBatchBackend() + coordinator = CostRoutingCoordinator(orchestrator, batch_backend=backend) + server, thread = _start(orchestrator, coordinator=coordinator) + try: + client = _client(server.server_address[1]) + uploaded = client.files.create( + file=("input.jsonl", io.BytesIO(_batch_input_jsonl(("a",), model="mock-batch-model"))), + purpose="batch", + ) + batch = client.batches.create( + input_file_id=uploaded.id, + endpoint="/v1/chat/completions", + completion_window="24h", + ) + assert batch.status == "in_progress" + assert client.batches.cancel(batch.id).status == "cancelling" + finally: + server.shutdown() + thread.join(timeout=5) + + +def test_sdk_batch_transient_download_retries_with_single_output_writer() -> None: + """A transient download remains retryable and concurrent readers publish once.""" + agent = ModelAgent("batch_worker", "mock-batch-model", tags=("files",)) + orchestrator = TaskOrchestrator([agent]) + transport = _FakeFileTransport() + orchestrator.client.proxy_upload = transport.proxy_upload # type: ignore[method-assign] + orchestrator.client.proxy_get_bytes = transport.proxy_get_bytes # type: ignore[method-assign] + backend = _ControllableBatchBackend(status="completed", transient_download=True) + coordinator = CostRoutingCoordinator(orchestrator, batch_backend=backend) + server, thread = _start(orchestrator, coordinator=coordinator) + try: + client = _client(server.server_address[1]) + uploaded = client.files.create( + file=("input.jsonl", io.BytesIO(_batch_input_jsonl(("a",), model="mock-batch-model"))), + purpose="batch", + ) + batch = client.batches.create( + input_file_id=uploaded.id, + endpoint="/v1/chat/completions", + completion_window="24h", + ) + assert batch.status == "completed" + assert batch.output_file_id is None + + output_ids: list[str | None] = [] + + def retrieve() -> None: + output_ids.append( + _client(server.server_address[1]).batches.retrieve(batch.id).output_file_id + ) + + readers = [threading.Thread(target=retrieve) for _ in range(2)] + for reader in readers: + reader.start() + for reader in readers: + reader.join(timeout=5) + assert len(output_ids) == 2 + assert output_ids[0] == output_ids[1] + assert output_ids[0] is not None + assert backend.retrieve_count == 2 + finally: + server.shutdown() + thread.join(timeout=5) + + def test_sdk_batches_create_rejects_unsupported_endpoint() -> None: """A non-chat batch endpoint fails closed rather than being mishandled.""" server, thread = _start(TaskOrchestrator([ModelAgent("worker_agent", "mock-model", tags=("files",))])) @@ -196,6 +323,38 @@ def test_sdk_batches_create_rejects_unsupported_endpoint() -> None: thread.join(timeout=5) +@pytest.mark.parametrize( + ("purpose", "completion_window"), + [("user_data", "24h"), ("batch", "48h")], +) +def test_sdk_batches_enforces_file_purpose_and_completion_window( + purpose: str, completion_window: str +) -> None: + """Batch creation rejects promises the backend cannot honor.""" + orchestrator = TaskOrchestrator([ + ModelAgent("worker_agent", "mock-model", tags=("files",)) + ]) + transport = _FakeFileTransport() + orchestrator.client.proxy_upload = transport.proxy_upload # type: ignore[method-assign] + orchestrator.client.proxy_get_bytes = transport.proxy_get_bytes # type: ignore[method-assign] + server, thread = _start(orchestrator) + try: + client = _client(server.server_address[1]) + uploaded = client.files.create( + file=("input.jsonl", io.BytesIO(_batch_input_jsonl(("a",), model="mock-model"))), + purpose=purpose, + ) + with pytest.raises(openai.BadRequestError): + client.batches.create( + input_file_id=uploaded.id, + endpoint="/v1/chat/completions", + completion_window=completion_window, + ) + finally: + server.shutdown() + thread.join(timeout=5) + + def test_sdk_batches_retrieve_unknown_id_is_a_real_404() -> None: server, thread = _start(TaskOrchestrator([ModelAgent("worker_agent", "mock-model", tags=("files",))])) try: @@ -249,6 +408,27 @@ def fake_proxy_send(agent: ModelAgent, endpoint: str, payload: dict) -> dict: thread.join(timeout=5) +@pytest.mark.parametrize("response_format", ["text", "srt", "vtt"]) +def test_sdk_audio_transcription_text_formats(response_format: str) -> None: + """Stock SDK receives bounded text transports for every text format.""" + agent = ModelAgent("transcribe_worker", "mock-transcribe", tags=("transcription",)) + orchestrator = TaskOrchestrator([agent]) + orchestrator.client.proxy_send = ( # type: ignore[method-assign] + lambda _agent, _endpoint, _payload: {"text": "bounded transcript"} + ) + server, thread = _start(orchestrator) + try: + transcript = _client(server.server_address[1]).audio.transcriptions.create( + file=("clip.wav", io.BytesIO(b"RIFF nonempty"), "audio/wav"), + model="mock-transcribe", + response_format=response_format, + ) + assert transcript == "bounded transcript" + finally: + server.shutdown() + thread.join(timeout=5) + + def test_sdk_chat_completions_with_audio_output_modalities() -> None: """chat.completions.create(modalities=["text","audio"], audio={...}).