diff --git a/contextual_orchestrator/cost_ledger.py b/contextual_orchestrator/cost_ledger.py index d3943c5be..8ca3c6dfd 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 0097b722e..974d26c55 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, diff --git a/contextual_orchestrator/server.py b/contextual_orchestrator/server.py index c58d4cb79..91ea164e0 100644 --- a/contextual_orchestrator/server.py +++ b/contextual_orchestrator/server.py @@ -45,7 +45,7 @@ "model", "input", "instructions", "stream", "metadata", "reasoning", } | OPENAI_PASSTHROUGH_PARAM_KEYS ALLOWED_BATCH_KEYS = {"requests", "attribution", "routing", "model"} -ALLOWED_EMBEDDINGS_BATCH_KEYS = {"model", "input", "inputs", "endpoint", "metadata", "attribution"} +ALLOWED_EMBEDDINGS_BATCH_KEYS = {"model", "input", "inputs", "endpoint", "encoding_format", "dimensions", "user", "metadata", "attribution"} ALLOWED_MESSAGE_ROLES = {"system", "user", "assistant", "tool"} ALLOWED_MODES = {"auto", "route", "conduct"} ALLOWED_SIMULATE_KEYS = {"prompt", "mode", "include_orchestration_trace"} @@ -177,6 +177,47 @@ def _reject_unknown_keys(body: dict[str, Any], allowed: set[str]) -> None: raise RequestError(400, "unknown_fields", "request contains unsupported fields", {"fields": unknown}) + +def _validate_embeddings_encoding_format(body: dict[str, Any]) -> str | None: + """OpenAI embeddings ``encoding_format`` — float or base64 when present.""" + if "encoding_format" not in body: + return None + encoding_format = body.get("encoding_format") + if encoding_format not in {"float", "base64"}: + raise RequestError( + 400, + "invalid_encoding_format", + "encoding_format must be float or base64", + ) + return encoding_format + + +def _validate_embeddings_dimensions(body: dict[str, Any]) -> int | None: + """OpenAI embeddings ``dimensions`` — positive integer when present.""" + if "dimensions" not in body: + return None + dimensions = body.get("dimensions") + if not isinstance(dimensions, int) or isinstance(dimensions, bool) or dimensions < 1: + raise RequestError( + 400, + "invalid_dimensions", + "dimensions must be a positive integer", + ) + return dimensions + + +def _validate_embeddings_user(body: dict[str, Any]) -> str | None: + """OpenAI embeddings ``user`` — non-empty string ≤64 when present.""" + if "user" not in body: + return None + user = body.get("user") + if not isinstance(user, str) or not user.strip(): + raise RequestError(400, "invalid_user", "user must be a non-empty string") + if len(user) > 64: + raise RequestError(400, "invalid_user", "user must be at most 64 characters") + return user + + def _validate_mode(mode: Any) -> str: if not isinstance(mode, str) or mode not in ALLOWED_MODES: raise RequestError(400, "invalid_mode", "mode must be auto, route, or conduct") @@ -799,6 +840,9 @@ def do_POST(self) -> None: # noqa: N802 return if path == "/v1/batch/embeddings": _reject_unknown_keys(body, ALLOWED_EMBEDDINGS_BATCH_KEYS) + _validate_embeddings_encoding_format(body) + _validate_embeddings_dimensions(body) + _validate_embeddings_user(body) inputs = _validate_embeddings_inputs(body) model_name = str(body.get("model", "contextual-orchestrator")) attribution = _embeddings_attribution(body) diff --git a/docs/rest_api_design.md b/docs/rest_api_design.md index 9378e5a37..9a924aa52 100644 --- a/docs/rest_api_design.md +++ b/docs/rest_api_design.md @@ -15,7 +15,7 @@ |---|---|---| | `GET` | `/openapi.json` | API contract | | `POST` | `/v1/chat/completions` | Compatibility chat endpoint | -| `POST` | `/v1/batch/embeddings` | Submit a bulk, latency-tolerant embeddings batch; oversized inputs are token-split before routing via pg-llm-batch | +| `POST` | `/v1/batch/embeddings` | Embeddings batch; accepts encoding_format float|base64, positive dimensions, user string | | `GET` | `/v1/batch/embeddings/{batch_id}` | Poll an embeddings batch; returns reduced vectors + recorded cost once completed | | `GET` | `/api/v1/agent_pools` | List model agents | | `GET` | `/api/v1/orchestration_policies/default_policy` | Read active policy | diff --git a/tests/test_embeddings_encoding_dimensions.py b/tests/test_embeddings_encoding_dimensions.py new file mode 100644 index 000000000..ceb7dfc49 --- /dev/null +++ b/tests/test_embeddings_encoding_dimensions.py @@ -0,0 +1,129 @@ +"""OpenAI embeddings encoding_format, dimensions, and user validation on batch path.""" + +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_embeddings_dimensions, + _validate_embeddings_encoding_format, + _validate_embeddings_user, + build_server, +) + +_TEST_AUTH_TOKEN = "emb_enc_token" # noqa: S105 + + +def build() -> TaskOrchestrator: + return TaskOrchestrator( + [ModelAgent("general_agent", "mock-generalist", tags=("reasoning", "writing"))] + ) + + +def test_validate_encoding_format_dimensions_user() -> None: + assert _validate_embeddings_encoding_format({}) is None + assert _validate_embeddings_encoding_format({"encoding_format": "float"}) == "float" + assert _validate_embeddings_encoding_format({"encoding_format": "base64"}) == "base64" + try: + _validate_embeddings_encoding_format({"encoding_format": "json"}) + raise AssertionError("bad format") + except RequestError as exc: + assert exc.code == "invalid_encoding_format" + assert _validate_embeddings_dimensions({"dimensions": 256}) == 256 + try: + _validate_embeddings_dimensions({"dimensions": 0}) + raise AssertionError("zero") + except RequestError as exc: + assert exc.code == "invalid_dimensions" + try: + _validate_embeddings_dimensions({"dimensions": True}) + raise AssertionError("bool") + except RequestError as exc: + assert exc.code == "invalid_dimensions" + assert _validate_embeddings_user({"user": "acct_1"}) == "acct_1" + try: + _validate_embeddings_user({"user": ""}) + raise AssertionError("empty user") + except RequestError as exc: + assert exc.code == "invalid_user" + + +def test_http_encoding_format_accepted() -> 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: + request = urllib.request.Request( + f"http://127.0.0.1:{port}/v1/batch/embeddings", + data=json.dumps( + { + "input": ["hello"], + "model": "mock-embed", + "encoding_format": "float", + "dimensions": 8, + "user": "buyer-1", + } + ).encode("utf-8"), + headers={ + "content-type": "application/json", + "authorization": f"Bearer {_TEST_AUTH_TOKEN}", + "connection": "close", + }, + method="POST", + ) + with urllib.request.urlopen(request, timeout=10) as response: + assert response.status in {200, 202} + finally: + server.shutdown() + thread.join(timeout=5) + + +def test_http_bad_encoding_format_rejected() -> 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: + request = urllib.request.Request( + f"http://127.0.0.1:{port}/v1/batch/embeddings", + data=json.dumps( + { + "input": ["hello"], + "encoding_format": "hex", + } + ).encode("utf-8"), + headers={ + "content-type": "application/json", + "authorization": f"Bearer {_TEST_AUTH_TOKEN}", + "connection": "close", + }, + method="POST", + ) + try: + urllib.request.urlopen(request, timeout=5) + raise AssertionError("expected 400") + except urllib.error.HTTPError as exc: + body = json.loads(exc.read().decode("utf-8")) + assert exc.code == 400 + assert body["error"]["code"] == "invalid_encoding_format" + finally: + server.shutdown() + thread.join(timeout=5) + + +if __name__ == "__main__": + test_validate_encoding_format_dimensions_user() + test_http_encoding_format_accepted() + test_http_bad_encoding_format_rejected() + print("ok")