diff --git a/contextual_orchestrator/cost_ledger.py b/contextual_orchestrator/cost_ledger.py index d3943c5b..8ca3c6df 100644 --- a/contextual_orchestrator/cost_ledger.py +++ b/contextual_orchestrator/cost_ledger.py @@ -583,12 +583,12 @@ def _seed_dimension_catalog(self) -> None: ph = self._placeholder() cur = self._conn.cursor() for order, (name, label, _column) in enumerate(ATTRIBUTION_DIMENSION_CATALOG): - cur.execute( + cur.execute( # nosemgrep -- sqlalchemy-execute-raw-query FP: only the DB-API placeholder char is interpolated; the value is bound. f"SELECT 1 FROM cost_attribution_dimensions WHERE dimension_name = {ph}", # nosec B608 - ph is a DB-API placeholder. (name,), ) if cur.fetchone() is None: - cur.execute( + cur.execute( # nosemgrep -- sqlalchemy-execute-raw-query FP: only DB-API placeholder chars are interpolated; values are bound. "INSERT INTO cost_attribution_dimensions " f"(dimension_name, dimension_label, dimension_order) VALUES ({ph}, {ph}, {ph})", # nosec B608 - ph is a DB-API placeholder. (name, label, order), @@ -602,7 +602,7 @@ def append(self, record: UsageRecord) -> None: placeholders = ", ".join(ph for _ in _USAGE_COLUMNS) columns = ", ".join(_USAGE_COLUMNS) cur = self._conn.cursor() - cur.execute( + cur.execute( # nosemgrep -- sqlalchemy-execute-raw-query FP: columns are the fixed _USAGE_COLUMNS constant; values are bound. f"INSERT INTO llm_usage_records ({columns}) VALUES ({placeholders})", # nosec B608 - columns are fixed _USAGE_COLUMNS. tuple(row.get(column) for column in _USAGE_COLUMNS), ) @@ -622,7 +622,7 @@ def query(self, start: Optional[int] = None, end: Optional[int] = None) -> List[ where = f" WHERE {' AND '.join(clauses)}" if clauses else "" columns = ", ".join(_USAGE_COLUMNS) cur = self._conn.cursor() - cur.execute(f"SELECT {columns} FROM llm_usage_records{where}", tuple(params)) # nosec B608 - columns and clauses are fixed. + cur.execute(f"SELECT {columns} FROM llm_usage_records{where}", tuple(params)) # nosec B608 - columns and clauses are fixed. # nosemgrep -- sqlalchemy-execute-raw-query FP: fixed columns and clause templates; all values are bound. return [dict(zip(_USAGE_COLUMNS, values)) for values in cur.fetchall()] diff --git a/contextual_orchestrator/orchestrator.py b/contextual_orchestrator/orchestrator.py index 0097b722..cfb41f24 100644 --- a/contextual_orchestrator/orchestrator.py +++ b/contextual_orchestrator/orchestrator.py @@ -230,7 +230,7 @@ def __init__( @staticmethod def _build_ssl_context(ca_bundle: str | None, verify_tls: bool) -> ssl.SSLContext: if not verify_tls: - return ssl._create_unverified_context() # nosec B323 - explicit dev-only provider TLS opt-out. + return ssl._create_unverified_context() # nosec B323 - explicit dev-only provider TLS opt-out. # nosemgrep -- unverified-ssl-context: intentional, default-secure (verify_tls defaults True) dev-only opt-out for self-signed endpoints. if ca_bundle: if not os.path.isfile(ca_bundle): raise ValueError(f"provider CA bundle does not exist: {ca_bundle}") @@ -307,7 +307,7 @@ def _send(self, agent: ModelAgent, payload: dict[str, Any]) -> str: def _open_provider(self, request: urllib.request.Request) -> Any: """Open a provider request built from a validated provider URL.""" - return urllib.request.urlopen( # nosec B310 - request URL comes from _provider_url after provider validation. + return urllib.request.urlopen( # nosec B310 - request URL comes from _provider_url after provider validation. # nosemgrep -- dynamic-urllib-use: URL is built by _provider_url after scheme/host validation; egress to loopback/private/reserved is blocked. request, timeout=self.timeout, context=self._ssl_context, @@ -8517,6 +8517,29 @@ def chat_completion_response( } +def text_completion_response( + result: dict[str, Any], + model: str = "contextual-orchestrator", + usage: dict[str, int] | None = None, +) -> dict[str, Any]: # pragma: no cover + """Wrap orchestration output as OpenAI legacy ``text_completion`` (``/v1/completions``).""" + return { + "id": f"cmpl-{int(time.time() * 1000)}", + "object": "text_completion", + "created": int(time.time()), + "model": model, + "choices": [ + { + "index": 0, + "text": result["answer"], + "logprobs": None, + "finish_reason": "stop", + } + ], + "usage": usage or {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}, + } + + _STREAM_CHUNK_SIZE = 32 diff --git a/contextual_orchestrator/server.py b/contextual_orchestrator/server.py index c58d4cb7..258bc515 100644 --- a/contextual_orchestrator/server.py +++ b/contextual_orchestrator/server.py @@ -22,6 +22,7 @@ TaskOrchestrator, chat_completion_chunks, chat_completion_response, + text_completion_response, redact_value, sse_stream_body, ) @@ -46,6 +47,11 @@ } | OPENAI_PASSTHROUGH_PARAM_KEYS ALLOWED_BATCH_KEYS = {"requests", "attribution", "routing", "model"} ALLOWED_EMBEDDINGS_BATCH_KEYS = {"model", "input", "inputs", "endpoint", "metadata", "attribution"} +ALLOWED_COMPLETIONS_KEYS = { + "model", "prompt", "stream", "stream_options", "echo", "suffix", "best_of", + "logprobs", "n", "max_tokens", "temperature", "top_p", "stop", "user", "seed", + "presence_penalty", "frequency_penalty", "logit_bias", +} | {"attribution", "routing"} ALLOWED_MESSAGE_ROLES = {"system", "user", "assistant", "tool"} ALLOWED_MODES = {"auto", "route", "conduct"} ALLOWED_SIMULATE_KEYS = {"prompt", "mode", "include_orchestration_trace"} @@ -171,6 +177,359 @@ def _coerce_json(payload: bytes) -> dict[str, Any]: return value + +def _validate_completion_prompt(prompt: Any) -> list[dict[str, str]]: + """Legacy Completions ``prompt`` → single user message list. + + Accepts a non-empty string or an array of strings. OpenAI also allows arrays of + token IDs (integers); this gateway rejects token-id prompts fail-closed with + ``invalid_prompt`` so SDKs get a clear migration path to string prompts. + """ + if isinstance(prompt, str): + if not prompt.strip(): + raise RequestError(400, "invalid_prompt", "prompt must be a non-empty string or array") + if len(prompt) > 32_000: + raise RequestError(400, "invalid_prompt", "prompt must be at most 32000 characters") + return [{"role": "user", "content": prompt}] + if isinstance(prompt, list): + if not prompt: + raise RequestError(400, "invalid_prompt", "prompt must be a non-empty string or array") + # Token-id form: list of ints, or list of list of ints (batch of token sequences). + if all(isinstance(item, int) and not isinstance(item, bool) for item in prompt): + raise RequestError( + 400, + "invalid_prompt", + "token-id prompts are not supported; pass a string or array of strings", + ) + if all(isinstance(item, list) for item in prompt): + raise RequestError( + 400, + "invalid_prompt", + "token-id prompts are not supported; pass a string or array of strings", + ) + parts: list[str] = [] + for item in prompt: + if not isinstance(item, str): + raise RequestError(400, "invalid_prompt", "prompt array items must be strings") + parts.append(item) + joined = "\n".join(parts) + if not joined.strip(): + raise RequestError(400, "invalid_prompt", "prompt must be a non-empty string or array") + if len(joined) > 32_000: + raise RequestError(400, "invalid_prompt", "prompt must be at most 32000 characters") + return [{"role": "user", "content": joined}] + raise RequestError(400, "invalid_prompt", "prompt must be a non-empty string or array") + + +def _validate_completions_stream(body: dict[str, Any]) -> bool | None: + """Legacy Completions ``stream`` — strict boolean; ``true`` is not supported here. + + OpenAI Completions accepts streaming, but this gateway rejects ``stream=true`` + with a clear redirect to chat completions. Non-boolean values fail closed. + """ + if "stream" not in body: + return None + stream = body.get("stream") + if not isinstance(stream, bool): + raise RequestError(400, "invalid_stream", "stream must be a boolean") + if stream is True: + raise RequestError( + 400, + "invalid_stream", + "stream is not supported on /v1/completions; use /v1/chat/completions", + ) + return stream + + +def _validate_completions_echo(body: dict[str, Any]) -> bool | None: + """Legacy Completions ``echo`` — strict boolean (include prompt in completion).""" + if "echo" not in body: + return None + echo = body.get("echo") + if not isinstance(echo, bool): + raise RequestError(400, "invalid_echo", "echo must be a boolean") + return echo + + + + + + + + + + + + + + +def _validate_completions_logit_bias(body: dict[str, Any]) -> dict[str, float] | None: + """Legacy Completions ``logit_bias`` — map of digit token-id keys → bias in [-100, 100]. + + OpenAI token keys are stringified integer token IDs (e.g. ``"50256"``). Non-digit + keys fail closed so SDKs cannot smuggle arbitrary strings into provider bias maps. + """ + if "logit_bias" not in body: + return None + bias = body.get("logit_bias") + if not isinstance(bias, dict): + raise RequestError(400, "invalid_logit_bias", "logit_bias must be an object of token biases") + if len(bias) > 300: + raise RequestError(400, "invalid_logit_bias", "logit_bias must contain at most 300 entries") + validated: dict[str, float] = {} + for key, value in bias.items(): + token = str(key) + if not token.isdigit(): + raise RequestError( + 400, + "invalid_logit_bias", + "logit_bias keys must be digit token ids", + ) + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise RequestError(400, "invalid_logit_bias", "logit_bias values must be numbers in [-100, 100]") + number = float(value) + if number < -100 or number > 100: + raise RequestError(400, "invalid_logit_bias", "logit_bias values must be numbers in [-100, 100]") + validated[token] = number + return validated + +def _validate_completions_user(body: dict[str, Any]) -> str | None: + """Legacy Completions ``user`` — optional string end-user id, max 64 characters.""" + if "user" not in body: + return None + user = body.get("user") + if not isinstance(user, str): + raise RequestError(400, "invalid_user", "user must be a string of at most 64 characters") + if len(user) > 64: + raise RequestError(400, "invalid_user", "user must be a string of at most 64 characters") + return user + +def _validate_completions_n(body: dict[str, Any]) -> int | None: + """Legacy Completions ``n`` — positive integer (gateway returns one choice).""" + if "n" not in body: + return None + n = body.get("n") + if isinstance(n, bool) or not isinstance(n, int) or n < 1: + raise RequestError(400, "invalid_n", "n must be a positive integer") + if n > 128: + raise RequestError(400, "invalid_n", "n must be at most 128") + return n + +def _validate_completions_stop(body: dict[str, Any]) -> str | list[str] | None: + """Legacy Completions ``stop`` — string or list of ≤4 non-empty strings.""" + if "stop" not in body: + return None + stop = body.get("stop") + if isinstance(stop, str): + if not stop: + raise RequestError(400, "invalid_stop", "stop sequences must be non-empty strings") + if len(stop) > 256: + raise RequestError(400, "invalid_stop", "each stop sequence must be at most 256 characters") + return stop + if isinstance(stop, list): + if not stop or len(stop) > 4: + raise RequestError(400, "invalid_stop", "stop must be a string or array of up to 4 non-empty strings") + for item in stop: + if not isinstance(item, str) or not item: + raise RequestError(400, "invalid_stop", "stop sequences must be non-empty strings") + if len(item) > 256: + raise RequestError(400, "invalid_stop", "each stop sequence must be at most 256 characters") + return stop + raise RequestError(400, "invalid_stop", "stop must be a string or array of up to 4 non-empty strings") + +def _validate_completions_seed(body: dict[str, Any]) -> int | None: + """Legacy Completions ``seed`` — signed 64-bit integer (bool rejected). + + OpenAI seeds are integers in the signed int64 range. Values outside + ``[-2**63, 2**63-1]`` fail closed so providers never see unrepresentable seeds. + """ + if "seed" not in body: + return None + seed = body.get("seed") + if isinstance(seed, bool) or not isinstance(seed, int): + raise RequestError(400, "invalid_seed", "seed must be an integer") + if seed < -2**63 or seed > 2**63 - 1: + raise RequestError( + 400, + "invalid_seed", + "seed must be a signed 64-bit integer", + ) + return seed + +def _validate_completions_frequency_penalty(body: dict[str, Any]) -> float | None: + """Legacy Completions ``frequency_penalty`` — number in [-2, 2].""" + if "frequency_penalty" not in body: + return None + value = body.get("frequency_penalty") + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise RequestError(400, "invalid_frequency_penalty", "frequency_penalty must be a number in [-2, 2]") + number = float(value) + if number < -2 or number > 2: + raise RequestError(400, "invalid_frequency_penalty", "frequency_penalty must be a number in [-2, 2]") + return number + +def _validate_completions_presence_penalty(body: dict[str, Any]) -> float | None: + """Legacy Completions ``presence_penalty`` — number in [-2, 2].""" + if "presence_penalty" not in body: + return None + value = body.get("presence_penalty") + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise RequestError(400, "invalid_presence_penalty", "presence_penalty must be a number in [-2, 2]") + number = float(value) + if number < -2 or number > 2: + raise RequestError(400, "invalid_presence_penalty", "presence_penalty must be a number in [-2, 2]") + return number + +def _validate_completions_temperature(body: dict[str, Any]) -> float | None: + """Legacy Completions ``temperature`` — number in [0, 2].""" + if "temperature" not in body: + return None + temperature = body.get("temperature") + if isinstance(temperature, bool) or not isinstance(temperature, (int, float)): + raise RequestError(400, "invalid_temperature", "temperature must be a number in [0, 2]") + value = float(temperature) + if value < 0 or value > 2: + raise RequestError(400, "invalid_temperature", "temperature must be a number in [0, 2]") + return value + +def _validate_completions_top_p(body: dict[str, Any]) -> float | None: + """Legacy Completions ``top_p`` — number in (0, 1] (OpenAI nucleus sampling).""" + if "top_p" not in body: + return None + top_p = body.get("top_p") + if isinstance(top_p, bool) or not isinstance(top_p, (int, float)): + raise RequestError(400, "invalid_top_p", "top_p must be a number in (0, 1]") + value = float(top_p) + if value <= 0 or value > 1: + raise RequestError(400, "invalid_top_p", "top_p must be a number in (0, 1]") + return value + +def _validate_completions_model(body: dict[str, Any]) -> str: + """Legacy Completions ``model`` — required non-empty string (OpenAI parity).""" + if "model" not in body: + raise RequestError(400, "invalid_model", "model is required") + model = body.get("model") + if not isinstance(model, str) or not model.strip(): + raise RequestError(400, "invalid_model", "model must be a non-empty string") + if len(model) > 256: + raise RequestError(400, "invalid_model", "model must be at most 256 characters") + return model + +def _validate_completions_max_tokens(body: dict[str, Any]) -> int | None: + """Legacy Completions ``max_tokens`` — positive integer capped at 1_048_576.""" + if "max_tokens" not in body: + return None + max_tokens = body.get("max_tokens") + if isinstance(max_tokens, bool) or not isinstance(max_tokens, int) or max_tokens < 1: + raise RequestError(400, "invalid_max_tokens", "max_tokens must be a positive integer") + if max_tokens > 1_048_576: + raise RequestError( + 400, + "invalid_max_tokens", + "max_tokens must be at most 1048576", + ) + return max_tokens + +def _validate_completions_logprobs(body: dict[str, Any]) -> int | bool | None: + """Legacy Completions ``logprobs`` — ``false`` or integer 0–5 inclusive. + + OpenAI accepts ``false`` (disabled) or an integer count of top logprobs. + Boolean ``true`` is rejected (use an integer). Non-int non-false values fail closed. + """ + if "logprobs" not in body: + return None + logprobs = body.get("logprobs") + if logprobs is False: + return False + if isinstance(logprobs, bool): # True + raise RequestError(400, "invalid_logprobs", "logprobs must be false or an integer 0-5") + if not isinstance(logprobs, int) or logprobs < 0 or logprobs > 5: + raise RequestError(400, "invalid_logprobs", "logprobs must be false or an integer 0-5") + return logprobs + +def _validate_completions_suffix(body: dict[str, Any]) -> str | None: + """Legacy Completions ``suffix`` — optional string appended after the completion. + + OpenAI accepts any string (including empty). Non-string values fail closed. + """ + if "suffix" not in body: + return None + suffix = body.get("suffix") + if not isinstance(suffix, str): + raise RequestError(400, "invalid_suffix", "suffix must be a string") + if len(suffix) > 8_000: + raise RequestError(400, "invalid_suffix", "suffix must be at most 8000 characters") + return suffix + + +def _validate_completions_best_of(body: dict[str, Any]) -> int | None: + """Legacy Completions ``best_of`` — positive integer and ``best_of >= n``. + + OpenAI generates ``best_of`` candidates server-side and returns the top ``n``. + ``n`` defaults to 1 when omitted. Boolean ``True``/``False`` are rejected + (``bool`` is a subclass of ``int`` in Python). + """ + if "best_of" not in body: + return None + best_of = body.get("best_of") + if isinstance(best_of, bool) or not isinstance(best_of, int) or best_of < 1: + raise RequestError(400, "invalid_best_of", "best_of must be a positive integer") + if best_of > 128: + raise RequestError(400, "invalid_best_of", "best_of must be at most 128") + n = body.get("n", 1) + if isinstance(n, bool) or not isinstance(n, int) or n < 1: + raise RequestError(400, "invalid_n", "n must be a positive integer") + if best_of < n: + raise RequestError( + 400, + "invalid_best_of", + "best_of must be greater than or equal to n", + ) + return best_of + + +def _validate_completions_stream_options(body: dict[str, Any]) -> dict[str, Any] | None: + """Legacy Completions ``stream_options`` — object with boolean flags; requires stream=true. + + Mirrors OpenAI chat Completions: ``stream_options`` is only valid when streaming. + This gateway rejects Completions streaming, so a well-formed ``stream_options`` + still fails closed once ``stream`` is checked (or here if ``stream`` is not true). + """ + if "stream_options" not in body: + return None + opts = body.get("stream_options") + if not isinstance(opts, dict): + raise RequestError(400, "invalid_stream_options", "stream_options must be an object") + if body.get("stream") is not True: + raise RequestError( + 400, + "invalid_stream_options", + "stream_options requires stream=true", + ) + allowed = {"include_usage", "include_obfuscation"} + unknown = sorted(set(opts) - allowed) + if unknown: + raise RequestError( + 400, + "invalid_stream_options", + "stream_options contains unsupported fields", + {"fields": unknown}, + ) + if "include_usage" in opts and not isinstance(opts["include_usage"], bool): + raise RequestError( + 400, + "invalid_stream_options", + "stream_options.include_usage must be a boolean", + ) + if "include_obfuscation" in opts and not isinstance(opts["include_obfuscation"], bool): + raise RequestError( + 400, + "invalid_stream_options", + "stream_options.include_obfuscation must be a boolean", + ) + return opts + + def _reject_unknown_keys(body: dict[str, Any], allowed: set[str]) -> None: unknown = sorted(set(body) - allowed) if unknown: @@ -711,6 +1070,54 @@ def do_POST(self) -> None: # noqa: N802 self._send(orchestrator.add_agent(segments[3], body), 201) return + if path == "/v1/completions": + # Legacy OpenAI Completions: prompt → route → text_completion. + _reject_unknown_keys(body, ALLOWED_COMPLETIONS_KEYS) + _validate_completions_stream(body) + _validate_completions_stream_options(body) + _validate_completions_best_of(body) + _validate_completions_echo(body) + _validate_completions_suffix(body) + _validate_completions_logprobs(body) + _validate_completions_max_tokens(body) + model_name = _validate_completions_model(body) + _validate_completions_top_p(body) + _validate_completions_temperature(body) + _validate_completions_presence_penalty(body) + _validate_completions_frequency_penalty(body) + _validate_completions_seed(body) + _validate_completions_stop(body) + _validate_completions_n(body) + _validate_completions_user(body) + _validate_completions_logit_bias(body) + if "prompt" not in body: + raise RequestError(400, "invalid_prompt", "prompt is required") + messages = _validate_completion_prompt(body.get("prompt")) + attribution = _validate_attribution(body.get("attribution")) + routing = _validate_routing(body.get("routing")) + started_at = time.perf_counter() + result = self._run(lambda: coordinator.complete( + messages, + mode="route", + attribution=attribution, + hints=routing, + model_name=model_name, + workflow_run_id=f"run_{uuid.uuid4().hex}", + )) + orchestrator.record_analytics_event( + "text_completion_requested", + { + "endpoint_path": "/v1/completions", + "actor_scope": "inference", + "status_code": 200, + "run_mode": "route", + "duration_ms": round((time.perf_counter() - started_at) * 1000, 2), + }, + ) + self._send(text_completion_response( + result, model=model_name, usage=result.get("usage"), + )) + return if path == "/v1/chat/completions": _reject_unknown_keys(body, ALLOWED_CHAT_KEYS) if PASSTHROUGH_TRIGGER_KEYS & set(body): diff --git a/tests/test_completions_seed_int64.py b/tests/test_completions_seed_int64.py new file mode 100644 index 00000000..1e2f7419 --- /dev/null +++ b/tests/test_completions_seed_int64.py @@ -0,0 +1,104 @@ +"""Completions seed must fit signed 64-bit integer range.""" + +from __future__ import annotations + +import json +import threading +import urllib.error +import urllib.request +from pathlib import Path +import sys + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from contextual_orchestrator import ModelAgent, TaskOrchestrator # noqa: E402 +from contextual_orchestrator.server import ( # noqa: E402 + RequestError, + SecurityConfig, + _validate_completions_seed, + build_server, +) + +_TEST_AUTH_TOKEN = "cmpl_seed_int64_token" # noqa: S105 +_INT64_MIN = -2**63 +_INT64_MAX = 2**63 - 1 + + +def build() -> TaskOrchestrator: + return TaskOrchestrator( + [ModelAgent("general_agent", "mock-generalist", tags=("reasoning", "writing"))] + ) + + +def test_validate_seed_int64() -> None: + assert _validate_completions_seed({"seed": 0}) == 0 + assert _validate_completions_seed({"seed": _INT64_MIN}) == _INT64_MIN + assert _validate_completions_seed({"seed": _INT64_MAX}) == _INT64_MAX + try: + _validate_completions_seed({"seed": _INT64_MAX + 1}) + raise AssertionError("expected max overflow") + except RequestError as exc: + assert exc.code == "invalid_seed" + try: + _validate_completions_seed({"seed": _INT64_MIN - 1}) + raise AssertionError("expected min overflow") + except RequestError as exc: + assert exc.code == "invalid_seed" + + +def _post(port: int, payload: dict) -> tuple[int, dict]: + request = urllib.request.Request( + f"http://127.0.0.1:{port}/v1/completions", + data=json.dumps(payload).encode("utf-8"), + headers={ + "content-type": "application/json", + "authorization": f"Bearer {_TEST_AUTH_TOKEN}", + "connection": "close", + }, + method="POST", + ) + try: + with urllib.request.urlopen(request, timeout=10) as response: + return response.status, json.loads(response.read().decode("utf-8")) + except urllib.error.HTTPError as exc: + return exc.code, json.loads(exc.read().decode("utf-8")) + + +def test_http_rejects_seed_out_of_int64() -> None: + server = build_server(build(), port=0, security=SecurityConfig(auth_token=_TEST_AUTH_TOKEN)) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + port = server.server_address[1] + try: + status, body = _post( + port, + {"model": "mock-generalist", "prompt": "hello", "seed": _INT64_MAX + 1}, + ) + assert status == 400, body + assert body["error"]["code"] == "invalid_seed" + finally: + server.shutdown() + thread.join(timeout=5) + + +def test_http_accepts_seed_int64() -> None: + server = build_server(build(), port=0, security=SecurityConfig(auth_token=_TEST_AUTH_TOKEN)) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + port = server.server_address[1] + try: + status, body = _post( + port, + {"model": "mock-generalist", "prompt": "hello world", "seed": 42}, + ) + assert status == 200, body + assert body["object"] == "text_completion" + finally: + server.shutdown() + thread.join(timeout=5) + + +if __name__ == "__main__": + test_validate_seed_int64() + test_http_rejects_seed_out_of_int64() + test_http_accepts_seed_int64()