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..7a8810bf9 100644 --- a/contextual_orchestrator/server.py +++ b/contextual_orchestrator/server.py @@ -183,6 +183,54 @@ def _validate_mode(mode: Any) -> str: return mode +def _normalize_message_content(content: Any) -> str: + """Normalize OpenAI message content to a plain string for orchestration. + + Accepts a string or an array of text content parts (``{"type":"text","text":...}``). + Empty or whitespace-only text parts fail closed with ``invalid_message_content``. + Non-text part types fail closed on the orchestrated route. + """ + if isinstance(content, str): + return content + if not isinstance(content, list) or not content: + raise RequestError(400, "invalid_message", "message role or content is invalid") + text_parts: list[str] = [] + for index, part in enumerate(content): + if isinstance(part, str): + if not part.strip(): + raise RequestError( + 400, + "invalid_message_content", + f"content part[{index}] text must be a non-empty string", + {"part_index": index}, + ) + text_parts.append(part) + continue + if not isinstance(part, dict): + raise RequestError(400, "invalid_message", "message role or content is invalid") + part_type = part.get("type", "text") + if part_type == "text": + text = part.get("text") + if not isinstance(text, str) or not text.strip(): + raise RequestError( + 400, + "invalid_message_content", + f"content part[{index}] text must be a non-empty string", + {"part_index": index, "part_type": "text"}, + ) + text_parts.append(text) + continue + raise RequestError( + 400, + "invalid_message_content", + f"content part type {part_type!r} is not supported on the orchestrated route", + {"part_type": part_type, "part_index": index}, + ) + if not text_parts: + raise RequestError(400, "invalid_message", "message role or content is invalid") + return "\n".join(text_parts) + + def _validate_messages(messages: Any) -> list[dict[str, str]]: if not isinstance(messages, list) or not messages: raise RequestError(400, "invalid_message", "messages must be a non-empty array") @@ -191,9 +239,9 @@ def _validate_messages(messages: Any) -> list[dict[str, str]]: if not isinstance(message, dict): raise RequestError(400, "invalid_message", "each message must be an object") role = message.get("role") - content = message.get("content") - if not isinstance(role, str) or role not in ALLOWED_MESSAGE_ROLES or not isinstance(content, str): + if not isinstance(role, str) or role not in ALLOWED_MESSAGE_ROLES: raise RequestError(400, "invalid_message", "message role or content is invalid") + content = _normalize_message_content(message.get("content")) validated.append({"role": role, "content": content}) return validated diff --git a/tests/test_empty_text_content_parts.py b/tests/test_empty_text_content_parts.py new file mode 100644 index 000000000..374753a89 --- /dev/null +++ b/tests/test_empty_text_content_parts.py @@ -0,0 +1,166 @@ +"""Reject empty text content parts; accept multi-part text substrate.""" + +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, + _normalize_message_content, + _validate_messages, + build_server, +) + +_TEST_AUTH_TOKEN = "empty_text_parts_token" # noqa: S105 + + +def build() -> TaskOrchestrator: + return TaskOrchestrator( + [ModelAgent("general_agent", "mock-generalist", tags=("reasoning", "writing"))] + ) + + +def test_normalize_accepts_nonempty_text_parts() -> None: + assert _normalize_message_content("hello") == "hello" + assert ( + _normalize_message_content( + [{"type": "text", "text": "line one"}, {"type": "text", "text": "line two"}] + ) + == "line one\nline two" + ) + + +def test_normalize_rejects_empty_text_parts() -> None: + try: + _normalize_message_content([{"type": "text", "text": ""}]) + raise AssertionError("expected invalid_message_content empty") + except RequestError as exc: + assert exc.code == "invalid_message_content" + assert exc.detail.get("part_index") == 0 + try: + _normalize_message_content([{"type": "text", "text": " "}]) + raise AssertionError("expected invalid_message_content whitespace") + except RequestError as exc: + assert exc.code == "invalid_message_content" + try: + _normalize_message_content( + [ + {"type": "text", "text": "ok"}, + {"type": "text", "text": ""}, + ] + ) + raise AssertionError("expected invalid_message_content mixed empty") + except RequestError as exc: + assert exc.code == "invalid_message_content" + assert exc.detail.get("part_index") == 1 + try: + _normalize_message_content([""]) + raise AssertionError("expected invalid_message_content bare empty string part") + except RequestError as exc: + assert exc.code == "invalid_message_content" + + +def test_validate_messages_accepts_content_parts() -> None: + validated = _validate_messages( + [ + { + "role": "user", + "content": [ + {"type": "text", "text": "Summarize"}, + {"type": "text", "text": "the architecture."}, + ], + } + ] + ) + assert validated == [ + {"role": "user", "content": "Summarize\nthe architecture."} + ] + + +def _post(port: int, payload: dict) -> tuple[int, dict]: + request = urllib.request.Request( + f"http://127.0.0.1:{port}/v1/chat/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_empty_text_content_parts() -> 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", + "messages": [ + { + "role": "user", + "content": [{"type": "text", "text": ""}], + } + ], + "mode": "route", + }, + ) + assert status == 400, body + assert body["error"]["code"] == "invalid_message_content" + finally: + server.shutdown() + thread.join(timeout=5) + + +def test_http_accepts_nonempty_text_content_parts() -> 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", + "messages": [ + { + "role": "user", + "content": [{"type": "text", "text": "Write a short hello."}], + } + ], + "mode": "route", + }, + ) + assert status == 200, body + assert body["object"] == "chat.completion" + assert body["choices"][0]["message"]["content"] + finally: + server.shutdown() + thread.join(timeout=5) + + +if __name__ == "__main__": + test_normalize_accepts_nonempty_text_parts() + test_normalize_rejects_empty_text_parts() + test_validate_messages_accepts_content_parts() + test_http_rejects_empty_text_content_parts() + test_http_accepts_nonempty_text_content_parts() + print("ok")