diff --git a/docs/superpowers/plans/2026-08-13-tenant-cloud-routing.md b/docs/superpowers/plans/2026-08-13-tenant-cloud-routing.md new file mode 100644 index 000000000..ded357692 --- /dev/null +++ b/docs/superpowers/plans/2026-08-13-tenant-cloud-routing.md @@ -0,0 +1,97 @@ +# Tenant Cloud Routing Implementation Plan + +> Execute with strict red-green-refactor TDD on the stacked PR based on PR #96. + +**Goal:** Add a tenant-scoped, PostgreSQL-backed provider/model registry, a direct OpenAI-compatible model-group endpoint with deterministic fallback, and Cloud Native deployment evidence. + +**Architecture:** Provider secret values remain in the existing pgcrypto KV. New normalized tenant-routing metadata references tenant-qualified KV names. A request-scoped group executor creates existing `ModelAgent` values and calls the existing `ModelClient`; the Cloud Gateway is a separate importable/runtime module so the current standalone server remains compatible. + +**Stack:** Python standard library, optional psycopg DB extra, PostgreSQL 18/pgcrypto, Docker Compose, Kubernetes manifests, GitHub Actions. + +--- + +## Task 1 — Lock the contracts with failing tests + +**Files:** +- Create `tests/test_tenant_registry.py` +- Create `tests/test_model_group_fallback.py` +- Create `tests/test_cloud_gateway_http.py` +- Create `tests/test_cloud_native_contract.py` + +1. Write tests for secret non-disclosure, rotation, tenant isolation, endpoint ownership, ordering, disablement, and normalized SQL names. +2. Write tests for first-failure/second-success, empty-result fallback, no out-of-group attempt, deterministic evidence, and all-failed behavior. +3. Write HTTP tests for auth, tenant header, admin CRUD, OpenAI completion shape, minimal liveness, database readiness, and web UI secret handling. +4. Write deployment-contract tests for two Compose gateways, two Kubernetes replicas, probes, provider-key isolation, and live-workflow secret names. +5. Run the exact test files and retain the expected import failures as RED evidence. + +## Task 2 — Implement the tenant registry + +**Files:** +- Create `contextual_orchestrator/tenant_registry.py` + +1. Add immutable domain records and stable domain exceptions. +2. Add an in-memory backend with injectable shared state for deterministic tests. +3. Add PostgreSQL schema and CRUD using parameter binding and pgcrypto. +4. Namespace every KV credential by tenant and label. +5. Resolve only enabled, same-tenant group members in deterministic order. +6. Add beginner-readable public docstrings. + +## Task 3 — Implement sequential model-group fallback + +**Files:** +- Create `contextual_orchestrator/model_group.py` + +1. Build request-scoped `ModelAgent` values from resolved endpoint metadata. +2. Call the existing `ModelClient` without copying provider transport logic. +3. Reject empty/non-string output and continue to the next member. +4. Return secret-free attempt evidence and usage. +5. Raise one stable redacted error after complete exhaustion. + +## Task 4 — Add the Cloud Gateway and web control plane + +**Files:** +- Create `contextual_orchestrator/cloud_admin.py` +- Create `contextual_orchestrator/cloud_gateway.py` + +1. Add KV-resolved admin/inference authentication. +2. Add `/livez`, `/readyz`, and authenticated detailed readiness. +3. Add tenant, credential, group, endpoint, and membership JSON routes. +4. Add OpenAI-compatible `/v1/chat/completions` using `model` as the group name. +5. Add a same-origin admin UI that stores no raw token or provider secret. +6. Add bounded bodies, stable errors, no-store headers, and strict validation. + +## Task 5 — Add Cloud Native deployment and bootstrap + +**Files:** +- Create `scripts/bootstrap_tenant_registry.py` +- Create `scripts/verify_live_provider_fallback.py` +- Create `deploy/docker-compose.cloud.yml` +- Create `deploy/kubernetes/namespace.yaml` +- Create `deploy/kubernetes/config-map.yaml` +- Create `deploy/kubernetes/deployment.yaml` +- Create `deploy/kubernetes/service.yaml` +- Create `deploy/kubernetes/network-policy.yaml` +- Create `deploy/kubernetes/pod-disruption-budget.yaml` +- Create `deploy/kubernetes/bootstrap-job.yaml` +- Create `.github/workflows/live-tenant-provider-fallback.yml` + +1. Keep provider keys only in the one-shot bootstrap environment. +2. Prove two gateway processes share one PostgreSQL registry. +3. Add bounded live probes for OpenRouter, NVIDIA NIM, and Bytez. +4. Force the first group member to fail and verify the next valid provider wins. +5. Emit only secret-redacted summaries. + +## Task 6 — Documentation and exact-head verification + +**Files:** +- Create `docs/adr/0011-tenant-provider-registry.md` +- Create `docs/tenant-cloud-routing.md` +- Create `docs/doctoring/tenant-cloud-routing-references.md` +- Update `CHANGELOG.md` + +1. Run the four focused test files. +2. Run the full branch-coverage suite and public-docstring gate. +3. Run deployment-contract and Postgres integration workflows. +4. Run Security, fuzz, SAST, SBOM, and package checks on the same contributor head. +5. Review every automated/human thread and fix every valid finding. +6. Keep the PR Draft until the security prerequisite, exact-head checks, and independent approval are all satisfied. diff --git a/docs/superpowers/specs/2026-08-13-tenant-cloud-routing-design.md b/docs/superpowers/specs/2026-08-13-tenant-cloud-routing-design.md new file mode 100644 index 000000000..9ebc99921 --- /dev/null +++ b/docs/superpowers/specs/2026-08-13-tenant-cloud-routing-design.md @@ -0,0 +1,101 @@ +# Tenant Cloud Routing Design + +**Date:** 2026-08-13 +**Status:** Accepted for the bounded stacked implementation +**Base authority:** PR #96 exact head, not protected `main` + +## Problem + +The current gateway resolves one named provider credential from a process-global KV and keeps its agent pool in process memory. That is insufficient for a Cloud Native deployment where several Docker services or Kubernetes Pods must share one tenant's provider keys and model routing policy without treating one process as configuration authority. + +## Approaches considered + +### A. Keep process-local agent JSON and copy it into every Pod + +Rejected. Rotation, endpoint disablement, and fallback order would drift between replicas, and a rollout would be required for every routing change. + +### B. Add a new general-purpose secret-vault service + +Rejected for this slice. It duplicates the existing pgcrypto credential boundary and would create another identity and availability dependency before a second proven implementation requires it. + +### C. Reuse the encrypted KV for secret values and add normalized tenant routing metadata + +Selected. Provider secrets remain in `provider_credentials`; tenant-qualified credential names prevent collisions. Separate normalized tables own tenant, credential metadata, groups, endpoints, and ordered membership. Every gateway replica resolves the group from PostgreSQL on each request and then uses the existing `ModelClient` provider trust boundary. + +## Identity boundary + +Keyverse/cwl-idp, or an equivalent verified identity proxy, remains the identity authority. The gateway does not create users, passwords, or identity tokens. The direct Cloud Gateway API requires an authenticated admin/inference bearer resolved from the shared KV. The browser UI stores no bearer in local or session storage and assumes same-origin identity-proxy injection in production. + +## Data model + +```mermaid +erDiagram + tenant_records ||--o{ tenant_provider_credentials : owns + tenant_records ||--o{ tenant_model_groups : owns + tenant_records ||--o{ tenant_model_endpoints : owns + tenant_model_groups ||--o{ tenant_group_memberships : orders + tenant_model_endpoints ||--o{ tenant_group_memberships : participates + tenant_provider_credentials ||--o{ tenant_model_endpoints : authenticates +``` + +The secret is not duplicated in tenant metadata. `tenant_provider_credentials.credential_key` points to the pgcrypto-encrypted `provider_credentials.credential_name` row. + +## Request flow + +```mermaid +sequenceDiagram + participant Caller + participant GatewayA + participant Postgres + participant Provider1 + participant Provider2 + + Caller->>GatewayA: POST /v1/chat/completions\nmodel=general_chat\nX-Contextual-Tenant=acme_corporation + GatewayA->>Postgres: resolve enabled group members for tenant + Postgres-->>GatewayA: ordered endpoint + credential_key metadata + GatewayA->>Postgres: decrypt endpoint 1 credential through KV seam + GatewayA->>Provider1: strict OpenAI-compatible request + Provider1--xGatewayA: failure/invalid completion + GatewayA->>Postgres: decrypt endpoint 2 credential through KV seam + GatewayA->>Provider2: strict OpenAI-compatible request + Provider2-->>GatewayA: complete valid response + GatewayA-->>Caller: completion + secret-free routing evidence +``` + +## Fallback semantics + +This slice implements only `sequential_failover`. + +- Membership order is explicit and unique within a group. +- Fallback never leaves the requested tenant or group. +- Disabled tenant, credential, group, endpoint, or membership is excluded. +- A candidate wins only after returning a non-empty complete string through `ModelClient`. +- Attempt evidence contains endpoint/provider/model identifiers and stable outcome codes, never exception text, prompts, responses, or credentials. +- Immediate race and delayed hedge remain owned by issue #102. + +## Cloud Native deployment + +- Gateway replicas are stateless with respect to tenant routing configuration. +- PostgreSQL is the shared control-plane authority. +- `/livez` performs no dependency access. +- `/readyz` performs only a bounded database ping and never calls an LLM. +- Provider keys enter through a one-shot bootstrap Job; gateway Pods do not receive provider API-key environment variables. +- Two Compose gateway services and a Kubernetes Deployment with two replicas exercise the shared-state contract. + +## Verification + +Offline tests are authoritative for merge gating. A separate manually dispatched workflow may consume `OPENROUTER_API_KEY`, `NVIDIA_NIM_API_KEY`, and `BYTEZ_API_KEY` to seed the registry, prove cross-process visibility, probe each official OpenAI-compatible provider surface, and demonstrate fallback after an intentionally failing first endpoint. + +## References — APA 7th + +Dean, J., & Barroso, L. A. (2013). The tail at scale. *Communications of the ACM, 56*(2), 74–80. https://doi.org/10.1145/2408776.2408794 + +Kubernetes Authors. (2026). *Liveness, readiness, and startup probes*. Kubernetes Documentation. https://kubernetes.io/docs/concepts/workloads/pods/probes/ + +PostgreSQL Global Development Group. (2026). *pgcrypto—Cryptographic functions*. PostgreSQL 18 Documentation. https://www.postgresql.org/docs/current/pgcrypto.html + +OpenRouter. (2026). *API reference*. https://openrouter.ai/docs/api_reference/overview + +NVIDIA. (2026). *NIM for large language models API reference*. https://docs.nvidia.com/nim/large-language-models/latest/api-reference.html + +Bytez. (2026). *OpenAI-compatible chat completions*. https://docs.bytez.com/http-reference/oaiCompliant/chatCompletions diff --git a/tests/test_cloud_gateway_http.py b/tests/test_cloud_gateway_http.py new file mode 100644 index 000000000..80ae6be59 --- /dev/null +++ b/tests/test_cloud_gateway_http.py @@ -0,0 +1,231 @@ +"""HTTP contract tests for the Cloud Native tenant gateway.""" + +from __future__ import annotations + +import json +import threading +import urllib.error +import urllib.request + +from contextual_orchestrator.cloud_gateway import CloudGatewaySecurity, build_cloud_gateway +from contextual_orchestrator.credentials import InMemoryCredentialBackend, set_backend +from contextual_orchestrator.model_group import ModelGroupExecutor +from contextual_orchestrator.tenant_registry import InMemoryTenantRegistry + + +class _CloudClient: + def __init__(self) -> None: + self.calls: list[str] = [] + + def chat(self, agent, messages, temperature=0.2): + del messages, temperature + self.calls.append(agent.id) + if agent.id == "openrouter_primary_endpoint": + raise RuntimeError("primary failed") + return "cloud fallback response" + + def take_usage(self): + return {"prompt_tokens": 2, "completion_tokens": 3, "total_tokens": 5} + + +class _CountingRegistry(InMemoryTenantRegistry): + def __init__(self, *args, **kwargs) -> None: + super().__init__(*args, **kwargs) + self.ping_count = 0 + self.ping_result = True + + def ping(self) -> bool: + self.ping_count += 1 + return self.ping_result + + +def _request(base_url: str, path: str, *, method: str = "GET", body=None, token=None, tenant=None): + headers = {"accept": "application/json"} + data = None + if body is not None: + data = json.dumps(body).encode("utf-8") + headers["content-type"] = "application/json" + if token: + headers["authorization"] = f"Bearer {token}" + if tenant: + headers["x-contextual-tenant"] = tenant + request = urllib.request.Request(base_url + path, data=data, headers=headers, method=method) + try: + with urllib.request.urlopen(request, timeout=3) as response: # nosec B310 - loopback test server. + return response.status, dict(response.headers), response.read().decode("utf-8") + except urllib.error.HTTPError as exc: + return exc.code, dict(exc.headers), exc.read().decode("utf-8") + + +def _start_gateway(): + backend = InMemoryCredentialBackend() + backend.set("contextual_admin_token", "admin-secret-token") + backend.set("contextual_inference_token", "inference-secret-token") + set_backend(backend) + registry = _CountingRegistry(credential_backend=backend) + client = _CloudClient() + executor = ModelGroupExecutor(registry, client) + security = CloudGatewaySecurity( + admin_credential_key="contextual_admin_token", + inference_credential_key="contextual_inference_token", + ) + server = build_cloud_gateway(registry, executor, security=security, port=0) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + host, port = server.server_address + return server, thread, registry, client, f"http://{host}:{port}" + + +def teardown_function() -> None: + """Reset the process credential backend after every test.""" + set_backend(None) + + +def test_liveness_is_dependency_free_and_readiness_is_generic() -> None: + server, thread, registry, _, base_url = _start_gateway() + try: + registry.ping_result = False + status, _, raw = _request(base_url, "/livez") + assert status == 200 + assert json.loads(raw) == {"status": "live", "service": "contextual-orchestrator"} + assert registry.ping_count == 0 + + status, _, raw = _request(base_url, "/readyz") + assert status == 503 + assert json.loads(raw) == {"status": "not_ready"} + assert registry.ping_count == 1 + finally: + server.shutdown() + server.server_close() + thread.join(timeout=3) + + +def test_admin_crud_is_shared_and_never_echoes_secret() -> None: + server, thread, _, _, base_url = _start_gateway() + try: + admin = "admin-secret-token" + status, _, _ = _request( + base_url, + "/api/v1/tenants", + method="POST", + token=admin, + body={"tenant_id": "acme_corporation", "display_name": "ACME Corporation"}, + ) + assert status == 201 + + status, headers, raw = _request( + base_url, + "/api/v1/tenants/acme_corporation/provider_credentials", + method="POST", + token=admin, + body={ + "provider_name": "openrouter_provider", + "credential_label": "openrouter_primary_key", + "secret_value": "provider-secret-value", + }, + ) + assert status == 201 + assert headers["Cache-Control"] == "no-store" + assert "provider-secret-value" not in raw + credential = json.loads(raw) + + status, _, raw = _request( + base_url, + "/api/v1/tenants/acme_corporation/provider_credentials", + token=admin, + ) + assert status == 200 + assert "provider-secret-value" not in raw + assert json.loads(raw)["items"][0]["credential_id"] == credential["credential_id"] + + status, _, raw = _request( + base_url, + "/api/v1/tenants/acme_corporation/provider_credentials", + ) + assert status == 401 + assert "acme_corporation" not in raw + finally: + server.shutdown() + server.server_close() + thread.join(timeout=3) + + +def test_openai_model_field_routes_one_tenant_group_with_fallback() -> None: + server, thread, registry, client, base_url = _start_gateway() + try: + registry.create_tenant("acme_corporation", "ACME Corporation") + first_key = registry.register_provider_credential( + "acme_corporation", "openrouter_provider", "openrouter_primary_key", "one" + ) + second_key = registry.register_provider_credential( + "acme_corporation", "nvidia_provider", "nvidia_secondary_key", "two" + ) + group = registry.create_model_group("acme_corporation", "general_chat_group") + first = registry.create_model_endpoint( + "acme_corporation", + "openrouter_primary_endpoint", + "openrouter_provider", + "openrouter-model-id", + "mock://openrouter", + first_key.credential_id, + ) + second = registry.create_model_endpoint( + "acme_corporation", + "nvidia_secondary_endpoint", + "nvidia_provider", + "nvidia-model-id", + "mock://nvidia", + second_key.credential_id, + ) + registry.add_group_membership( + "acme_corporation", group.group_id, first.endpoint_id, fallback_order=10 + ) + registry.add_group_membership( + "acme_corporation", group.group_id, second.endpoint_id, fallback_order=20 + ) + + status, _, raw = _request( + base_url, + "/v1/chat/completions", + method="POST", + token="inference-secret-token", + tenant="acme_corporation", + body={ + "model": "general_chat_group", + "messages": [{"role": "user", "content": "hello"}], + "temperature": 0.1, + }, + ) + payload = json.loads(raw) + assert status == 200 + assert payload["object"] == "chat.completion" + assert payload["model"] == "general_chat_group" + assert payload["choices"][0]["message"] == { + "role": "assistant", + "content": "cloud fallback response", + } + assert payload["contextual_routing"]["served_endpoint_name"] == "nvidia_secondary_endpoint" + assert payload["contextual_routing"]["attempt_count"] == 2 + assert client.calls == ["openrouter_primary_endpoint", "nvidia_secondary_endpoint"] + assert "one" not in raw and "two" not in raw + finally: + server.shutdown() + server.server_close() + thread.join(timeout=3) + + +def test_admin_page_uses_same_origin_without_browser_secret_storage() -> None: + server, thread, _, _, base_url = _start_gateway() + try: + status, headers, raw = _request(base_url, "/admin") + assert status == 200 + assert headers["Content-Type"].startswith("text/html") + assert "Tenant provider registry" in raw + assert "localStorage" not in raw + assert "sessionStorage" not in raw + assert "secret_value" in raw + assert "credentials: 'same-origin'" in raw + finally: + server.shutdown() + server.server_close() + thread.join(timeout=3) diff --git a/tests/test_cloud_native_contract.py b/tests/test_cloud_native_contract.py new file mode 100644 index 000000000..7a6fd40d3 --- /dev/null +++ b/tests/test_cloud_native_contract.py @@ -0,0 +1,77 @@ +"""Static deployment contracts for the Cloud Native tenant gateway.""" + +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +PROVIDER_SECRET_NAMES = tuple( + f"{provider_prefix}_API_KEY" for provider_prefix in ("OPENROUTER", "NVIDIA_NIM", "BYTEZ") +) +DISALLOWED_REVIEW_SECRET = "COPILOT" + "_GITHUB_TOKEN" + + +def test_compose_runs_two_gateways_against_one_database_without_provider_keys() -> None: + compose = (ROOT / "deploy" / "docker-compose.cloud.yml").read_text(encoding="utf-8") + assert "gateway_one:" in compose + assert "gateway_two:" in compose + assert "tenant_postgres:" in compose + assert compose.count("CONTEXTUAL_ORCHESTRATOR_KV_DSN") >= 3 + assert "tenant_bootstrap:" in compose + + gateway_region = compose.split("tenant_bootstrap:", 1)[0] + for secret_name in PROVIDER_SECRET_NAMES: + assert secret_name not in gateway_region + bootstrap_region = compose.split("tenant_bootstrap:", 1)[1] + for secret_name in PROVIDER_SECRET_NAMES: + assert secret_name in bootstrap_region + + +def test_kubernetes_deployment_has_replicas_and_distinct_probes() -> None: + deployment = (ROOT / "deploy" / "kubernetes" / "deployment.yaml").read_text( + encoding="utf-8" + ) + assert "replicas: 2" in deployment + assert "path: /livez" in deployment + assert "path: /readyz" in deployment + assert "startupProbe:" in deployment + for secret_name in PROVIDER_SECRET_NAMES: + assert secret_name not in deployment + assert "runAsNonRoot: true" in deployment + assert "readOnlyRootFilesystem: true" in deployment + + +def test_bootstrap_job_is_the_only_kubernetes_provider_secret_consumer() -> None: + bootstrap = (ROOT / "deploy" / "kubernetes" / "bootstrap-job.yaml").read_text( + encoding="utf-8" + ) + for secret_name in PROVIDER_SECRET_NAMES: + assert f"name: {secret_name}" in bootstrap + assert "secretKeyRef:"in bootstrap + assert "restartPolicy: Never" in bootstrap + assert "python scripts/bootstrap_tenant_registry.py" in bootstrap + + +def test_kubernetes_service_resilience_and_network_contracts_exist() -> None: + service = (ROOT / "deploy" / "kubernetes" / "service.yaml").read_text(encoding="utf-8") + disruption = (ROOT / "deploy" / "kubernetes" / "pod-disruption-budget.yaml").read_text( + encoding="utf-8" + ) + network = (ROOT / "deploy" / "kubernetes" / "network-policy.yaml").read_text( + encoding="utf-8" + ) + assert "kind: Service" in service + assert "minAvailable: 1" in disruption + assert "policyTypes:" in network + assert "Ingress" in network and "Egress" in network + + +def test_live_workflow_uses_only_the_approved_provider_secret_names() -> None: + workflow = (ROOT / ".github" / "workflows" / "live-tenant-provider-fallback.yml").read_text( + encoding="utf-8" + ) + assert "workflow_dispatch:" in workflow + for secret_name in PROVIDER_SECRET_NAMES: + assert f"secrets.{secret_name}" in workflow + assert DISALLOWED_REVIEW_SECRET not in workflow + assert "verify_live_provider_fallback.py" in workflow + assert "contents: read" in workflow diff --git a/tests/test_model_group_fallback.py b/tests/test_model_group_fallback.py new file mode 100644 index 000000000..95c68add4 --- /dev/null +++ b/tests/test_model_group_fallback.py @@ -0,0 +1,172 @@ +"""Behavior tests for deterministic tenant model-group fallback.""" + +from __future__ import annotations + +from dataclasses import dataclass + +import pytest + +from contextual_orchestrator.credentials import InMemoryCredentialBackend, set_backend +from contextual_orchestrator.model_group import ModelGroupExecutor, ModelGroupUnavailable +from contextual_orchestrator.tenant_registry import InMemoryTenantRegistry + + +@dataclass +class _Outcome: + value: object + usage: dict[str, int] | None = None + + +class _FakeClient: + def __init__(self, outcomes: dict[str, _Outcome]) -> None: + self.outcomes = outcomes + self.calls: list[str] = [] + self._usage: dict[str, int] | None = None + + def chat(self, agent, messages, temperature=0.2): + del messages, temperature + self.calls.append(agent.id) + outcome = self.outcomes[agent.id] + self._usage = outcome.usage + if isinstance(outcome.value, BaseException): + raise outcome.value + return outcome.value + + def take_usage(self): + usage = self._usage + self._usage = None + return usage + + +def _configured_registry() -> InMemoryTenantRegistry: + backend = InMemoryCredentialBackend() + set_backend(backend) + registry = InMemoryTenantRegistry(credential_backend=backend) + registry.create_tenant("acme_corporation", "ACME Corporation") + + first_key = registry.register_provider_credential( + "acme_corporation", "openrouter_provider", "openrouter_primary_key", "secret-one" + ) + second_key = registry.register_provider_credential( + "acme_corporation", "nvidia_provider", "nvidia_secondary_key", "secret-two" + ) + outside_key = registry.register_provider_credential( + "acme_corporation", "bytez_provider", "bytez_outside_key", "secret-three" + ) + + group = registry.create_model_group("acme_corporation", "general_chat_group") + first_endpoint = registry.create_model_endpoint( + "acme_corporation", + "openrouter_primary_endpoint", + "openrouter_provider", + "openrouter-model-id", + "mock://openrouter", + first_key.credential_id, + ) + second_endpoint = registry.create_model_endpoint( + "acme_corporation", + "nvidia_secondary_endpoint", + "nvidia_provider", + "nvidia-model-id", + "mock://nvidia", + second_key.credential_id, + ) + registry.create_model_endpoint( + "acme_corporation", + "bytez_outside_endpoint", + "bytez_provider", + "bytez-model-id", + "mock://bytez", + outside_key.credential_id, + ) + registry.add_group_membership( + "acme_corporation", group.group_id, first_endpoint.endpoint_id, fallback_order=10 + ) + registry.add_group_membership( + "acme_corporation", group.group_id, second_endpoint.endpoint_id, fallback_order=20 + ) + return registry + + +def teardown_function() -> None: + """Reset the process credential backend after each test.""" + set_backend(None) + + +def test_first_failure_falls_back_to_second_complete_response() -> None: + registry = _configured_registry() + client = _FakeClient( + { + "openrouter_primary_endpoint": _Outcome(TimeoutError("provider timeout")), + "nvidia_secondary_endpoint": _Outcome( + "verified response", {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5} + ), + "bytez_outside_endpoint": _Outcome("must not run"), + } + ) + + result = ModelGroupExecutor(registry, client).complete( + "acme_corporation", + "general_chat_group", + [{"role": "user", "content": "hello"}], + ) + + assert result.content == "verified response" + assert result.served_endpoint_name == "nvidia_secondary_endpoint" + assert result.served_model == "nvidia-model-id" + assert result.usage == {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5} + assert client.calls == ["openrouter_primary_endpoint", "nvidia_secondary_endpoint"] + assert [attempt.outcome for attempt in result.attempts] == ["failed", "succeeded"] + assert "provider timeout" not in repr(result.attempts) + assert "bytez_outside_endpoint" not in client.calls + + +def test_empty_or_non_string_completion_is_not_a_winner() -> None: + registry = _configured_registry() + for invalid in ("", " ", None, {"content": "not a string"}): + client = _FakeClient( + { + "openrouter_primary_endpoint": _Outcome(invalid), + "nvidia_secondary_endpoint": _Outcome("second endpoint wins"), + "bytez_outside_endpoint": _Outcome("must not run"), + } + ) + result = ModelGroupExecutor(registry, client).complete( + "acme_corporation", + "general_chat_group", + [{"role": "user", "content": "hello"}], + ) + assert result.content == "second endpoint wins" + assert [attempt.error_code for attempt in result.attempts] == [ + "invalid_completion", + None, + ] + + +def test_all_failed_error_is_stable_and_secret_free() -> None: + registry = _configured_registry() + client = _FakeClient( + { + "openrouter_primary_endpoint": _Outcome(RuntimeError("secret-one leaked here")), + "nvidia_secondary_endpoint": _Outcome(RuntimeError("secret-two leaked here")), + "bytez_outside_endpoint": _Outcome("must not run"), + } + ) + + with pytest.raises(ModelGroupUnavailable) as captured: + ModelGroupExecutor(registry, client).complete( + "acme_corporation", + "general_chat_group", + [{"role": "user", "content": "hello"}], + ) + + error = captured.value + assert str(error) == "model group is unavailable" + assert [attempt.endpoint_name for attempt in error.attempts] == [ + "openrouter_primary_endpoint", + "nvidia_secondary_endpoint", + ] + assert all(attempt.error_code == "provider_failed" for attempt in error.attempts) + assert "secret-one" not in repr(error.attempts) + assert "secret-two" not in repr(error.attempts) + assert client.calls == ["openrouter_primary_endpoint", "nvidia_secondary_endpoint"] diff --git a/tests/test_tenant_registry.py b/tests/test_tenant_registry.py new file mode 100644 index 000000000..6aa91fe89 --- /dev/null +++ b/tests/test_tenant_registry.py @@ -0,0 +1,184 @@ +"""Contract tests for tenant-scoped provider and model-group metadata.""" + +from __future__ import annotations + +import re + +import pytest + +from contextual_orchestrator.credentials import ( + InMemoryCredentialBackend, + get_credential, + set_backend, +) +from contextual_orchestrator.tenant_registry import ( + InMemoryTenantRegistry, + TENANT_SCHEMA_SQL, + TenantBoundaryError, + TenantRegistryState, +) + + +def _registry_pair() -> tuple[InMemoryTenantRegistry, InMemoryTenantRegistry]: + backend = InMemoryCredentialBackend() + set_backend(backend) + state = TenantRegistryState() + return ( + InMemoryTenantRegistry(state=state, credential_backend=backend), + InMemoryTenantRegistry(state=state, credential_backend=backend), + ) + + +def teardown_function() -> None: + """Reset the process credential backend after every test.""" + set_backend(None) + + +def test_credential_rotation_is_shared_and_never_disclosed() -> None: + first, second = _registry_pair() + first.create_tenant("acme_corporation", "ACME Corporation") + + created = first.register_provider_credential( + "acme_corporation", + "openrouter_provider", + "openrouter_primary_key", + "first-secret-value", + ) + listed = second.list_provider_credentials("acme_corporation") + + assert listed == [created] + assert "first-secret-value" not in repr(created) + assert "first-secret-value" not in str(created.as_dict()) + assert get_credential(created.credential_key) == "first-secret-value" + + rotated = second.register_provider_credential( + "acme_corporation", + "openrouter_provider", + "openrouter_primary_key", + "second-secret-value", + ) + assert rotated.credential_id == created.credential_id + assert rotated.credential_key == created.credential_key + assert get_credential(rotated.credential_key) == "second-secret-value" + assert first.list_provider_credentials("acme_corporation") == [rotated] + + +def test_cross_tenant_endpoint_reference_fails_closed() -> None: + registry, _ = _registry_pair() + registry.create_tenant("acme_corporation", "ACME Corporation") + registry.create_tenant("beta_corporation", "Beta Corporation") + credential = registry.register_provider_credential( + "acme_corporation", + "nvidia_provider", + "nvidia_primary_key", + "nvidia-secret-value", + ) + + with pytest.raises(TenantBoundaryError, match="tenant-owned credential"): + registry.create_model_endpoint( + "beta_corporation", + "nvidia_secondary_endpoint", + "nvidia_provider", + "discovered-model-id", + "https://integrate.api.nvidia.com/v1", + credential.credential_id, + priority=20, + ) + + +def test_group_resolution_is_ordered_scoped_and_disable_aware() -> None: + registry, _ = _registry_pair() + registry.create_tenant("acme_corporation", "ACME Corporation") + registry.create_tenant("beta_corporation", "Beta Corporation") + openrouter_key = registry.register_provider_credential( + "acme_corporation", + "openrouter_provider", + "openrouter_primary_key", + "openrouter-secret", + ) + nvidia_key = registry.register_provider_credential( + "acme_corporation", + "nvidia_provider", + "nvidia_secondary_key", + "nvidia-secret", + ) + beta_key = registry.register_provider_credential( + "beta_corporation", + "bytez_provider", + "bytez_primary_key", + "bytez-secret", + ) + + group = registry.create_model_group("acme_corporation", "general_chat_group") + first_endpoint = registry.create_model_endpoint( + "acme_corporation", + "openrouter_primary_endpoint", + "openrouter_provider", + "openrouter-model-id", + "https://openrouter.ai/api/v1", + openrouter_key.credential_id, + priority=50, + ) + second_endpoint = registry.create_model_endpoint( + "acme_corporation", + "nvidia_secondary_endpoint", + "nvidia_provider", + "nvidia-model-id", + "https://integrate.api.nvidia.com/v1", + nvidia_key.credential_id, + priority=10, + ) + beta_endpoint = registry.create_model_endpoint( + "beta_corporation", + "bytez_primary_endpoint", + "bytez_provider", + "bytez-model-id", + "https://api.bytez.com/models/v2/openai/v1", + beta_key.credential_id, + ) + + registry.add_group_membership( + "acme_corporation", group.group_id, second_endpoint.endpoint_id, fallback_order=20 + ) + registry.add_group_membership( + "acme_corporation", group.group_id, first_endpoint.endpoint_id, fallback_order=10 + ) + with pytest.raises(TenantBoundaryError, match="same tenant"): + registry.add_group_membership( + "acme_corporation", group.group_id, beta_endpoint.endpoint_id, fallback_order=30 + ) + + resolved = registry.resolve_model_group("acme_corporation", "general_chat_group") + assert [item.endpoint_name for item in resolved] == [ + "openrouter_primary_endpoint", + "nvidia_secondary_endpoint", + ] + assert [item.fallback_order for item in resolved] == [10, 20] + + registry.set_model_endpoint_enabled( + "acme_corporation", first_endpoint.endpoint_id, enabled=False + ) + resolved_after_disable = registry.resolve_model_group( + "acme_corporation", "general_chat_group" + ) + assert [item.endpoint_name for item in resolved_after_disable] == [ + "nvidia_secondary_endpoint" + ] + + +def test_database_contract_is_normalized_and_uses_descriptive_names() -> None: + expected_tables = { + "provider_credentials", + "tenant_records", + "tenant_provider_credentials", + "tenant_model_groups", + "tenant_model_endpoints", + "tenant_group_memberships", + } + found_tables = set(re.findall(r"CREATE TABLE IF NOT EXISTS\s+([a-z_]+)", TENANT_SCHEMA_SQL)) + assert expected_tables <= found_tables + assert all("_" in table_name for table_name in found_tables) + assert "encrypted_value" in TENANT_SCHEMA_SQL + assert "UNIQUE (tenant_id, credential_label)" in TENANT_SCHEMA_SQL + assert "UNIQUE (model_group_id, fallback_order)" in TENANT_SCHEMA_SQL + assert "REFERENCES tenant_records" in TENANT_SCHEMA_SQL