feat(api): validate embeddings encoding_format, dimensions, and user

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()]
 
 

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,

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)

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 |

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")