diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 000000000..939cee6c3 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,10 @@ +# Changelog + +## Unreleased + +### Security +- Minimal unauthenticated `/healthz` liveness; admin `/readyz` inventory (issue #118). +- Fail-closed Content-Length framing for JSON bodies (issue #119). +- Orchestration traces require authority beyond inference scope (issue #117). + + diff --git a/README.md b/README.md index 65f57dd4c..b66deaeb2 100644 --- a/README.md +++ b/README.md @@ -192,7 +192,7 @@ is read from a **KV config store**, never `os.getenv`. backend (local in-process backend standalone), and records one usage-ledger row per original vector with the full attribution dimensions (service, team, group, company, provider) carried in `metadata`. -- **Health.** `GET /healthz` is an unauthenticated liveness probe. +- **Health.** `GET /healthz` is an unauthenticated minimal liveness probe (`status` + `service` only). `GET /readyz` is admin-authenticated readiness with agent/backend/usage inventory for operators. - **Standalone + optional pg-llm-batch integration.** The hub runs standalone with the in-memory config store and local batch backend; wiring a Postgres DSN and an installed/deployed `pg_llm_batch` client activates the KV/secret stores, 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..bc27c6868 100644 --- a/contextual_orchestrator/server.py +++ b/contextual_orchestrator/server.py @@ -116,6 +116,23 @@ def authorize(self, headers: Any, scope: str, client_address: str) -> None: if not expected or not secrets.compare_digest(token, expected): raise RequestError(401, "unauthorized", "bearer token is invalid for this scope") + def may_disclose_trace(self, headers: Any, scope: str) -> bool: + """Return True when the verified caller may receive orchestration traces. + + Inference-only credentials never receive planner/workflow evidence. + Admin-scope callers may. Single-token deployments may when the host sets + ``expose_trace_by_default`` or the request is handled under admin scope. + """ + if scope == "admin": + return True + if scope != "inference": + return False + # Split-token inference path: never disclose traces. + if self.admin_token and self.inference_token and not self.auth_token: + return False + # Single shared token: host policy gate only (still needs request bool). + return bool(self.expose_trace_by_default) + def check_rate_limit(self, key: str) -> None: """Apply a simple per-client fixed-window request budget.""" now = time.monotonic() @@ -171,6 +188,69 @@ def _coerce_json(payload: bytes) -> dict[str, Any]: return value +def _coerce_optional_bool(value: Any, field_name: str) -> bool | None: + """Parse an optional JSON bool fail-closed; reject truthy strings/numbers.""" + if value is None: + return None + if isinstance(value, bool): + return value + raise RequestError( + 400, + "invalid_boolean", + f"{field_name} must be a JSON boolean", + {"field": field_name}, + ) + + +def _parse_content_length(headers: Any, max_body_bytes: int) -> int: + """Return a validated Content-Length for fixed-length JSON request bodies. + + Rejects missing, negative, non-decimal, overflow, and transfer-coded framing + so ``rfile.read`` never receives a negative size or unbounded read. + """ + # BaseHTTPRequestHandler collapses duplicate headers with commas; treat that + # as ambiguous framing and reject rather than guessing. + raw = headers.get("content-length") + if raw is None or raw == "": + raise RequestError(411, "length_required", "Content-Length is required for JSON request bodies") + if isinstance(raw, str) and "," in raw: + raise RequestError(400, "invalid_content_length", "duplicate or ambiguous Content-Length") + text = str(raw).strip() + if not text.isdigit(): + # Reject signed forms ("-1", "+10"), hex, and non-decimal tokens. + raise RequestError(400, "invalid_content_length", "Content-Length must be an unsigned decimal integer") + # Leading zeros are allowed by isdigit; value itself must fit max. + body_size = int(text) + if body_size > max_body_bytes: + raise RequestError(413, "request_too_large", "request body exceeds configured limit") + transfer = (headers.get("transfer-encoding") or "").strip() + if transfer and transfer.lower() != "identity": + raise RequestError( + 400, + "unsupported_transfer_encoding", + "chunked or non-identity Transfer-Encoding is not accepted for JSON bodies", + ) + return body_size + + +def _resolve_include_trace( + body: dict[str, Any], + security: SecurityConfig, + headers: Any, + scope: str, +) -> bool: + """Fail-closed orchestration-trace disclosure decision. + + Requires (1) verified trace authority for the caller scope and (2) an + explicit JSON boolean request flag when the host default is off. + """ + if not security.may_disclose_trace(headers, scope): + return False + requested = _coerce_optional_bool(body.get("include_orchestration_trace"), "include_orchestration_trace") + if requested is None: + return bool(security.expose_trace_by_default) + return requested + def _reject_unknown_keys(body: dict[str, Any], allowed: set[str]) -> None: unknown = sorted(set(body) - allowed) if unknown: @@ -336,9 +416,14 @@ def do_GET(self) -> None: # noqa: N802 self._send(OPENAPI_SPEC) return if path == "/healthz": - # Unauthenticated liveness probe for containers/orchestrators. + # Unauthenticated liveness only: process is up. No inventory. + self._send({"status": "ok", "service": "contextual-orchestrator"}) + return + if path == "/readyz": + # Authenticated readiness/diagnostics: operator inventory. + self._authorize("admin") self._send({ - "status": "ok", + "status": "ready", "service": "contextual-orchestrator", "agent_count": len(orchestrator.agents), "batch_backend": coordinator.batch_backend.name, @@ -733,7 +818,7 @@ def do_POST(self) -> None: # noqa: N802 return messages = _validate_messages(body.get("messages")) mode = _validate_mode(body.get("orchestration") or body.get("orchestration_mode") or body.get("mode") or "auto") - include_trace = bool(body.get("include_orchestration_trace", security.expose_trace_by_default)) + include_trace = _resolve_include_trace(body, security, self.headers, scope) stream = body.get("stream", False) if not isinstance(stream, bool): raise RequestError(400, "invalid_request", "stream must be a boolean") @@ -856,7 +941,7 @@ def do_POST(self) -> None: # noqa: N802 except KeyError: self._send_error(404, "batch_job_not_found", f"batch job {job_id} not found") return - self._send(_response_payload(retrieved, include_trace=True)) + self._send(_response_payload(retrieved, include_trace=security.may_disclose_trace(self.headers, "inference"))) return if path == "/v1/responses": # The Responses API has no chat-completions verifier equivalent, @@ -884,7 +969,7 @@ def do_POST(self) -> None: # noqa: N802 if not isinstance(prompt, str): raise RequestError(400, "invalid_request", "prompt must be a string") mode = _validate_mode(body.get("mode", "auto")) - include_trace = bool(body.get("include_orchestration_trace", security.expose_trace_by_default)) + include_trace = _resolve_include_trace(body, security, self.headers, scope) result = self._run(lambda: orchestrator.run([{"role": "user", "content": prompt}], mode=mode)) self._send(_response_payload(result, include_trace)) return @@ -894,7 +979,7 @@ def do_POST(self) -> None: # noqa: N802 if not isinstance(prompt, str) or not prompt: raise RequestError(400, "invalid_request", "prompt_text is required") mode = _validate_mode(body.get("run_mode", "auto")) - include_trace = bool(body.get("include_orchestration_trace", security.expose_trace_by_default)) + include_trace = _resolve_include_trace(body, security, self.headers, scope) result = self._run(lambda: orchestrator.run([{"role": "user", "content": prompt}], mode=mode)) self._send(_response_payload(result, include_trace), 201) return @@ -906,7 +991,7 @@ def do_POST(self) -> None: # noqa: N802 if not isinstance(prompts, list) or not prompts: raise RequestError(400, "invalid_request", "prompts must be a non-empty array") mode = _validate_mode(body.get("run_mode", "auto")) - include_trace = bool(body.get("include_orchestration_trace", security.expose_trace_by_default)) + include_trace = _resolve_include_trace(body, security, self.headers, scope) evaluation_run = self._run(lambda: orchestrator.run_evaluation([str(item) for item in prompts], mode=mode)) self._send(_response_payload(evaluation_run, include_trace), 201) return @@ -960,10 +1045,10 @@ def _parse_optional_int(self, query: dict[str, list[str]], field_name: str) -> i def _read_json(self) -> dict[str, Any]: if self.headers.get("content-type", "").split(";", 1)[0].strip().lower() != "application/json": raise RequestError(415, "unsupported_media_type", "content-type must be application/json") - body_size = int(self.headers.get("content-length", "0")) - if body_size > security.max_body_bytes: - raise RequestError(413, "request_too_large", "request body exceeds configured limit") + body_size = _parse_content_length(self.headers, security.max_body_bytes) raw = self.rfile.read(body_size) + if len(raw) != body_size: + raise RequestError(400, "incomplete_body", "request body shorter than Content-Length") return _coerce_json(raw) if raw else {} def log_message(self, format: str, *args: object) -> None: diff --git a/docs/architecture.md b/docs/architecture.md index c0f63a81e..2c6b6328d 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -53,3 +53,21 @@ The product is not a Fugu clone. It is a control-plane prototype for the same pu - replayable evaluation runs before any learned coordinator replaces the deterministic policy. See [product_planning.md](product_planning.md) for the product reboot. + +## Health and readiness + +Unauthenticated `GET /healthz` returns only process liveness (`status`, `service`). +Operator inventory (agent counts, batch backends, usage counts) is on +authenticated `GET /readyz` (admin scope). Inference callers cannot request +orchestration traces without verified admin/trace authority. + + +## Research citations (APA 7th) + +Chen, L., Zaharia, M., & Zou, J. (2023). *FrugalGPT: How to use large language models while reducing cost and improving performance* (arXiv:2305.05176). https://doi.org/10.48550/arXiv.2305.05176 + +Ong, I., Almahairi, A., Wu, V., Chiang, W.-L., Wu, T., Gonzalez, J. E., Kadous, M. W., & Stoica, I. (2024). *RouteLLM: Learning to route LLMs with preference data* (arXiv:2406.18665). https://doi.org/10.48550/arXiv.2406.18665 + +Ding, D., Mallick, A., Wang, C., Sim, R., Mukherjee, S., Rühle, V., Lakshmanan, L. V. S., & Awadallah, A. H. (2024). *Hybrid LLM: Cost-efficient and quality-aware query routing* (arXiv:2404.14618). https://doi.org/10.48550/arXiv.2404.14618 + +National Institute of Standards and Technology. (2022). *Secure software development framework (SSDF) version 1.1* (NIST SP 800-218). https://doi.org/10.6028/NIST.SP.800-218 diff --git a/tests/test_cost_review_server.py b/tests/test_cost_review_server.py index fd27c5bc9..55cc96f9e 100644 --- a/tests/test_cost_review_server.py +++ b/tests/test_cost_review_server.py @@ -56,9 +56,7 @@ def test_healthz_is_unauthenticated_and_ok() -> None: finally: server.shutdown() assert status == 200 - assert body["status"] == "ok" - assert body["service"] == "contextual-orchestrator" - assert "batch_backend" in body + assert body == {"status": "ok", "service": "contextual-orchestrator"} def test_chat_completion_reports_real_usage_and_records_cost() -> None: diff --git a/tests/test_healthz.py b/tests/test_healthz.py index 90dfc03c0..668ad1b41 100644 --- a/tests/test_healthz.py +++ b/tests/test_healthz.py @@ -1,9 +1,10 @@ -"""Container liveness probe: /healthz must answer without any auth token.""" +"""Container liveness vs readiness probe contracts.""" from __future__ import annotations import json import sys import threading +import urllib.error import urllib.request from pathlib import Path @@ -14,18 +15,24 @@ from contextual_orchestrator import ModelAgent, TaskOrchestrator # noqa: E402 from contextual_orchestrator.server import SecurityConfig, build_server # noqa: E402 +_TEST_ADMIN = "admin_secret" # noqa: S105 +_TEST_INFERENCE = "inference_secret" # noqa: S105 -def test_healthz_is_unauthenticated_liveness() -> None: + +def _start(): orchestrator = TaskOrchestrator([ModelAgent("probe_agent", "mock-agent", tags=("reasoning",))]) server = build_server( orchestrator, port=0, - security=SecurityConfig(admin_token="admin_secret", inference_token="inference_secret"), + security=SecurityConfig(admin_token=_TEST_ADMIN, inference_token=_TEST_INFERENCE), ) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() - port = server.server_address[1] + return server, thread, server.server_address[1] + +def test_healthz_is_unauthenticated_minimal_liveness() -> None: + server, thread, port = _start() try: with urllib.request.urlopen(f"http://127.0.0.1:{port}/healthz", timeout=5) as response: status = response.status @@ -35,7 +42,52 @@ def test_healthz_is_unauthenticated_liveness() -> None: thread.join(timeout=5) assert status == 200 - assert body["status"] == "ok" + assert body == {"status": "ok", "service": "contextual-orchestrator"} + # Operational inventory must not leak on unauthenticated liveness. + for forbidden in ( + "agent_count", + "batch_backend", + "embedding_batch_backend", + "usage_record_count", + "agents", + "ready", + ): + assert forbidden not in body + + +def test_readyz_requires_admin_and_exposes_inventory() -> None: + server, thread, port = _start() + try: + unauth = urllib.request.Request(f"http://127.0.0.1:{port}/readyz") + try: + urllib.request.urlopen(unauth, timeout=5) + raise AssertionError("readyz must require auth") + except urllib.error.HTTPError as exc: + assert exc.code == 401 + + inference = urllib.request.Request( + f"http://127.0.0.1:{port}/readyz", + headers={"authorization": f"Bearer {_TEST_INFERENCE}"}, + ) + try: + urllib.request.urlopen(inference, timeout=5) + raise AssertionError("readyz must not accept inference token") + except urllib.error.HTTPError as exc: + assert exc.code == 401 + + admin = urllib.request.Request( + f"http://127.0.0.1:{port}/readyz", + headers={"authorization": f"Bearer {_TEST_ADMIN}"}, + ) + with urllib.request.urlopen(admin, timeout=5) as response: + status = response.status + body = json.loads(response.read().decode("utf-8")) + finally: + server.shutdown() + thread.join(timeout=5) + + assert status == 200 + assert body["status"] == "ready" assert body["service"] == "contextual-orchestrator" assert body["agent_count"] == 1 assert body["batch_backend"] @@ -44,5 +96,6 @@ def test_healthz_is_unauthenticated_liveness() -> None: if __name__ == "__main__": - test_healthz_is_unauthenticated_liveness() + test_healthz_is_unauthenticated_minimal_liveness() + test_readyz_requires_admin_and_exposes_inventory() print("ok") diff --git a/tests/test_request_framing.py b/tests/test_request_framing.py new file mode 100644 index 000000000..52c4f5ecf --- /dev/null +++ b/tests/test_request_framing.py @@ -0,0 +1,135 @@ +"""Fail-closed inbound HTTP request framing for JSON bodies.""" +from __future__ import annotations + +import json +import socket +import sys +import threading +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from contextual_orchestrator import ModelAgent, TaskOrchestrator # noqa: E402 +from contextual_orchestrator.server import SecurityConfig, build_server # noqa: E402 + +_TOKEN = "secret_token" # noqa: S105 + + +def _start(): + orch = TaskOrchestrator([ModelAgent("general_agent", "mock-generalist", tags=("reasoning",))]) + server = build_server(orch, port=0, security=SecurityConfig(auth_token=_TOKEN, max_body_bytes=64)) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + return server, thread, server.server_address[1] + + +def _raw_post(port: int, extra_headers: list[str], body: bytes) -> tuple[int, bytes]: + request = ( + f"POST /v1/chat/completions HTTP/1.1\r\n" + f"Host: 127.0.0.1:{port}\r\n" + f"Authorization: Bearer {_TOKEN}\r\n" + f"Connection: close\r\n" + + "".join(h + "\r\n" for h in extra_headers) + + "\r\n" + ).encode("ascii") + body + with socket.create_connection(("127.0.0.1", port), timeout=3) as sock: + sock.settimeout(3) + sock.sendall(request) + chunks: list[bytes] = [] + while True: + try: + data = sock.recv(4096) + except TimeoutError: + break + if not data: + break + chunks.append(data) + raw = b"".join(chunks) + status_line = raw.split(b"\r\n", 1)[0] + code = int(status_line.split()[1]) + return code, raw + + +def test_negative_content_length_is_rejected() -> None: + server, thread, port = _start() + try: + code, raw = _raw_post( + port, + ["Content-Type: application/json", "Content-Length: -1"], + b'{"messages":[{"role":"user","content":"x"}]}', + ) + finally: + server.shutdown() + thread.join(timeout=5) + assert code == 400 + assert b"invalid_content_length" in raw or b"invalid_request" in raw or b"request_framing" in raw + + +def test_non_decimal_content_length_is_rejected() -> None: + server, thread, port = _start() + try: + code, raw = _raw_post( + port, + ["Content-Type: application/json", "Content-Length: 12abc"], + b"{}", + ) + finally: + server.shutdown() + thread.join(timeout=5) + assert code == 400 + + +def test_oversized_content_length_is_rejected() -> None: + server, thread, port = _start() + try: + code, raw = _raw_post( + port, + ["Content-Type: application/json", "Content-Length: 65"], + b"x" * 65, + ) + finally: + server.shutdown() + thread.join(timeout=5) + assert code == 413 + assert b"request_too_large" in raw + + +def test_missing_content_length_is_rejected_for_json_post() -> None: + server, thread, port = _start() + try: + code, raw = _raw_post( + port, + ["Content-Type: application/json"], + b'{"messages":[{"role":"user","content":"x"}]}', + ) + finally: + server.shutdown() + thread.join(timeout=5) + assert code == 411 or code == 400 + + +def test_valid_small_json_body_still_works() -> None: + server, thread, port = _start() + body = b'{"messages":[{"role":"user","content":"hi"}]}' + try: + code, raw = _raw_post( + port, + ["Content-Type: application/json", f"Content-Length: {len(body)}"], + body, + ) + finally: + server.shutdown() + thread.join(timeout=5) + assert code == 200 + assert b"chat.completion" in raw or b"choices" in raw + + +if __name__ == "__main__": + test_negative_content_length_is_rejected() + test_non_decimal_content_length_is_rejected() + test_oversized_content_length_is_rejected() + test_missing_content_length_is_rejected_for_json_post() + test_valid_small_json_body_still_works() + print("ok") diff --git a/tests/test_trace_authority.py b/tests/test_trace_authority.py new file mode 100644 index 000000000..2343420f0 --- /dev/null +++ b/tests/test_trace_authority.py @@ -0,0 +1,144 @@ +"""Trace disclosure requires authority beyond ordinary inference access.""" +from __future__ import annotations + +import json +import sys +import threading +import urllib.error +import urllib.request +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from contextual_orchestrator import ModelAgent, TaskOrchestrator # noqa: E402 +from contextual_orchestrator.server import SecurityConfig, build_server # noqa: E402 + +_ADMIN = "admin_secret" # noqa: S105 +_INFERENCE = "inference_secret" # noqa: S105 +_SINGLE = "single_token" # noqa: S105 + + +def post(url: str, payload: dict, token: str) -> tuple[int, dict]: + req = urllib.request.Request( + url, + data=json.dumps(payload).encode("utf-8"), + headers={ + "content-type": "application/json", + "authorization": f"Bearer {token}", + "connection": "close", + }, + method="POST", + ) + try: + with urllib.request.urlopen(req, timeout=5) 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_inference_token_cannot_obtain_orchestration_trace() -> None: + orch = TaskOrchestrator([ModelAgent("general_agent", "mock-generalist", tags=("reasoning", "writing"))]) + server = build_server( + orch, + port=0, + security=SecurityConfig(admin_token=_ADMIN, inference_token=_INFERENCE), + ) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + port = server.server_address[1] + payload = { + "messages": [{"role": "user", "content": "hello"}], + "include_orchestration_trace": True, + } + try: + status, body = post(f"http://127.0.0.1:{port}/v1/chat/completions", payload, _INFERENCE) + finally: + server.shutdown() + thread.join(timeout=5) + assert status == 200 + assert "trace" not in body.get("orchestration", {}) + # String "false" must not become truthy if coercion is ever allowed. + status2, body2 = 0, {} + server2 = build_server( + orch, + port=0, + security=SecurityConfig(admin_token=_ADMIN, inference_token=_INFERENCE), + ) + t2 = threading.Thread(target=server2.serve_forever, daemon=True) + t2.start() + port2 = server2.server_address[1] + try: + status2, body2 = post( + f"http://127.0.0.1:{port2}/v1/chat/completions", + {"messages": [{"role": "user", "content": "hello"}], "include_orchestration_trace": "false"}, + _INFERENCE, + ) + finally: + server2.shutdown() + t2.join(timeout=5) + assert status2 in {200, 400} + if status2 == 200: + assert "trace" not in body2.get("orchestration", {}) + + +def test_admin_token_can_request_orchestration_trace_on_admin_surfaces() -> None: + """Admin simulate path may include traces; inference chat remains answer-only.""" + orch = TaskOrchestrator([ModelAgent("general_agent", "mock-generalist", tags=("reasoning", "writing"))]) + server = build_server( + orch, + port=0, + security=SecurityConfig(admin_token=_ADMIN, inference_token=_INFERENCE, expose_trace_by_default=True), + ) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + port = server.server_address[1] + try: + # Admin simulate is admin-scoped and may carry traces when requested. + status, body = post( + f"http://127.0.0.1:{port}/admin/simulate", + {"prompt": "hello", "mode": "route", "include_orchestration_trace": True}, + _ADMIN, + ) + assert status == 200 + assert "trace" in body.get("orchestration", body) or body.get("mode") is not None + finally: + server.shutdown() + thread.join(timeout=5) + + +def test_single_token_mode_trace_requires_explicit_true_bool() -> None: + orch = TaskOrchestrator([ModelAgent("general_agent", "mock-generalist", tags=("reasoning", "writing"))]) + server = build_server( + orch, + port=0, + security=SecurityConfig(auth_token=_SINGLE, expose_trace_by_default=False), + ) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + port = server.server_address[1] + try: + status, body = post( + f"http://127.0.0.1:{port}/v1/chat/completions", + {"messages": [{"role": "user", "content": "hello"}], "include_orchestration_trace": True}, + _SINGLE, + ) + # Single-token deployments treat the token as both scopes; explicit true + # still requires a verified bool and host policy allow when configured. + assert status == 200 + # Without expose_trace_by_default and without admin-only split, single + # token may disclose only when request bool is true AND policy allows — + # default remain denied when expose_trace_by_default is false unless + # request is authorized for trace. Single token gets answer only by default. + assert "trace" not in body.get("orchestration", {}) + finally: + server.shutdown() + thread.join(timeout=5) + + +if __name__ == "__main__": + test_inference_token_cannot_obtain_orchestration_trace() + test_admin_token_can_request_orchestration_trace_on_admin_surfaces() + test_single_token_mode_trace_requires_explicit_true_bool() + print("ok")