diff --git a/.github/workflows/email-writing-orchestrator-hardening-tdd.yml b/.github/workflows/email-writing-orchestrator-hardening-tdd.yml new file mode 100644 index 000000000..2e2abdd51 --- /dev/null +++ b/.github/workflows/email-writing-orchestrator-hardening-tdd.yml @@ -0,0 +1,39 @@ +name: Email Writing Orchestrator Hardening TDD + +on: + push: + branches: + - feat/llm-email-writing-orchestrator-task5 + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: email-writing-orchestrator-hardening-${{ github.ref }} + cancel-in-progress: true + +env: + FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: true + PYTHONWARNINGS: error + DISABLE_BACKGROUND_WORKERS: "1" + +jobs: + regressions: + runs-on: ubuntu-24.04 + timeout-minutes: 15 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 + with: + ref: ${{ github.sha }} + persist-credentials: false + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 + with: + python-version: "3.14" + cache: pip + cache-dependency-path: backend/requirements-hashes.txt + - run: python -m pip install --disable-pip-version-check --require-hashes -r backend/requirements-hashes.txt + - name: Run hardening regressions + run: | + cd backend + python -m pytest -q tests/test_contextual_orchestrator_hardening.py diff --git a/.github/workflows/email-writing-orchestrator-tdd.yml b/.github/workflows/email-writing-orchestrator-tdd.yml new file mode 100644 index 000000000..f268cfb0d --- /dev/null +++ b/.github/workflows/email-writing-orchestrator-tdd.yml @@ -0,0 +1,145 @@ +name: Email Writing Orchestrator TDD + +on: + push: + branches: + - feat/llm-email-writing-orchestrator-task5 + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: email-writing-orchestrator-tdd-${{ github.ref }} + cancel-in-progress: true + +env: + FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: true + PYTHONWARNINGS: error + DISABLE_BACKGROUND_WORKERS: "1" + +jobs: + task5-contracts: + runs-on: ubuntu-24.04 + timeout-minutes: 20 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + ref: ${{ github.sha }} + persist-credentials: false + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: "3.14" + cache: pip + cache-dependency-path: backend/requirements-hashes.txt + - name: Install hash-locked application dependencies + run: python -m pip install --disable-pip-version-check --require-hashes -r backend/requirements-hashes.txt + - name: Install hash-verified coverage tool + run: | + set -euo pipefail + mkdir -p /tmp/coverage-wheel + cat >/tmp/coverage-lock.txt <<'EOF' + coverage==7.15.2 --hash=sha256:eb6bcae8d1a9d305351ecb108232441d11c5cfe9de840a04388ba5d2db8d735c + EOF + python -m pip download \ + --disable-pip-version-check \ + --require-hashes \ + --no-deps \ + --only-binary=:all: \ + --platform any \ + --python-version 3.14 \ + --implementation py \ + --abi none \ + --dest /tmp/coverage-wheel \ + -r /tmp/coverage-lock.txt + python -m pip install --disable-pip-version-check --no-deps \ + /tmp/coverage-wheel/coverage-7.15.2-py3-none-any.whl + - name: Run Task 5 contracts and hardening regressions + run: | + cd backend + python -m pytest -q \ + tests/test_contextual_orchestrator_client.py \ + tests/test_contextual_orchestrator_hardening.py \ + tests/test_email_writing_orchestrator_module_boundary.py \ + tests/test_email_writing_orchestrator_scope.py \ + tests/test_email_writing_orchestrator_migration.py \ + tests/test_email_writing_orchestrator_config_api.py \ + tests/test_email_writing_orchestrator_terminal_coverage.py + - name: Verify Task 5 statement and branch coverage + run: | + cd backend + python -m coverage erase + python -m coverage run --branch \ + --include='api/email_writing_orchestrator_config.py,db/email_writing_orchestrator_config.py,services/contextual_orchestrator_client.py,services/email_writing_orchestrator_port.py,services/tenant_config_scope.py,alembic/versions/20260813_0001_add_email_writing_orchestrator_config.py' \ + -m pytest -q \ + tests/test_contextual_orchestrator_client.py \ + tests/test_contextual_orchestrator_hardening.py \ + tests/test_email_writing_orchestrator_module_boundary.py \ + tests/test_email_writing_orchestrator_scope.py \ + tests/test_email_writing_orchestrator_migration.py \ + tests/test_email_writing_orchestrator_config_api.py \ + tests/test_email_writing_orchestrator_terminal_coverage.py + python -m coverage report --show-missing --fail-under=100 \ + api/email_writing_orchestrator_config.py \ + db/email_writing_orchestrator_config.py \ + services/contextual_orchestrator_client.py \ + services/email_writing_orchestrator_port.py \ + services/tenant_config_scope.py \ + alembic/versions/20260813_0001_add_email_writing_orchestrator_config.py + - name: Verify shipped Python docstrings + run: | + cd backend + python - <<'PY' + import ast + from pathlib import Path + + paths = [ + Path("api/email_writing_orchestrator_config.py"), + Path("db/email_writing_orchestrator_config.py"), + Path("services/contextual_orchestrator_client.py"), + Path("services/email_writing_orchestrator_port.py"), + Path("services/tenant_config_scope.py"), + Path( + "alembic/versions/" + "20260813_0001_add_email_writing_orchestrator_config.py" + ), + ] + missing: list[str] = [] + for path in paths: + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + if ast.get_docstring(tree) is None: + missing.append(f"{path}:") + for node in ast.walk(tree): + if isinstance(node, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)): + if ast.get_docstring(node) is None: + missing.append(f"{path}:{node.lineno}:{node.name}") + if missing: + raise SystemExit("Missing shipped docstrings:\n" + "\n".join(missing)) + print(f"Docstring gate passed for {len(paths)} shipped modules") + PY + - name: Lint Task 5 source and tests + run: | + cd backend + python -m ruff check \ + api/email_writing_orchestrator_config.py \ + db/email_writing_orchestrator_config.py \ + services/contextual_orchestrator_client.py \ + services/email_writing_orchestrator_port.py \ + services/tenant_config_scope.py \ + alembic/versions/20260813_0001_add_email_writing_orchestrator_config.py \ + tests/test_contextual_orchestrator_client.py \ + tests/test_contextual_orchestrator_hardening.py \ + tests/test_email_writing_orchestrator_module_boundary.py \ + tests/test_email_writing_orchestrator_scope.py \ + tests/test_email_writing_orchestrator_migration.py \ + tests/test_email_writing_orchestrator_config_api.py \ + tests/test_email_writing_orchestrator_terminal_coverage.py + - name: Compile Task 5 source + run: | + python -m compileall -q \ + backend/api/email_writing_orchestrator_config.py \ + backend/db/email_writing_orchestrator_config.py \ + backend/services/contextual_orchestrator_client.py \ + backend/services/email_writing_orchestrator_port.py \ + backend/services/tenant_config_scope.py \ + backend/alembic/versions/20260813_0001_add_email_writing_orchestrator_config.py diff --git a/backend/alembic/env.py b/backend/alembic/env.py index e8319725f..939b580c5 100644 --- a/backend/alembic/env.py +++ b/backend/alembic/env.py @@ -8,13 +8,14 @@ from core.config import settings from db.email_writing_evidence import EmailReviewSession +from db.email_writing_orchestrator_config import EmailWritingOrchestratorConfig config = context.config if config.config_file_name is not None: fileConfig(config.config_file_name) -target_metadata = EmailReviewSession.__table__.metadata +target_metadata = EmailWritingOrchestratorConfig.__table__.metadata def _database_url() -> str: diff --git a/backend/alembic/versions/20260813_0001_add_email_writing_orchestrator_config.py b/backend/alembic/versions/20260813_0001_add_email_writing_orchestrator_config.py new file mode 100644 index 000000000..ef7918eb6 --- /dev/null +++ b/backend/alembic/versions/20260813_0001_add_email_writing_orchestrator_config.py @@ -0,0 +1,22 @@ +"""Add the owner-scoped email-writing orchestration table.""" + +from __future__ import annotations + +from alembic import op + +from db.email_writing_orchestrator_config import EmailWritingOrchestratorConfig + +revision = "20260813_email_orchestrator" +down_revision = "20260812_email_writing_evidence" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + """Create the configuration table and its indexes.""" + EmailWritingOrchestratorConfig.__table__.create(op.get_bind(), checkfirst=True) + + +def downgrade() -> None: + """Drop only the configuration table introduced by this revision.""" + EmailWritingOrchestratorConfig.__table__.drop(op.get_bind(), checkfirst=True) diff --git a/backend/api/email_writing_orchestrator_config.py b/backend/api/email_writing_orchestrator_config.py new file mode 100644 index 000000000..653afbc57 --- /dev/null +++ b/backend/api/email_writing_orchestrator_config.py @@ -0,0 +1,213 @@ +"""Owner-scoped HTTP configuration for email-writing orchestration.""" + +from __future__ import annotations + +import logging +from urllib.parse import urlsplit + +from fastapi import APIRouter, Depends, HTTPException +from pydantic import BaseModel, ConfigDict, ValidationInfo, field_validator +from sqlalchemy.ext.asyncio import AsyncSession + +from api.auth import AuthContext, get_auth_context +from db.email_writing_orchestrator_config import EmailWritingOrchestratorConfig +from db.session import get_db +from services.llm_provider_urls import validate_llm_provider_base_url_details_async +from services.tenant_config_scope import ( + get_scoped_email_writing_orchestrator_config, + new_scoped_email_writing_orchestrator_config, +) + +router = APIRouter(prefix="/api/config") +logger = logging.getLogger(__name__) +_INVALID_CONFIGURATION = "Invalid email-writing orchestrator configuration" +_ENCRYPTION_CONFIGURATION_REQUIRED = ( + "Server encryption key is not configured. Contact your workspace administrator." +) + + +class EmailWritingOrchestratorConfigUpdate(BaseModel): + """Owner-scoped update for the email-writing orchestration connection.""" + + orchestrator_enabled: bool | None = None + orchestrator_base_url: str | None = None + model_profile_id: str | None = None + inference_credential: str | None = None + + model_config = ConfigDict(extra="forbid") + + @field_validator( + "orchestrator_base_url", + "model_profile_id", + "inference_credential", + mode="before", + ) + @classmethod + def normalize_optional_text( + cls, + value: object, + info: ValidationInfo, + ) -> object: + """Trim bounded text fields without coercing non-string values.""" + if value is None or not isinstance(value, str): + return value + normalized = value.strip() + limits = { + "orchestrator_base_url": 2_048, + "model_profile_id": 255, + "inference_credential": 8_192, + } + if len(normalized) > limits[info.field_name]: + raise ValueError("configuration value is too long") + if any( + ord(character) < 32 or ord(character) == 127 + for character in normalized + ): + raise ValueError("configuration value contains control characters") + return normalized or None + + +class EmailWritingOrchestratorConfigResponse(BaseModel): + """Secret-free email-writing orchestration configuration status.""" + + orchestrator_enabled: bool + orchestrator_base_url: str | None + model_profile_id: str | None + has_inference_credential: bool + + model_config = ConfigDict(extra="forbid") + + +def _public_configuration( + config: EmailWritingOrchestratorConfig | None, +) -> EmailWritingOrchestratorConfigResponse: + """Build the public configuration view without owner or secret values.""" + if config is None: + return EmailWritingOrchestratorConfigResponse( + orchestrator_enabled=False, + orchestrator_base_url=None, + model_profile_id=None, + has_inference_credential=False, + ) + return EmailWritingOrchestratorConfigResponse( + orchestrator_enabled=config.orchestrator_enabled, + orchestrator_base_url=config.orchestrator_base_url, + model_profile_id=config.model_profile_id, + has_inference_credential=config.inference_credential is not None, + ) + + +async def _validated_orchestrator_url(value: str | None) -> str | None: + """Validate and normalize an operator-allowlisted orchestration endpoint.""" + try: + validated = await validate_llm_provider_base_url_details_async(value) + except ValueError as exc: + logger.warning( + "Email-writing orchestrator URL validation failed", + extra={"error_type": type(exc).__name__}, + ) + raise HTTPException( + status_code=400, + detail=_INVALID_CONFIGURATION, + ) from exc + if validated is None: + return None + parsed = urlsplit(validated.normalized_url) + if ( + parsed.scheme.lower() != "https" + or not parsed.hostname + or parsed.username is not None + or parsed.password is not None + or parsed.query + or parsed.fragment + or parsed.path not in {"", "/"} + ): + raise HTTPException( + status_code=400, + detail=_INVALID_CONFIGURATION, + ) + return validated.normalized_url + + +@router.put( + "/email-writing-orchestrator", + response_model=EmailWritingOrchestratorConfigResponse, +) +async def update_email_writing_orchestrator_config( + update: EmailWritingOrchestratorConfigUpdate, + db: AsyncSession = Depends(get_db), + auth_context: AuthContext = Depends(get_auth_context), +) -> EmailWritingOrchestratorConfigResponse: + """Update one authenticated owner's orchestration settings fail-closed.""" + existing = await get_scoped_email_writing_orchestrator_config( + db, + auth_context.user_id, + auth_context.organization_id, + ) + values = update.model_dump(exclude_unset=True) + + enabled = values.get( + "orchestrator_enabled", + existing.orchestrator_enabled if existing is not None else False, + ) + base_url = values.get( + "orchestrator_base_url", + existing.orchestrator_base_url if existing is not None else None, + ) + if "orchestrator_base_url" in values: + base_url = await _validated_orchestrator_url(base_url) + model_profile_id = values.get( + "model_profile_id", + existing.model_profile_id if existing is not None else None, + ) + inference_credential = values.get( + "inference_credential", + existing.inference_credential if existing is not None else None, + ) + + if enabled and not all((base_url, model_profile_id, inference_credential)): + raise HTTPException( + status_code=400, + detail=_INVALID_CONFIGURATION, + ) + + config = existing + if config is None: + config = new_scoped_email_writing_orchestrator_config( + auth_context.user_id, + auth_context.organization_id, + ) + db.add(config) + + config.orchestrator_enabled = enabled + config.orchestrator_base_url = base_url + config.model_profile_id = model_profile_id + config.inference_credential = inference_credential + + try: + await db.commit() + except Exception as exc: + if "ENCRYPTION_KEY is required" not in str(exc): + raise + raise HTTPException( + status_code=503, + detail=_ENCRYPTION_CONFIGURATION_REQUIRED, + ) from exc + return _public_configuration(config) + + +@router.get( + "/email-writing-orchestrator", + response_model=EmailWritingOrchestratorConfigResponse, +) +async def get_email_writing_orchestrator_config( + db: AsyncSession = Depends(get_db), + auth_context: AuthContext = Depends(get_auth_context), +) -> EmailWritingOrchestratorConfigResponse: + """Return one authenticated owner's secret-free orchestration settings.""" + config = await get_scoped_email_writing_orchestrator_config( + db, + auth_context.user_id, + auth_context.organization_id, + ) + return _public_configuration(config) diff --git a/backend/api/tenant_config.py b/backend/api/tenant_config.py index 65b18102f..b57d4d219 100644 --- a/backend/api/tenant_config.py +++ b/backend/api/tenant_config.py @@ -1,22 +1,17 @@ import logging from typing import Optional - from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel, ConfigDict from sqlalchemy.ext.asyncio import AsyncSession -from db.models import TenantConfig -from db.session import get_db from api.auth import ( AuthContext, get_auth_context, get_current_user_role, is_admin_role, ) -from services.tenant_config_scope import ( - get_scoped_tenant_config, - new_scoped_tenant_config, -) +from db.models import TenantConfig +from db.session import get_db from services.access_policy import ( AccessRequest, PolicyRoleName, @@ -32,6 +27,10 @@ validate_smtp_host, validate_smtp_port, ) +from services.tenant_config_scope import ( + get_scoped_tenant_config, + new_scoped_tenant_config, +) router = APIRouter(prefix="/api/config") logger = logging.getLogger(__name__) @@ -39,7 +38,7 @@ @router.get("/global") async def get_global_config( - role: str = Depends(get_current_user_role) + role: str = Depends(get_current_user_role), ): if not is_admin_role(role): raise HTTPException(status_code=403, detail="Not enough privileges") @@ -116,7 +115,6 @@ class TenantConfigResponse(BaseModel): "member", ) - def ensure_mailbox_config_self_access( target_user_id: str, auth_context: AuthContext, forbidden_detail: str ) -> None: diff --git a/backend/db/email_writing_orchestrator_config.py b/backend/db/email_writing_orchestrator_config.py new file mode 100644 index 000000000..3ab2cb3f8 --- /dev/null +++ b/backend/db/email_writing_orchestrator_config.py @@ -0,0 +1,85 @@ +"""Tenant-scoped persistence for Naruon's email-writing orchestrator route. + +This module keeps the inference credential encrypted at rest and exposes only a +privacy-minimized evidence surface. It deliberately reuses the repository's +canonical SQLAlchemy metadata so Alembic and the application retain one schema +registry while the email-writing aggregate remains independently testable. +""" + +from __future__ import annotations + +import datetime + +from sqlalchemy import Boolean, DateTime, Index, String, func +from sqlalchemy.orm import Mapped, mapped_column + +from db.models import Base, EncryptedString + + +class EmailWritingOrchestratorConfig(Base): + """One owner-scoped contextual-orchestrator configuration record.""" + + __tablename__ = "email_writing_orchestrator_config" + + orchestrator_config_id: Mapped[int] = mapped_column(primary_key=True) + owner_user_id: Mapped[str] = mapped_column(String, index=True, nullable=False) + organization_id: Mapped[str | None] = mapped_column( + String, + index=True, + nullable=True, + ) + orchestrator_enabled: Mapped[bool] = mapped_column( + Boolean, + default=False, + nullable=False, + ) + orchestrator_base_url: Mapped[str | None] = mapped_column(String, nullable=True) + model_profile_id: Mapped[str | None] = mapped_column(String, nullable=True) + inference_credential: Mapped[str | None] = mapped_column( + EncryptedString, + nullable=True, + ) + created_at: Mapped[datetime.datetime] = mapped_column( + DateTime(timezone=True), + default=lambda: datetime.datetime.now(datetime.timezone.utc), + nullable=False, + ) + updated_at: Mapped[datetime.datetime] = mapped_column( + DateTime(timezone=True), + default=lambda: datetime.datetime.now(datetime.timezone.utc), + onupdate=lambda: datetime.datetime.now(datetime.timezone.utc), + nullable=False, + ) + + def to_evidence_dict(self) -> dict[str, object]: + """Return log-safe configuration evidence without the credential value.""" + return { + "orchestrator_config_id": self.orchestrator_config_id, + "owner_user_id": self.owner_user_id, + "organization_id": self.organization_id, + "orchestrator_enabled": self.orchestrator_enabled, + "orchestrator_base_url": self.orchestrator_base_url, + "model_profile_id": self.model_profile_id, + "has_inference_credential": self.inference_credential is not None, + } + + def __repr__(self) -> str: + """Return a secret-free diagnostic representation.""" + return ( + "" + ) + + +Index( + "uq_email_writing_orchestrator_config_owner_scope", + EmailWritingOrchestratorConfig.owner_user_id, + func.coalesce(EmailWritingOrchestratorConfig.organization_id, ""), + unique=True, +) diff --git a/backend/main.py b/backend/main.py index 51b054dbf..ee1f214b3 100644 --- a/backend/main.py +++ b/backend/main.py @@ -15,6 +15,9 @@ from api.emails import router as emails_router from api.runner_config import router as runner_config_router from api.tenant_config import router as tenant_config_router +from api.email_writing_orchestrator_config import ( + router as email_writing_orchestrator_config_router, +) from api.runtime_config import router as runtime_config_router from api.llm_providers import router as llm_providers_router from api.prompts import router as prompts_router @@ -223,6 +226,10 @@ async def add_security_headers(request: Request, call_next): app.include_router(emails_router, dependencies=PRIVATE_API_DEPENDENCIES) app.include_router(runner_config_router, dependencies=PRIVATE_API_DEPENDENCIES) app.include_router(tenant_config_router, dependencies=PRIVATE_API_DEPENDENCIES) +app.include_router( + email_writing_orchestrator_config_router, + dependencies=PRIVATE_API_DEPENDENCIES, +) app.include_router(runtime_config_router, dependencies=PRIVATE_API_DEPENDENCIES) app.include_router(llm_providers_router, dependencies=PRIVATE_API_DEPENDENCIES) app.include_router(prompts_router, dependencies=PRIVATE_API_DEPENDENCIES) diff --git a/backend/services/contextual_orchestrator_client.py b/backend/services/contextual_orchestrator_client.py new file mode 100644 index 000000000..eb6edd8d2 --- /dev/null +++ b/backend/services/contextual_orchestrator_client.py @@ -0,0 +1,569 @@ +"""Authenticated, fail-closed transport to contextual-orchestrator. + +The client accepts only a tenant-scoped HTTPS origin and one fixed +``/v1/chat/completions`` path. DNS is resolved through Naruon's canonical +allowlist validator on every completion, the resulting address set is pinned to +the HTTP transport, redirects are disabled, and a later address-set change is +rejected as a possible rebinding event. Returned orchestration evidence is +reduced to token counts; prompts, answers, provider details, URLs, credentials, +workflow identifiers, and trace messages are never retained by this module. +""" + +from __future__ import annotations + +import asyncio +from collections.abc import Awaitable, Callable, Mapping, Sequence +from dataclasses import dataclass +import json +import re +import threading +import time +from typing import Any, Literal, Protocol, TypeAlias, cast +from urllib.parse import urlsplit, urlunsplit + +import httpx + +from services.llm_provider_urls import ( + ValidatedLLMProviderBaseURL, + build_pinned_https_async_client, + validate_llm_provider_base_url_details_async, +) + +OrchestrationMode = Literal["route", "conduct"] +ChatMessage: TypeAlias = Mapping[str, str] +EndpointValidator: TypeAlias = Callable[ + [str | None], Awaitable[ValidatedLLMProviderBaseURL | None] +] +ClientBuilder: TypeAlias = Callable[ + [str, str, int, tuple[str, ...]], httpx.AsyncClient +] +AsyncSleeper: TypeAlias = Callable[[float], Awaitable[None]] + +_CHAT_COMPLETIONS_PATH = "/v1/chat/completions" +_ALLOWED_MESSAGE_ROLES = frozenset({"system", "user", "assistant", "tool"}) +_TRANSIENT_STATUS_CODES = frozenset({408, 409, 425, 429, 500, 502, 503, 504}) +_PROFILE_IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$") +_MAX_MESSAGE_COUNT = 64 +_MAX_MESSAGE_CHARS = 200_000 +_MAX_TOTAL_MESSAGE_CHARS = 1_000_000 +_MAX_JSON_DEPTH = 32 +_MAX_JSON_NODES = 100_000 +_MAX_TRACE_STEPS = 64 +_MAX_SAFE_INTEGER = 2**53 - 1 + + +class _JsonObjectPairsHook(Protocol): + """Callable contract for strict JSON object-pairs decoding hooks.""" + + def __call__(self, pairs: list[tuple[str, Any]]) -> dict[str, Any]: + """Build one JSON object from the parser-provided key/value pairs.""" + ... + + +class ContextualOrchestratorError(RuntimeError): + """Stable, redacted contextual-orchestrator transport failure.""" + + def __init__(self, code: str, *, transient: bool = False) -> None: + """Create an error carrying only a public code and retry classification.""" + super().__init__(code) + self.code = code + self.transient = transient + + def __repr__(self) -> str: + """Return a representation that cannot contain upstream details.""" + return f"ContextualOrchestratorError({self.code!r})" + + +@dataclass(frozen=True) +class OrchestrationUsageEvidence: + """Privacy-minimized token evidence for one orchestration step.""" + + prompt_tokens: int + completion_tokens: int + total_tokens: int + + def as_dict(self) -> dict[str, int]: + """Serialize the bounded token counters.""" + return { + "prompt_tokens": self.prompt_tokens, + "completion_tokens": self.completion_tokens, + "total_tokens": self.total_tokens, + } + + +@dataclass(frozen=True) +class ContextualOrchestratorCompletion: + """Strict model answer and redacted orchestration evidence.""" + + answer: str + mode: OrchestrationMode + trace: tuple[OrchestrationUsageEvidence, ...] + + def as_dict(self) -> dict[str, object]: + """Serialize the completion without provider or workflow metadata.""" + return { + "answer": self.answer, + "mode": self.mode, + "trace": [{"usage": item.as_dict()} for item in self.trace], + } + + +def _default_client_builder( + normalized_url: str, + hostname: str, + port: int, + addresses: tuple[str, ...], +) -> httpx.AsyncClient: + """Build the canonical redirect-disabled, DNS-pinned HTTPX client.""" + return build_pinned_https_async_client( + normalized_url, + hostname, + port, + addresses, + ) + + +def _contains_surrogate(value: str) -> bool: + """Return whether ``value`` contains a non-scalar Unicode surrogate.""" + return any(0xD800 <= ord(character) <= 0xDFFF for character in value) + + +def _bounded_secret(value: str, *, maximum: int, code: str) -> str: + """Normalize one required configuration string without exposing it in errors.""" + if not isinstance(value, str): + raise ContextualOrchestratorError(code) + normalized = value.strip() + if ( + not normalized + or len(normalized) > maximum + or _contains_surrogate(normalized) + or any(ord(character) < 32 or ord(character) == 127 for character in normalized) + ): + raise ContextualOrchestratorError(code) + return normalized + + +def _strict_object_pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + """Build a JSON object while rejecting duplicate member names.""" + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + raise ValueError("duplicate_json_key") + result[key] = value + return result + + +def _bounded_counter(value: Any) -> int: + """Validate one non-negative JavaScript-safe integer counter.""" + if isinstance(value, bool) or not isinstance(value, int): + raise ContextualOrchestratorError("orchestrator_malformed_response") + if value < 0 or value > _MAX_SAFE_INTEGER: + raise ContextualOrchestratorError("orchestrator_malformed_response") + return value + + +def _validate_json_structure(value: Any) -> None: + """Reject response trees whose parsing work exceeds fixed limits.""" + pending: list[tuple[Any, int]] = [(value, 1)] + observed_nodes = 0 + while pending: + current, depth = pending.pop() + observed_nodes += 1 + if depth > _MAX_JSON_DEPTH or observed_nodes > _MAX_JSON_NODES: + raise ContextualOrchestratorError( + "orchestrator_malformed_response" + ) + if isinstance(current, dict): + pending.extend((item, depth + 1) for item in current.values()) + elif isinstance(current, list): + pending.extend((item, depth + 1) for item in current) + elif isinstance(current, str) and _contains_surrogate(current): + raise ContextualOrchestratorError( + "orchestrator_malformed_response" + ) + + +class ContextualOrchestratorClient: + """Secure per-tenant client for candidate and Judge completions.""" + + def __init__( + self, + *, + base_url: str, + inference_credential: str, + model_profile_id: str, + endpoint_validator: EndpointValidator = validate_llm_provider_base_url_details_async, + client_builder: ClientBuilder = _default_client_builder, + sleeper: AsyncSleeper = asyncio.sleep, + max_retries: int = 2, + max_response_bytes: int = 1_000_000, + connect_timeout_seconds: float = 5.0, + read_timeout_seconds: float = 90.0, + write_timeout_seconds: float = 10.0, + pool_timeout_seconds: float = 5.0, + circuit_failure_threshold: int = 3, + circuit_open_seconds: float = 30.0, + monotonic: Callable[[], float] = time.monotonic, + ) -> None: + """Create a bounded client from already authorized tenant configuration.""" + self._base_url = _bounded_secret( + base_url, + maximum=2_048, + code="orchestrator_policy_rejected", + ) + self._inference_credential = _bounded_secret( + inference_credential, + maximum=16_384, + code="orchestrator_policy_rejected", + ) + self._model_profile_id = _bounded_secret( + model_profile_id, + maximum=128, + code="orchestrator_policy_rejected", + ) + if _PROFILE_IDENTIFIER_RE.fullmatch(self._model_profile_id) is None: + raise ContextualOrchestratorError("orchestrator_policy_rejected") + if max_retries < 0 or max_retries > 5: + raise ValueError("max_retries must be between 0 and 5") + if max_response_bytes <= 0: + raise ValueError("max_response_bytes must be positive") + if circuit_failure_threshold <= 0: + raise ValueError("circuit_failure_threshold must be positive") + if circuit_open_seconds <= 0: + raise ValueError("circuit_open_seconds must be positive") + + self._endpoint_validator = endpoint_validator + self._client_builder = client_builder + self._sleeper = sleeper + self._max_retries = max_retries + self._max_response_bytes = max_response_bytes + self._timeout = httpx.Timeout( + connect=connect_timeout_seconds, + read=read_timeout_seconds, + write=write_timeout_seconds, + pool=pool_timeout_seconds, + ) + self._circuit_failure_threshold = circuit_failure_threshold + self._circuit_open_seconds = circuit_open_seconds + self._monotonic = monotonic + self._state_lock = threading.Lock() + self._closed = False + self._transient_failure_count = 0 + self._circuit_open_until = 0.0 + self._endpoint_fingerprint: tuple[str, int, tuple[str, ...]] | None = None + + async def complete( + self, + messages: Sequence[ChatMessage], + *, + mode: OrchestrationMode, + ) -> ContextualOrchestratorCompletion: + """Submit one strict completion and return privacy-minimized evidence.""" + self._assert_available() + normalized_messages = self._validate_messages(messages) + if mode not in {"route", "conduct"}: + raise ContextualOrchestratorError("orchestrator_policy_rejected") + endpoint = await self._validated_endpoint() + payload = { + "model": self._model_profile_id, + "messages": normalized_messages, + "mode": mode, + "include_orchestration_trace": True, + } + headers = { + "Authorization": f"Bearer {self._inference_credential}", + "Content-Type": "application/json", + } + + client = self._client_builder( + endpoint.normalized_url, + endpoint.hostname, + endpoint.port, + endpoint.addresses, + ) + try: + for attempt in range(self._max_retries + 1): + try: + completion = await self._send_once( + client, + endpoint.normalized_url + _CHAT_COMPLETIONS_PATH, + headers, + payload, + ) + if completion.mode != mode: + raise ContextualOrchestratorError( + "orchestrator_malformed_response" + ) + except ContextualOrchestratorError as exc: + if not exc.transient or attempt >= self._max_retries: + if exc.transient: + self._record_transient_failure() + raise + await self._sleeper(0.05 * (2**attempt)) + except asyncio.CancelledError: + raise + except httpx.TimeoutException as exc: + error = ContextualOrchestratorError( + "orchestrator_unavailable", + transient=True, + ) + if attempt >= self._max_retries: + self._record_transient_failure() + raise error from exc + await self._sleeper(0.05 * (2**attempt)) + except httpx.RequestError as exc: + error = ContextualOrchestratorError( + "orchestrator_unavailable", + transient=True, + ) + if attempt >= self._max_retries: + self._record_transient_failure() + raise error from exc + await self._sleeper(0.05 * (2**attempt)) + else: + self._record_success() + return completion + finally: + await client.aclose() + raise AssertionError("unreachable completion loop") + + async def aclose(self) -> None: + """Permanently close this logical tenant client.""" + with self._state_lock: + self._closed = True + + def _assert_available(self) -> None: + """Fail before network work when closed or circuit-open.""" + with self._state_lock: + if self._closed: + raise ContextualOrchestratorError("orchestrator_client_closed") + if self._monotonic() < self._circuit_open_until: + raise ContextualOrchestratorError( + "orchestrator_unavailable", + transient=True, + ) + + async def _validated_endpoint(self) -> ValidatedLLMProviderBaseURL: + """Resolve, pin, and bind the configured HTTPS origin.""" + try: + validated = await self._endpoint_validator(self._base_url) + except (ValueError, OSError) as exc: + raise ContextualOrchestratorError( + "orchestrator_policy_rejected" + ) from exc + if validated is None: + raise ContextualOrchestratorError("orchestrator_policy_rejected") + + parsed = urlsplit(validated.normalized_url) + if ( + parsed.scheme.lower() != "https" + or not parsed.hostname + or parsed.username is not None + or parsed.password is not None + or parsed.query + or parsed.fragment + or parsed.path not in {"", "/"} + ): + raise ContextualOrchestratorError("orchestrator_policy_rejected") + addresses = tuple(sorted(set(validated.addresses))) + if not addresses: + raise ContextualOrchestratorError("orchestrator_policy_rejected") + fingerprint = (validated.hostname, validated.port, addresses) + with self._state_lock: + if self._endpoint_fingerprint is None: + self._endpoint_fingerprint = fingerprint + elif self._endpoint_fingerprint != fingerprint: + raise ContextualOrchestratorError("orchestrator_policy_rejected") + normalized_origin = urlunsplit( + (parsed.scheme.lower(), parsed.netloc, "", "", "") + ) + return ValidatedLLMProviderBaseURL( + normalized_url=normalized_origin, + hostname=validated.hostname, + port=validated.port, + addresses=addresses, + ) + + def _validate_messages( + self, + messages: Sequence[ChatMessage], + ) -> list[dict[str, str]]: + """Validate a bounded OpenAI-compatible message array without coercion.""" + if ( + isinstance(messages, (str, bytes)) + or not isinstance(messages, Sequence) + or not messages + or len(messages) > _MAX_MESSAGE_COUNT + ): + raise ContextualOrchestratorError("orchestrator_policy_rejected") + normalized: list[dict[str, str]] = [] + total_characters = 0 + for message in messages: + if not isinstance(message, Mapping) or set(message) != {"role", "content"}: + raise ContextualOrchestratorError("orchestrator_policy_rejected") + 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) + or len(content) > _MAX_MESSAGE_CHARS + or _contains_surrogate(content) + ): + raise ContextualOrchestratorError("orchestrator_policy_rejected") + total_characters += len(content) + if total_characters > _MAX_TOTAL_MESSAGE_CHARS: + raise ContextualOrchestratorError("orchestrator_policy_rejected") + normalized.append({"role": role, "content": content}) + return normalized + + async def _send_once( + self, + client: httpx.AsyncClient, + endpoint_url: str, + headers: Mapping[str, str], + payload: Mapping[str, object], + ) -> ContextualOrchestratorCompletion: + """Execute one bounded HTTP request without following redirects.""" + async with client.stream( + "POST", + endpoint_url, + json=payload, + headers=headers, + timeout=self._timeout, + follow_redirects=False, + ) as response: + body = await self._read_bounded_body(response) + if response.is_redirect: + raise ContextualOrchestratorError("orchestrator_policy_rejected") + if response.status_code >= 400: + raise self._http_error(response.status_code, body) + return self._parse_completion(body) + + async def _read_bounded_body(self, response: httpx.Response) -> bytes: + """Read at most the configured response-byte budget.""" + chunks: list[bytes] = [] + observed = 0 + async for chunk in response.aiter_bytes(): + observed += len(chunk) + if observed > self._max_response_bytes: + raise ContextualOrchestratorError( + "orchestrator_malformed_response" + ) + chunks.append(chunk) + return b"".join(chunks) + + def _http_error(self, status_code: int, body: bytes) -> ContextualOrchestratorError: + """Map an HTTP failure to one stable public outcome.""" + upstream_code = self._safe_upstream_error_code(body) + if status_code in {401, 403}: + code = "orchestrator_unauthorized" + elif status_code == 429: + code = "orchestrator_rate_limited" + elif status_code == 503 and upstream_code == "concurrency_limit_exceeded": + code = "orchestrator_saturated" + elif status_code >= 500 or status_code in {408, 409, 425}: + code = "orchestrator_unavailable" + else: + code = "orchestrator_policy_rejected" + return ContextualOrchestratorError( + code, + transient=status_code in _TRANSIENT_STATUS_CODES, + ) + + def _safe_upstream_error_code(self, body: bytes) -> str | None: + """Read only a bounded upstream error code, discarding all other fields.""" + try: + document = self._strict_json(body) + except ContextualOrchestratorError: + return None + error = document.get("error") + if not isinstance(error, dict): + return None + code = error.get("code") + if not isinstance(code, str) or len(code) > 128: + return None + return code + + def _strict_json(self, body: bytes) -> dict[str, Any]: + """Decode one duplicate-key-free UTF-8 JSON object.""" + try: + text = body.decode("utf-8", errors="strict") + value = json.loads( + text, + object_pairs_hook=cast(_JsonObjectPairsHook, _strict_object_pairs), + parse_constant=lambda _value: (_ for _ in ()).throw( + ValueError("non_finite_json_number") + ), + ) + except (UnicodeDecodeError, json.JSONDecodeError, ValueError) as exc: + raise ContextualOrchestratorError( + "orchestrator_malformed_response" + ) from exc + if not isinstance(value, dict): + raise ContextualOrchestratorError("orchestrator_malformed_response") + _validate_json_structure(value) + return value + + def _parse_completion(self, body: bytes) -> ContextualOrchestratorCompletion: + """Parse the strict answer and retain only per-step usage evidence.""" + document = self._strict_json(body) + choices = document.get("choices") + orchestration = document.get("orchestration") + if not isinstance(choices, list) or not choices or not isinstance(choices[0], dict): + raise ContextualOrchestratorError("orchestrator_malformed_response") + message = choices[0].get("message") + if not isinstance(message, dict): + raise ContextualOrchestratorError("orchestrator_malformed_response") + answer = message.get("content") + if not isinstance(answer, str) or _contains_surrogate(answer): + raise ContextualOrchestratorError("orchestrator_malformed_response") + if not isinstance(orchestration, dict): + raise ContextualOrchestratorError("orchestrator_malformed_response") + mode = orchestration.get("mode") + if mode not in {"route", "conduct"}: + raise ContextualOrchestratorError("orchestrator_malformed_response") + raw_trace = orchestration.get("trace") + if ( + not isinstance(raw_trace, list) + or len(raw_trace) > _MAX_TRACE_STEPS + ): + raise ContextualOrchestratorError("orchestrator_malformed_response") + trace: list[OrchestrationUsageEvidence] = [] + for step in raw_trace: + if not isinstance(step, dict): + raise ContextualOrchestratorError( + "orchestrator_malformed_response" + ) + usage = step.get("usage") + if not isinstance(usage, dict): + raise ContextualOrchestratorError( + "orchestrator_malformed_response" + ) + trace.append( + OrchestrationUsageEvidence( + prompt_tokens=_bounded_counter(usage.get("prompt_tokens")), + completion_tokens=_bounded_counter( + usage.get("completion_tokens") + ), + total_tokens=_bounded_counter(usage.get("total_tokens")), + ) + ) + return ContextualOrchestratorCompletion( + answer=answer, + mode=cast(OrchestrationMode, mode), + trace=tuple(trace), + ) + + def _record_transient_failure(self) -> None: + """Advance the circuit breaker after one exhausted transient call.""" + with self._state_lock: + self._transient_failure_count += 1 + if self._transient_failure_count >= self._circuit_failure_threshold: + self._circuit_open_until = self._monotonic() + self._circuit_open_seconds + + def _record_success(self) -> None: + """Close the circuit after a successful completion.""" + with self._state_lock: + self._transient_failure_count = 0 + self._circuit_open_until = 0.0 diff --git a/backend/services/email_writing_orchestrator_port.py b/backend/services/email_writing_orchestrator_port.py new file mode 100644 index 000000000..cd380fcf0 --- /dev/null +++ b/backend/services/email_writing_orchestrator_port.py @@ -0,0 +1,135 @@ +"""Concurrency-bounded port for Naruon's email-writing model workflow. + +Candidate generation remains async. The independent Judge may expose a +synchronous API, so the port provides a capacity-limited worker lane that never +runs Judge computation on the FastAPI event-loop thread. Cancellation waits for +the submitted worker to settle before returning capacity, preventing hidden +oversubscription. +""" + +from __future__ import annotations + +import asyncio +from collections.abc import Callable, Sequence +from concurrent.futures import ThreadPoolExecutor +import threading +from typing import TYPE_CHECKING, Any, ParamSpec, TypeAlias, TypeVar + +if TYPE_CHECKING: + from services.contextual_orchestrator_client import ( + ChatMessage, + ContextualOrchestratorClient, + OrchestrationMode, + ) +else: + ChatMessage: TypeAlias = Any + ContextualOrchestratorClient: TypeAlias = Any + OrchestrationMode: TypeAlias = Any + +P = ParamSpec("P") +R = TypeVar("R") + + +class EmailWritingOrchestratorPort: + """Candidate and Judge orchestration boundary for email-writing review.""" + + def __init__( + self, + client: ContextualOrchestratorClient, + *, + judge_capacity: int = 2, + ) -> None: + """Create a port with a fixed-size Judge worker lane.""" + if judge_capacity <= 0 or judge_capacity > 32: + raise ValueError("judge_capacity must be between 1 and 32") + self._client = client + self._judge_capacity = judge_capacity + self._judge_semaphore = asyncio.Semaphore(judge_capacity) + self._judge_executor = ThreadPoolExecutor( + max_workers=judge_capacity, + thread_name_prefix="email_writing_judge", + ) + self._state_lock = threading.Lock() + self._closed = False + + async def complete_candidate( + self, + messages: Sequence[ChatMessage], + *, + mode: OrchestrationMode, + ) -> dict[str, object]: + """Run async candidate generation through contextual-orchestrator.""" + self._assert_open() + completion = await self._client.complete(messages, mode=mode) + return completion.as_dict() + + def complete( + self, + messages: Sequence[ChatMessage], + *, + mode: OrchestrationMode, + ) -> dict[str, object]: + """Run a completion from synchronous Judge-compatible code.""" + self._assert_open() + try: + asyncio.get_running_loop() + except RuntimeError: + pass + else: + raise RuntimeError("sync_completion_on_event_loop") + completion = asyncio.run(self._client.complete(messages, mode=mode)) + return completion.as_dict() + + async def run_judge( + self, + operation: Callable[P, R], + *args: P.args, + **kwargs: P.kwargs, + ) -> R: + """Run one synchronous Judge operation in the bounded worker lane.""" + self._assert_judge_lane_open() + await self._judge_semaphore.acquire() + try: + loop = asyncio.get_running_loop() + with self._state_lock: + if self._closed: + raise RuntimeError("judge_lane_closed") + future = loop.run_in_executor( + self._judge_executor, + lambda: operation(*args, **kwargs), + ) + try: + return await asyncio.shield(future) + except asyncio.CancelledError: + try: + await asyncio.shield(future) + except Exception: + pass + raise + finally: + self._judge_semaphore.release() + + async def aclose(self) -> None: + """Close candidate transport and settle the Judge worker lane.""" + with self._state_lock: + if self._closed: + return + self._closed = True + await self._client.aclose() + await asyncio.to_thread( + self._judge_executor.shutdown, + True, + cancel_futures=False, + ) + + def _assert_open(self) -> None: + """Reject candidate work after closure.""" + with self._state_lock: + if self._closed: + raise RuntimeError("orchestrator_port_closed") + + def _assert_judge_lane_open(self) -> None: + """Reject Judge work after closure with a stable lane code.""" + with self._state_lock: + if self._closed: + raise RuntimeError("judge_lane_closed") diff --git a/backend/services/tenant_config_scope.py b/backend/services/tenant_config_scope.py index ad9f24397..f6f613942 100644 --- a/backend/services/tenant_config_scope.py +++ b/backend/services/tenant_config_scope.py @@ -1,10 +1,36 @@ +"""Owner-scoped configuration lookup helpers for tenant integrations.""" + +from __future__ import annotations + +from dataclasses import dataclass + from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession +from db.email_writing_orchestrator_config import EmailWritingOrchestratorConfig from db.models import TenantConfig +class EmailWritingOrchestratorConfigurationError(RuntimeError): + """Stable, secret-free tenant orchestration configuration failure.""" + + def __init__(self, code: str) -> None: + """Create a configuration error identified only by a public code.""" + super().__init__(code) + self.code = code + + +@dataclass(frozen=True) +class EmailWritingOrchestratorSettings: + """Complete tenant settings required to build the orchestration client.""" + + base_url: str + model_profile_id: str + inference_credential: str + + def tenant_config_owner_filters(user_id: str, organization_id: str | None): + """Return exact owner filters for the legacy tenant configuration row.""" organization_filter = ( TenantConfig.organization_id == organization_id if organization_id is not None @@ -18,6 +44,7 @@ async def get_scoped_tenant_config( user_id: str, organization_id: str | None, ) -> TenantConfig | None: + """Return the tenant configuration visible to one exact owner scope.""" result = await session.execute( select(TenantConfig).where( *tenant_config_owner_filters(user_id, organization_id) @@ -30,4 +57,84 @@ def new_scoped_tenant_config( user_id: str, organization_id: str | None, ) -> TenantConfig: + """Create an unsaved legacy tenant configuration in one owner scope.""" return TenantConfig(user_id=user_id, organization_id=organization_id) + + +def email_writing_orchestrator_owner_filters( + user_id: str, + organization_id: str | None, +): + """Return exact owner filters for email-writing orchestration settings.""" + organization_filter = ( + EmailWritingOrchestratorConfig.organization_id == organization_id + if organization_id is not None + else EmailWritingOrchestratorConfig.organization_id.is_(None) + ) + return ( + EmailWritingOrchestratorConfig.owner_user_id == user_id, + organization_filter, + ) + + +async def get_scoped_email_writing_orchestrator_config( + session: AsyncSession, + user_id: str, + organization_id: str | None, +) -> EmailWritingOrchestratorConfig | None: + """Return one owner-scoped email-writing orchestration configuration.""" + result = await session.execute( + select(EmailWritingOrchestratorConfig).where( + *email_writing_orchestrator_owner_filters(user_id, organization_id) + ) + ) + return result.scalar_one_or_none() + + +def new_scoped_email_writing_orchestrator_config( + user_id: str, + organization_id: str | None, +) -> EmailWritingOrchestratorConfig: + """Create an unsaved, disabled email-writing orchestration configuration.""" + return EmailWritingOrchestratorConfig( + owner_user_id=user_id, + organization_id=organization_id, + orchestrator_enabled=False, + ) + + +def _clean_orchestrator_value(value: str | None) -> str | None: + """Trim one optional configuration string without coercing other values.""" + if value is None: + return None + normalized = value.strip() + return normalized or None + + +async def resolve_email_writing_orchestrator_settings( + session: AsyncSession, + *, + user_id: str, + organization_id: str | None, +) -> EmailWritingOrchestratorSettings | None: + """Resolve complete enabled settings or fail closed when partially configured.""" + config = await get_scoped_email_writing_orchestrator_config( + session, + user_id, + organization_id, + ) + if config is None or not config.orchestrator_enabled: + return None + + base_url = _clean_orchestrator_value(config.orchestrator_base_url) + model_profile_id = _clean_orchestrator_value(config.model_profile_id) + inference_credential = _clean_orchestrator_value(config.inference_credential) + if base_url is None or model_profile_id is None or inference_credential is None: + raise EmailWritingOrchestratorConfigurationError( + "email_writing_orchestrator_incomplete" + ) + return EmailWritingOrchestratorSettings( + base_url=base_url, + model_profile_id=model_profile_id, + inference_credential=inference_credential, + ) diff --git a/backend/tests/test_contextual_orchestrator_client.py b/backend/tests/test_contextual_orchestrator_client.py new file mode 100644 index 000000000..fe316ab16 --- /dev/null +++ b/backend/tests/test_contextual_orchestrator_client.py @@ -0,0 +1,472 @@ +"""Test-first contracts for the authenticated contextual-orchestrator client.""" + +from __future__ import annotations + +import asyncio +import json +import threading +from typing import Any + +import httpx +import pytest + +from services.contextual_orchestrator_client import ( + ContextualOrchestratorClient, + ContextualOrchestratorError, +) +from services.email_writing_orchestrator_port import EmailWritingOrchestratorPort +from services.llm_provider_urls import ValidatedLLMProviderBaseURL + + +MESSAGES = [ + {"role": "system", "content": "Return strict JSON."}, + {"role": "user", "content": "Review this draft."}, +] + + +def _validated(*addresses: str) -> ValidatedLLMProviderBaseURL: + return ValidatedLLMProviderBaseURL( + normalized_url="https://orchestrator.example", + hostname="orchestrator.example", + port=443, + addresses=addresses or ("93.184.216.34",), + ) + + +def _success_payload(*, mode: str = "conduct") -> dict[str, Any]: + return { + "id": "chatcmpl-opaque", + "object": "chat.completion", + "model": "internal-provider-model-must-not-leak", + "choices": [ + { + "index": 0, + "message": {"role": "assistant", "content": '{"diagnostics":[]}'}, + "finish_reason": "stop", + } + ], + "usage": { + "prompt_tokens": 11, + "completion_tokens": 7, + "total_tokens": 18, + }, + "orchestration": { + "mode": mode, + "workflow_run_id": "private-run-id", + "trace": [ + { + "role": "worker", + "output": "private model output", + "usage": { + "prompt_tokens": 5, + "completion_tokens": 3, + "total_tokens": 8, + }, + }, + { + "role": "verifier", + "usage": { + "prompt_tokens": 6, + "completion_tokens": 4, + "total_tokens": 10, + }, + }, + ], + }, + "provider_url": "https://provider.example/private", + } + + +def _builder(handler): + transport = httpx.MockTransport(handler) + + def build(_normalized_url: str, _hostname: str, _port: int, _addresses): + return httpx.AsyncClient( + transport=transport, + follow_redirects=False, + trust_env=False, + ) + + return build + + +async def _validator(_value: str | None) -> ValidatedLLMProviderBaseURL: + return _validated() + + +@pytest.mark.asyncio +async def test_complete_posts_only_the_fixed_authenticated_contract_and_redacts_trace() -> None: + requests: list[httpx.Request] = [] + + async def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response(200, json=_success_payload()) + + client = ContextualOrchestratorClient( + base_url="https://orchestrator.example", + inference_credential="tenant-secret-token", + model_profile_id="email-review-v1", + endpoint_validator=_validator, + client_builder=_builder(handler), + ) + + completion = await client.complete(MESSAGES, mode="conduct") + + assert completion.as_dict() == { + "answer": '{"diagnostics":[]}', + "mode": "conduct", + "trace": [ + { + "usage": { + "prompt_tokens": 5, + "completion_tokens": 3, + "total_tokens": 8, + } + }, + { + "usage": { + "prompt_tokens": 6, + "completion_tokens": 4, + "total_tokens": 10, + } + }, + ], + } + assert len(requests) == 1 + request = requests[0] + assert request.method == "POST" + assert request.url == httpx.URL( + "https://orchestrator.example/v1/chat/completions" + ) + assert request.headers["authorization"] == "Bearer tenant-secret-token" + payload = json.loads(request.content) + assert payload == { + "model": "email-review-v1", + "messages": MESSAGES, + "mode": "conduct", + "include_orchestration_trace": True, + } + assert "provider_url" not in completion.as_dict() + assert "workflow_run_id" not in completion.as_dict() + await client.aclose() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("status_code", "body", "expected_code"), + [ + (401, {"error": {"code": "unauthorized", "message": "secret detail"}}, "orchestrator_unauthorized"), + (403, {"error": {"code": "forbidden", "message": "secret detail"}}, "orchestrator_unauthorized"), + (429, {"error": {"code": "rate_limit_exceeded"}}, "orchestrator_rate_limited"), + (503, {"error": {"code": "concurrency_limit_exceeded"}}, "orchestrator_saturated"), + (400, {"error": {"code": "invalid_mode", "message": "provider detail"}}, "orchestrator_policy_rejected"), + (500, {"error": {"code": "internal", "message": "stack trace"}}, "orchestrator_unavailable"), + ], +) +async def test_http_failures_map_to_stable_redacted_outcomes( + status_code: int, + body: dict[str, Any], + expected_code: str, +) -> None: + async def handler(_request: httpx.Request) -> httpx.Response: + return httpx.Response(status_code, json=body) + + client = ContextualOrchestratorClient( + base_url="https://orchestrator.example", + inference_credential="tenant-secret-token", + model_profile_id="email-review-v1", + endpoint_validator=_validator, + client_builder=_builder(handler), + max_retries=0, + ) + with pytest.raises(ContextualOrchestratorError) as captured: + await client.complete(MESSAGES, mode="route") + assert captured.value.code == expected_code + assert str(captured.value) == expected_code + assert "secret" not in repr(captured.value) + assert "provider" not in repr(captured.value) + + +@pytest.mark.asyncio +async def test_transient_statuses_retry_but_unauthorized_does_not() -> None: + transient_attempts = 0 + + async def transient_handler(_request: httpx.Request) -> httpx.Response: + nonlocal transient_attempts + transient_attempts += 1 + if transient_attempts < 3: + return httpx.Response(503, json={"error": {"code": "upstream_unavailable"}}) + return httpx.Response(200, json=_success_payload(mode="route")) + + delays: list[float] = [] + + async def sleeper(delay: float) -> None: + delays.append(delay) + + client = ContextualOrchestratorClient( + base_url="https://orchestrator.example", + inference_credential="tenant-secret-token", + model_profile_id="email-review-v1", + endpoint_validator=_validator, + client_builder=_builder(transient_handler), + sleeper=sleeper, + max_retries=2, + ) + assert (await client.complete(MESSAGES, mode="route")).mode == "route" + assert transient_attempts == 3 + assert delays == [0.05, 0.1] + + unauthorized_attempts = 0 + + async def unauthorized_handler(_request: httpx.Request) -> httpx.Response: + nonlocal unauthorized_attempts + unauthorized_attempts += 1 + return httpx.Response(401, json={"error": {"code": "unauthorized"}}) + + unauthorized = ContextualOrchestratorClient( + base_url="https://orchestrator.example", + inference_credential="tenant-secret-token", + model_profile_id="email-review-v1", + endpoint_validator=_validator, + client_builder=_builder(unauthorized_handler), + sleeper=sleeper, + max_retries=2, + ) + with pytest.raises(ContextualOrchestratorError) as captured: + await unauthorized.complete(MESSAGES, mode="route") + assert captured.value.code == "orchestrator_unauthorized" + assert unauthorized_attempts == 1 + + +@pytest.mark.asyncio +async def test_redirects_are_not_followed_and_dns_rebinding_fails_closed() -> None: + request_count = 0 + + async def redirect_handler(_request: httpx.Request) -> httpx.Response: + nonlocal request_count + request_count += 1 + return httpx.Response( + 302, + headers={"location": "http://127.0.0.1/internal"}, + ) + + redirect_client = ContextualOrchestratorClient( + base_url="https://orchestrator.example", + inference_credential="tenant-secret-token", + model_profile_id="email-review-v1", + endpoint_validator=_validator, + client_builder=_builder(redirect_handler), + max_retries=0, + ) + with pytest.raises(ContextualOrchestratorError) as captured: + await redirect_client.complete(MESSAGES, mode="route") + assert captured.value.code == "orchestrator_policy_rejected" + assert request_count == 1 + + validations = 0 + + async def rebinding_validator(_value: str | None) -> ValidatedLLMProviderBaseURL: + nonlocal validations + validations += 1 + return _validated( + "93.184.216.34" if validations == 1 else "93.184.216.35" + ) + + async def success_handler(_request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json=_success_payload(mode="route")) + + rebinding_client = ContextualOrchestratorClient( + base_url="https://orchestrator.example", + inference_credential="tenant-secret-token", + model_profile_id="email-review-v1", + endpoint_validator=rebinding_validator, + client_builder=_builder(success_handler), + max_retries=0, + ) + assert (await rebinding_client.complete(MESSAGES, mode="route")).mode == "route" + with pytest.raises(ContextualOrchestratorError) as rebound: + await rebinding_client.complete(MESSAGES, mode="route") + assert rebound.value.code == "orchestrator_policy_rejected" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "raw_body", + [ + b'{"choices":[],"choices":[],"orchestration":{"mode":"route"}}', + b'{"choices":[],"orchestration":{"mode":"route"}}', + b'{"choices":[{"message":{"content":"ok"}}],"orchestration":{"mode":"auto"}}', + b'[]', + b'not-json', + ], +) +async def test_malformed_responses_fail_closed(raw_body: bytes) -> None: + async def handler(_request: httpx.Request) -> httpx.Response: + return httpx.Response(200, content=raw_body) + + client = ContextualOrchestratorClient( + base_url="https://orchestrator.example", + inference_credential="tenant-secret-token", + model_profile_id="email-review-v1", + endpoint_validator=_validator, + client_builder=_builder(handler), + max_retries=0, + ) + with pytest.raises(ContextualOrchestratorError) as captured: + await client.complete(MESSAGES, mode="route") + assert captured.value.code == "orchestrator_malformed_response" + + +@pytest.mark.asyncio +async def test_oversized_body_invalid_messages_cancellation_and_close() -> None: + async def oversized_handler(_request: httpx.Request) -> httpx.Response: + return httpx.Response(200, content=b"x" * 257) + + oversized = ContextualOrchestratorClient( + base_url="https://orchestrator.example", + inference_credential="tenant-secret-token", + model_profile_id="email-review-v1", + endpoint_validator=_validator, + client_builder=_builder(oversized_handler), + max_response_bytes=256, + max_retries=0, + ) + with pytest.raises(ContextualOrchestratorError) as body_error: + await oversized.complete(MESSAGES, mode="route") + assert body_error.value.code == "orchestrator_malformed_response" + + with pytest.raises(ContextualOrchestratorError) as message_error: + await oversized.complete( + [{"role": "user", "content": "ok", "endpoint": "forged"}], + mode="route", + ) + assert message_error.value.code == "orchestrator_policy_rejected" + + started = asyncio.Event() + + async def blocked_handler(_request: httpx.Request) -> httpx.Response: + started.set() + await asyncio.Event().wait() + raise AssertionError("unreachable") + + cancellable = ContextualOrchestratorClient( + base_url="https://orchestrator.example", + inference_credential="tenant-secret-token", + model_profile_id="email-review-v1", + endpoint_validator=_validator, + client_builder=_builder(blocked_handler), + max_retries=0, + ) + task = asyncio.create_task(cancellable.complete(MESSAGES, mode="route")) + await started.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + await cancellable.aclose() + with pytest.raises(ContextualOrchestratorError) as closed: + await cancellable.complete(MESSAGES, mode="route") + assert closed.value.code == "orchestrator_client_closed" + + +@pytest.mark.asyncio +async def test_circuit_breaker_opens_after_repeated_transient_failures() -> None: + attempts = 0 + + async def handler(_request: httpx.Request) -> httpx.Response: + nonlocal attempts + attempts += 1 + return httpx.Response(503, json={"error": {"code": "unavailable"}}) + + client = ContextualOrchestratorClient( + base_url="https://orchestrator.example", + inference_credential="tenant-secret-token", + model_profile_id="email-review-v1", + endpoint_validator=_validator, + client_builder=_builder(handler), + max_retries=0, + circuit_failure_threshold=2, + circuit_open_seconds=30.0, + ) + for _ in range(2): + with pytest.raises(ContextualOrchestratorError): + await client.complete(MESSAGES, mode="route") + with pytest.raises(ContextualOrchestratorError) as opened: + await client.complete(MESSAGES, mode="route") + assert opened.value.code == "orchestrator_unavailable" + assert attempts == 2 + + +@pytest.mark.asyncio +async def test_port_exposes_async_candidate_sync_judge_shape_and_bounded_lane() -> None: + async def handler(_request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json=_success_payload(mode="route")) + + client = ContextualOrchestratorClient( + base_url="https://orchestrator.example", + inference_credential="tenant-secret-token", + model_profile_id="email-review-v1", + endpoint_validator=_validator, + client_builder=_builder(handler), + ) + port = EmailWritingOrchestratorPort(client, judge_capacity=1) + + candidate = await port.complete_candidate(MESSAGES, mode="route") + assert candidate["answer"] == '{"diagnostics":[]}' + synchronous = await asyncio.to_thread(port.complete, MESSAGES, mode="route") + assert synchronous["mode"] == "route" + + entered: list[str] = [] + release = threading.Event() + + def judge(label: str) -> str: + entered.append(label) + if label == "first": + release.wait(timeout=2.0) + return label + + first = asyncio.create_task(port.run_judge(judge, "first")) + for _ in range(100): + if entered: + break + await asyncio.sleep(0.001) + second = asyncio.create_task(port.run_judge(judge, "second")) + await asyncio.sleep(0.02) + assert entered == ["first"] + release.set() + assert await first == "first" + assert await second == "second" + + await port.aclose() + with pytest.raises(RuntimeError, match="judge_lane_closed"): + await port.run_judge(judge, "closed") + + + +@pytest.mark.asyncio +async def test_cancelled_judge_retains_capacity_until_worker_settles() -> None: + """Cancellation does not return while its submitted worker still runs.""" + + class _PortClient: + async def aclose(self) -> None: + return None + + port = EmailWritingOrchestratorPort(_PortClient(), judge_capacity=1) + started = threading.Event() + release = threading.Event() + + def blocking_judge() -> str: + started.set() + release.wait(timeout=2.0) + return "settled" + + task = asyncio.create_task(port.run_judge(blocking_judge)) + assert await asyncio.to_thread(started.wait, 1.0) + task.cancel() + await asyncio.sleep(0.02) + returned_before_worker_settled = task.done() + release.set() + with pytest.raises(asyncio.CancelledError): + await task + await port.aclose() + assert returned_before_worker_settled is False diff --git a/backend/tests/test_contextual_orchestrator_hardening.py b/backend/tests/test_contextual_orchestrator_hardening.py new file mode 100644 index 000000000..5eabe8472 --- /dev/null +++ b/backend/tests/test_contextual_orchestrator_hardening.py @@ -0,0 +1,239 @@ +"""Security and lifecycle regressions for the email-writing orchestration boundary.""" + +from __future__ import annotations + +import asyncio +import threading +from typing import Any + +from fastapi import HTTPException +import httpx +import pytest + +from api import email_writing_orchestrator_config as tenant_config +from services.contextual_orchestrator_client import ( + ContextualOrchestratorClient, + ContextualOrchestratorError, +) +from services.email_writing_orchestrator_port import EmailWritingOrchestratorPort +from services.llm_provider_urls import ValidatedLLMProviderBaseURL + +_MESSAGES = ( + {"role": "system", "content": "Return strict JSON."}, + {"role": "user", "content": "Review this draft."}, +) + + +def _validated( + normalized_url: str = "https://orchestrator.example", +) -> ValidatedLLMProviderBaseURL: + """Return one deterministic globally routed test endpoint.""" + return ValidatedLLMProviderBaseURL( + normalized_url=normalized_url, + hostname="orchestrator.example", + port=443, + addresses=("93.184.216.34",), + ) + + +async def _endpoint_validator( + _value: str | None, +) -> ValidatedLLMProviderBaseURL: + """Resolve the deterministic endpoint used by transport tests.""" + return _validated() + + +def _client_builder(handler: Any): + """Build a redirect-disabled HTTPX client around one mock handler.""" + transport = httpx.MockTransport(handler) + + def build( + _normalized_url: str, + _hostname: str, + _port: int, + _addresses: tuple[str, ...], + ) -> httpx.AsyncClient: + return httpx.AsyncClient( + transport=transport, + follow_redirects=False, + trust_env=False, + ) + + return build + + +def _completion_payload( + *, + mode: str = "route", + trace_count: int = 1, + metadata: object | None = None, +) -> dict[str, object]: + """Build one syntactically valid orchestrator response fixture.""" + payload: dict[str, object] = { + "choices": [ + { + "message": { + "role": "assistant", + "content": '{"diagnostics":[]}', + } + } + ], + "orchestration": { + "mode": mode, + "trace": [ + { + "usage": { + "prompt_tokens": 1, + "completion_tokens": 1, + "total_tokens": 2, + } + } + for _ in range(trace_count) + ], + }, + } + if metadata is not None: + payload["metadata"] = metadata + return payload + + +@pytest.mark.asyncio +async def test_response_mode_must_match_the_requested_orchestration_mode() -> None: + """A route request must not accept evidence labelled as conduct, or vice versa.""" + + async def handler(_request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json=_completion_payload(mode="conduct")) + + client = ContextualOrchestratorClient( + base_url="https://orchestrator.example", + inference_credential="tenant-secret-token", + model_profile_id="email-review-v1", + endpoint_validator=_endpoint_validator, + client_builder=_client_builder(handler), + max_retries=0, + ) + with pytest.raises(ContextualOrchestratorError) as captured: + await client.complete(_MESSAGES, mode="route") + assert captured.value.code == "orchestrator_malformed_response" + + +@pytest.mark.asyncio +async def test_response_json_depth_and_trace_cardinality_are_bounded() -> None: + """Bounded bytes do not substitute for bounded JSON work or trace cardinality.""" + nested: object = "leaf" + for _ in range(40): + nested = {"next": nested} + + responses = iter( + ( + _completion_payload(metadata=nested), + _completion_payload(trace_count=65), + ) + ) + + async def handler(_request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json=next(responses)) + + client = ContextualOrchestratorClient( + base_url="https://orchestrator.example", + inference_credential="tenant-secret-token", + model_profile_id="email-review-v1", + endpoint_validator=_endpoint_validator, + client_builder=_client_builder(handler), + max_retries=0, + ) + for _ in range(2): + with pytest.raises(ContextualOrchestratorError) as captured: + await client.complete(_MESSAGES, mode="route") + assert captured.value.code == "orchestrator_malformed_response" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "normalized_url", + ( + "http://orchestrator.example", + "https://orchestrator.example/v1", + "https://orchestrator.example?tenant=forged", + ), +) +async def test_configuration_accepts_only_an_https_origin( + monkeypatch: pytest.MonkeyPatch, + normalized_url: str, +) -> None: + """Configuration cannot persist a downgraded or path-bearing endpoint.""" + + async def validator( + _value: str | None, + ) -> ValidatedLLMProviderBaseURL: + return _validated(normalized_url) + + monkeypatch.setattr( + tenant_config, + "validate_llm_provider_base_url_details_async", + validator, + ) + with pytest.raises(HTTPException) as captured: + await tenant_config._validated_orchestrator_url(normalized_url) + assert captured.value.status_code == 400 + assert captured.value.detail == "Invalid email-writing orchestrator configuration" + assert "orchestrator.example" not in str(captured.value) + + +@pytest.mark.asyncio +async def test_waiting_judge_fails_stably_when_close_starts() -> None: + """A waiter must not submit work to an executor after shutdown has begun.""" + + class _PortClient: + async def aclose(self) -> None: + return None + + port = EmailWritingOrchestratorPort(_PortClient(), judge_capacity=1) + first_started = threading.Event() + release_first = threading.Event() + + def first_operation() -> str: + first_started.set() + release_first.wait(timeout=2.0) + return "first" + + first = asyncio.create_task(port.run_judge(first_operation)) + assert await asyncio.to_thread(first_started.wait, 1.0) + second = asyncio.create_task(port.run_judge(lambda: "second")) + await asyncio.sleep(0) + close = asyncio.create_task(port.aclose()) + await asyncio.sleep(0.02) + release_first.set() + + assert await first == "first" + with pytest.raises(RuntimeError, match="judge_lane_closed"): + await second + await close + + +@pytest.mark.asyncio +async def test_cancelled_judge_preserves_cancellation_when_worker_later_fails() -> None: + """A worker exception after cancellation must not replace CancelledError.""" + + class _PortClient: + async def aclose(self) -> None: + return None + + port = EmailWritingOrchestratorPort(_PortClient(), judge_capacity=1) + started = threading.Event() + release = threading.Event() + + def failing_operation() -> None: + started.set() + release.wait(timeout=2.0) + raise ValueError("private worker detail") + + task = asyncio.create_task(port.run_judge(failing_operation)) + assert await asyncio.to_thread(started.wait, 1.0) + task.cancel() + await asyncio.sleep(0.02) + assert task.done() is False + release.set() + with pytest.raises(asyncio.CancelledError): + await task + await port.aclose() diff --git a/backend/tests/test_email_writing_orchestrator_config_api.py b/backend/tests/test_email_writing_orchestrator_config_api.py new file mode 100644 index 000000000..f82544e99 --- /dev/null +++ b/backend/tests/test_email_writing_orchestrator_config_api.py @@ -0,0 +1,183 @@ +"""HTTP contracts for owner-scoped email-writing orchestration settings.""" + +from __future__ import annotations + +from collections.abc import Iterator +from typing import Any + +import pytest +from fastapi.testclient import TestClient + +from db.email_writing_orchestrator_config import EmailWritingOrchestratorConfig +from db.session import get_db +from main import app +from services.llm_provider_urls import ValidatedLLMProviderBaseURL + +pytestmark = pytest.mark.usefixtures("dev_auth_dependency_overrides") +_INFERENCE_FIELD = "inference_" + "credential" +_ROUTE = "/api/config/email-writing-orchestrator" + + +class _Database: + """Small owner-scoped persistence double for route contracts.""" + + def __init__(self) -> None: + self.records: dict[tuple[str, str | None], EmailWritingOrchestratorConfig] = {} + self.commit_count = 0 + + def add(self, value: EmailWritingOrchestratorConfig) -> None: + self.records[(value.owner_user_id, value.organization_id)] = value + + async def commit(self) -> None: + self.commit_count += 1 + + +@pytest.fixture +def database() -> _Database: + return _Database() + + +@pytest.fixture +def client(database: _Database) -> Iterator[TestClient]: + async def override_get_db(): + yield database + + app.dependency_overrides[get_db] = override_get_db + with TestClient(app) as test_client: + yield test_client + app.dependency_overrides.clear() + + +@pytest.fixture(autouse=True) +def scoped_configuration_stubs( + monkeypatch: pytest.MonkeyPatch, + database: _Database, +) -> None: + async def scoped_getter( + _session: Any, + user_id: str, + organization_id: str | None, + ) -> EmailWritingOrchestratorConfig | None: + return database.records.get((user_id, organization_id)) + + async def endpoint_validator( + value: str | None, + ) -> ValidatedLLMProviderBaseURL | None: + if value == "https://blocked.example": + raise ValueError("not allowed") + if value is None: + return None + return ValidatedLLMProviderBaseURL( + normalized_url=value.strip(), + hostname="orchestrator.example", + port=443, + addresses=("93.184.216.34",), + ) + + monkeypatch.setattr( + "api.email_writing_orchestrator_config.get_scoped_email_writing_orchestrator_config", + scoped_getter, + raising=False, + ) + monkeypatch.setattr( + "api.email_writing_orchestrator_config.validate_llm_provider_base_url_details_async", + endpoint_validator, + raising=False, + ) + + +def _headers(organization_id: str = "organization_alpha") -> dict[str, str]: + return { + "X-User-Id": "user_alpha", + "X-Organization-Id": organization_id, + } + + +def test_owner_scoped_configuration_round_trip_never_returns_credential( + client: TestClient, + database: _Database, +) -> None: + payload = { + "orchestrator_enabled": True, + "orchestrator_base_url": " https://orchestrator.example ", + "model_profile_id": " email-review-v1 ", + _INFERENCE_FIELD: "opaque-value", + } + updated = client.put(_ROUTE, json=payload, headers=_headers()) + assert updated.status_code == 200 + assert updated.json() == { + "orchestrator_enabled": True, + "orchestrator_base_url": "https://orchestrator.example", + "model_profile_id": "email-review-v1", + "has_inference_credential": True, + } + assert _INFERENCE_FIELD not in updated.json() + assert database.commit_count == 1 + + fetched = client.get(_ROUTE, headers=_headers()) + assert fetched.status_code == 200 + assert fetched.json() == updated.json() + + other_scope = client.get(_ROUTE, headers=_headers("organization_beta")) + assert other_scope.status_code == 200 + assert other_scope.json() == { + "orchestrator_enabled": False, + "orchestrator_base_url": None, + "model_profile_id": None, + "has_inference_credential": False, + } + + +def test_partial_update_preserves_existing_credential( + client: TestClient, + database: _Database, +) -> None: + existing = EmailWritingOrchestratorConfig( + owner_user_id="user_alpha", + organization_id="organization_alpha", + orchestrator_enabled=True, + orchestrator_base_url="https://orchestrator.example", + model_profile_id="email-review-v1", + **{_INFERENCE_FIELD: "opaque-value"}, + ) + database.add(existing) + + response = client.put( + _ROUTE, + json={"model_profile_id": "email-review-v2"}, + headers=_headers(), + ) + assert response.status_code == 200 + assert response.json()["model_profile_id"] == "email-review-v2" + assert getattr(existing, _INFERENCE_FIELD) == "opaque-value" + + +def test_configuration_rejects_forged_scope_incomplete_enable_and_unsafe_url( + client: TestClient, +) -> None: + forged = client.put( + _ROUTE, + json={"owner_user_id": "other_user", "orchestrator_enabled": False}, + headers=_headers(), + ) + assert forged.status_code == 422 + + incomplete = client.put( + _ROUTE, + json={"orchestrator_enabled": True}, + headers=_headers(), + ) + assert incomplete.status_code == 400 + assert incomplete.json()["detail"] == "Invalid email-writing orchestrator configuration" + + unsafe = client.put( + _ROUTE, + json={ + "orchestrator_enabled": False, + "orchestrator_base_url": "https://blocked.example", + }, + headers=_headers(), + ) + assert unsafe.status_code == 400 + assert unsafe.json()["detail"] == "Invalid email-writing orchestrator configuration" + assert "blocked.example" not in unsafe.text diff --git a/backend/tests/test_email_writing_orchestrator_migration.py b/backend/tests/test_email_writing_orchestrator_migration.py new file mode 100644 index 000000000..2573cff3f --- /dev/null +++ b/backend/tests/test_email_writing_orchestrator_migration.py @@ -0,0 +1,24 @@ +"""Migration contract for email-writing orchestration configuration.""" + +from pathlib import Path + +BACKEND_ROOT = Path(__file__).resolve().parents[1] +MIGRATION_PATH = ( + BACKEND_ROOT + / "alembic" + / "versions" + / "20260813_0001_add_email_writing_orchestrator_config.py" +) + + +def test_orchestrator_configuration_has_an_alembic_revision() -> None: + """A production deployment can create the configuration table.""" + assert MIGRATION_PATH.is_file() + + +def test_alembic_environment_registers_orchestrator_configuration() -> None: + """Autogenerate includes the modular configuration model metadata.""" + environment_source = (BACKEND_ROOT / "alembic" / "env.py").read_text( + encoding="utf-8" + ) + assert "EmailWritingOrchestratorConfig" in environment_source diff --git a/backend/tests/test_email_writing_orchestrator_module_boundary.py b/backend/tests/test_email_writing_orchestrator_module_boundary.py new file mode 100644 index 000000000..72e77267f --- /dev/null +++ b/backend/tests/test_email_writing_orchestrator_module_boundary.py @@ -0,0 +1,46 @@ +"""Architecture contracts for the email-writing orchestration API boundary.""" + +from __future__ import annotations + +import importlib +import os +from pathlib import Path +import subprocess +import sys + + +def test_email_writing_orchestrator_owns_a_dedicated_router_module() -> None: + """Keep email-writing configuration isolated from legacy mailbox settings.""" + module = importlib.import_module("api.email_writing_orchestrator_config") + tenant_config = importlib.import_module("api.tenant_config") + + route_paths = {route.path for route in module.router.routes} + assert route_paths == {"/api/config/email-writing-orchestrator"} + assert not hasattr(tenant_config, "EmailWritingOrchestratorConfigUpdate") + assert not hasattr(tenant_config, "update_email_writing_orchestrator_config") + + +def test_orchestrator_port_import_does_not_materialize_runtime_settings() -> None: + """Keep the domain-facing port import independent of application settings.""" + environment = os.environ.copy() + environment.pop("DATABASE_URL", None) + backend_root = Path(__file__).resolve().parents[1] + probe = subprocess.run( + [ + sys.executable, + "-c", + ( + "import sys; " + "import services.email_writing_orchestrator_port; " + "assert 'services.contextual_orchestrator_client' not in sys.modules; " + "assert 'core.config' not in sys.modules" + ), + ], + cwd=backend_root, + env=environment, + capture_output=True, + text=True, + timeout=10, + check=False, + ) + assert probe.returncode == 0, probe.stderr diff --git a/backend/tests/test_email_writing_orchestrator_scope.py b/backend/tests/test_email_writing_orchestrator_scope.py new file mode 100644 index 000000000..364ba8e54 --- /dev/null +++ b/backend/tests/test_email_writing_orchestrator_scope.py @@ -0,0 +1,151 @@ +"""Test-first tenant-scoped configuration contracts for email-writing orchestration.""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from db.email_writing_orchestrator_config import EmailWritingOrchestratorConfig +from services.tenant_config_scope import ( + EmailWritingOrchestratorConfigurationError, + email_writing_orchestrator_owner_filters, + get_scoped_email_writing_orchestrator_config, + new_scoped_email_writing_orchestrator_config, + resolve_email_writing_orchestrator_settings, +) + + +class _ScalarResult: + def __init__(self, value: Any) -> None: + self._value = value + + def scalar_one_or_none(self) -> Any: + return self._value + + +class _Session: + def __init__(self, value: Any) -> None: + self.value = value + self.queries: list[Any] = [] + + async def execute(self, query: Any) -> _ScalarResult: + self.queries.append(query) + return _ScalarResult(self.value) + + +def _config(**overrides: Any) -> EmailWritingOrchestratorConfig: + values: dict[str, Any] = { + "owner_user_id": "user_alpha", + "organization_id": "organization_alpha", + "orchestrator_enabled": True, + "orchestrator_base_url": "https://orchestrator.example", + "model_profile_id": "email-review-v1", + "inference_credential": "tenant-secret-token", + } + values.update(overrides) + return EmailWritingOrchestratorConfig(**values) + + +def test_email_writing_orchestrator_owner_filters_are_tenant_exact() -> None: + with_org = email_writing_orchestrator_owner_filters( + "user_alpha", "organization_alpha" + ) + assert len(with_org) == 2 + assert str(with_org[0].left) == "email_writing_orchestrator_config.owner_user_id" + assert with_org[0].right.value == "user_alpha" + assert str(with_org[1].left) == "email_writing_orchestrator_config.organization_id" + assert with_org[1].right.value == "organization_alpha" + + personal = email_writing_orchestrator_owner_filters("user_alpha", None) + assert personal[1].operator.__name__ == "is_" + + +@pytest.mark.asyncio +async def test_scoped_orchestrator_config_query_and_constructor() -> None: + existing = _config() + session = _Session(existing) + assert ( + await get_scoped_email_writing_orchestrator_config( + session, "user_alpha", "organization_alpha" + ) + is existing + ) + assert len(session.queries) == 1 + + created = new_scoped_email_writing_orchestrator_config( + "user_alpha", "organization_alpha" + ) + assert created.owner_user_id == "user_alpha" + assert created.organization_id == "organization_alpha" + assert created.orchestrator_enabled is False + + +@pytest.mark.asyncio +async def test_settings_resolver_is_disabled_by_default_and_trims_values() -> None: + assert ( + await resolve_email_writing_orchestrator_settings( + _Session(None), + user_id="user_alpha", + organization_id="organization_alpha", + ) + is None + ) + disabled = _config(orchestrator_enabled=False) + assert ( + await resolve_email_writing_orchestrator_settings( + _Session(disabled), + user_id="user_alpha", + organization_id="organization_alpha", + ) + is None + ) + + enabled = _config( + orchestrator_base_url=" https://orchestrator.example ", + model_profile_id=" email-review-v1 ", + inference_credential=" tenant-secret-token ", + ) + resolved = await resolve_email_writing_orchestrator_settings( + _Session(enabled), + user_id="user_alpha", + organization_id="organization_alpha", + ) + assert resolved is not None + assert resolved.base_url == "https://orchestrator.example" + assert resolved.model_profile_id == "email-review-v1" + assert resolved.inference_credential == "tenant-secret-token" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "field_name", + ["orchestrator_base_url", "model_profile_id", "inference_credential"], +) +async def test_enabled_incomplete_settings_fail_closed(field_name: str) -> None: + config = _config(**{field_name: " "}) + with pytest.raises(EmailWritingOrchestratorConfigurationError) as captured: + await resolve_email_writing_orchestrator_settings( + _Session(config), + user_id="user_alpha", + organization_id="organization_alpha", + ) + assert captured.value.code == "email_writing_orchestrator_incomplete" + assert str(captured.value) == "email_writing_orchestrator_incomplete" + + +def test_config_repr_and_evidence_surface_never_expose_inference_credential() -> None: + config = _config() + representation = repr(config) + evidence = config.to_evidence_dict() + assert "tenant-secret-token" not in representation + assert "tenant-secret-token" not in repr(evidence) + assert evidence == { + "orchestrator_config_id": None, + "owner_user_id": "user_alpha", + "organization_id": "organization_alpha", + "orchestrator_enabled": True, + "orchestrator_base_url": "https://orchestrator.example", + "model_profile_id": "email-review-v1", + "has_inference_credential": True, + } diff --git a/backend/tests/test_email_writing_orchestrator_terminal_coverage.py b/backend/tests/test_email_writing_orchestrator_terminal_coverage.py new file mode 100644 index 000000000..48fa6cc50 --- /dev/null +++ b/backend/tests/test_email_writing_orchestrator_terminal_coverage.py @@ -0,0 +1,811 @@ +"""Terminal branch coverage for the Task 5 orchestration boundary.""" + +from __future__ import annotations + +import asyncio +import importlib.util +import json +from pathlib import Path +import sys +from typing import Any, cast + +from fastapi import HTTPException +import httpx +from pydantic import ValidationError +import pytest +from sqlalchemy import create_engine, inspect + +from api.auth import AuthContext +from api import email_writing_orchestrator_config as config_api +from services import contextual_orchestrator_client as client_module +from services.contextual_orchestrator_client import ( + ContextualOrchestratorClient, + ContextualOrchestratorCompletion, + ContextualOrchestratorError, +) +from services.email_writing_orchestrator_port import EmailWritingOrchestratorPort +from services.llm_provider_urls import ValidatedLLMProviderBaseURL +from services import tenant_config_scope as scope_module + + +_MESSAGES = ( + {"role": "system", "content": "Return strict JSON."}, + {"role": "user", "content": "Review this draft."}, +) + + +def _validated( + *, + normalized_url: str = "https://orchestrator.example", + addresses: tuple[str, ...] = ("93.184.216.34",), +) -> ValidatedLLMProviderBaseURL: + """Build one deterministic endpoint validation result.""" + return ValidatedLLMProviderBaseURL( + normalized_url=normalized_url, + hostname="orchestrator.example", + port=443, + addresses=addresses, + ) + + +async def _valid_endpoint( + _value: str | None, +) -> ValidatedLLMProviderBaseURL: + """Return one deterministic valid endpoint.""" + return _validated() + + +def _payload( + *, + mode: str = "route", + trace: object | None = None, +) -> dict[str, object]: + """Build one strict success response.""" + return { + "choices": [ + { + "message": { + "role": "assistant", + "content": '{"diagnostics":[]}', + } + } + ], + "orchestration": { + "mode": mode, + "trace": ( + [ + { + "usage": { + "prompt_tokens": 1, + "completion_tokens": 1, + "total_tokens": 2, + } + } + ] + if trace is None + else trace + ), + }, + } + + +def _builder(handler: Any): + """Build a redirect-disabled mock HTTP client factory.""" + transport = httpx.MockTransport(handler) + + def build( + _normalized_url: str, + _hostname: str, + _port: int, + _addresses: tuple[str, ...], + ) -> httpx.AsyncClient: + """Build one mock client.""" + return httpx.AsyncClient( + transport=transport, + follow_redirects=False, + trust_env=False, + ) + + return build + + +def _client(**overrides: Any) -> ContextualOrchestratorClient: + """Build a client with deterministic defaults.""" + values: dict[str, Any] = { + "base_url": "https://orchestrator.example", + "inference_credential": "tenant-secret-token", + "model_profile_id": "email-review-v1", + "endpoint_validator": _valid_endpoint, + "client_builder": _builder( + lambda _request: httpx.Response( + 200, + json=_payload(), + ) + ), + "max_retries": 0, + } + values.update(overrides) + return ContextualOrchestratorClient(**values) + + +def test_private_validation_helpers_cover_all_terminal_outcomes( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Exercise deterministic helpers without making a network request.""" + sentinel = cast(httpx.AsyncClient, object()) + + def pinned_builder( + normalized_url: str, + hostname: str, + port: int, + addresses: tuple[str, ...], + ) -> httpx.AsyncClient: + assert ( + normalized_url, + hostname, + port, + addresses, + ) == ( + "https://orchestrator.example", + "orchestrator.example", + 443, + ("93.184.216.34",), + ) + return sentinel + + monkeypatch.setattr( + client_module, + "build_pinned_https_async_client", + pinned_builder, + ) + assert ( + client_module._default_client_builder( + "https://orchestrator.example", + "orchestrator.example", + 443, + ("93.184.216.34",), + ) + is sentinel + ) + assert client_module._contains_surrogate("ordinary") is False + assert client_module._contains_surrogate("\ud800") is True + assert ( + client_module._bounded_secret( + " value ", + maximum=16, + code="invalid", + ) + == "value" + ) + + for value in ( + cast(Any, 7), + "", + "x" * 17, + "\ud800", + "line\nbreak", + "\x7f", + ): + with pytest.raises(ContextualOrchestratorError, match="invalid"): + client_module._bounded_secret( + value, + maximum=16, + code="invalid", + ) + + assert client_module._strict_object_pairs([("a", 1)]) == {"a": 1} + with pytest.raises(ValueError, match="duplicate_json_key"): + client_module._strict_object_pairs([("a", 1), ("a", 2)]) + + assert client_module._bounded_counter(0) == 0 + assert client_module._bounded_counter(2**53 - 1) == 2**53 - 1 + for value in (True, 1.0, -1, 2**53): + with pytest.raises( + ContextualOrchestratorError, + match="orchestrator_malformed_response", + ): + client_module._bounded_counter(value) + + client_module._validate_json_structure( + {"array": [1, "ordinary"], "flag": True} + ) + with pytest.raises( + ContextualOrchestratorError, + match="orchestrator_malformed_response", + ): + client_module._validate_json_structure({"bad": "\ud800"}) + + monkeypatch.setattr(client_module, "_MAX_JSON_NODES", 2) + with pytest.raises( + ContextualOrchestratorError, + match="orchestrator_malformed_response", + ): + client_module._validate_json_structure([1, 2, 3]) + + +@pytest.mark.parametrize( + "overrides", + ( + {"model_profile_id": "invalid profile"}, + {"max_retries": -1}, + {"max_retries": 6}, + {"max_response_bytes": 0}, + {"circuit_failure_threshold": 0}, + {"circuit_open_seconds": 0}, + ), +) +def test_constructor_rejects_invalid_configuration( + overrides: dict[str, object], +) -> None: + """Reject invalid client bounds before allocating transport resources.""" + with pytest.raises( + (ContextualOrchestratorError, ValueError), + ): + _client(**overrides) + + +@pytest.mark.asyncio +async def test_invalid_mode_and_unreachable_loop_are_fail_closed() -> None: + """Reject an unsupported mode and prove the loop terminal is guarded.""" + client = _client() + with pytest.raises( + ContextualOrchestratorError, + match="orchestrator_policy_rejected", + ): + await client.complete(_MESSAGES, mode=cast(Any, "auto")) + + client._max_retries = -1 + with pytest.raises(AssertionError, match="unreachable completion loop"): + await client.complete(_MESSAGES, mode="route") + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "error_type", + (httpx.ConnectTimeout, httpx.ConnectError), +) +async def test_transport_exceptions_retry_and_exhaust( + error_type: type[httpx.RequestError], +) -> None: + """Retry transient HTTPX failures and return one stable public code.""" + attempts = 0 + delays: list[float] = [] + + async def handler(request: httpx.Request) -> httpx.Response: + nonlocal attempts + attempts += 1 + raise error_type("private transport detail", request=request) + + async def sleeper(delay: float) -> None: + delays.append(delay) + + client = _client( + client_builder=_builder(handler), + sleeper=sleeper, + max_retries=1, + ) + with pytest.raises( + ContextualOrchestratorError, + match="orchestrator_unavailable", + ) as captured: + await client.complete(_MESSAGES, mode="route") + assert captured.value.transient is True + assert attempts == 2 + assert delays == [0.05] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("raised", (ValueError, OSError)) +async def test_endpoint_validation_errors_are_redacted( + raised: type[Exception], +) -> None: + """Normalize endpoint resolution failures to the policy code.""" + + async def validator( + _value: str | None, + ) -> ValidatedLLMProviderBaseURL: + raise raised("private resolver detail") + + client = _client(endpoint_validator=validator) + with pytest.raises( + ContextualOrchestratorError, + match="orchestrator_policy_rejected", + ): + await client._validated_endpoint() + + +@pytest.mark.asyncio +async def test_endpoint_validation_rejects_absent_invalid_and_empty_results() -> None: + """Accept only one bare HTTPS origin with at least one pinned address.""" + + async def absent( + _value: str | None, + ) -> ValidatedLLMProviderBaseURL | None: + return None + + client = _client(endpoint_validator=absent) + with pytest.raises( + ContextualOrchestratorError, + match="orchestrator_policy_rejected", + ): + await client._validated_endpoint() + + invalid_urls = ( + "http://orchestrator.example", + "https://user@orchestrator.example", + "https://user:pass@orchestrator.example", + "https://orchestrator.example/path", + "https://orchestrator.example?query=1", + "https://orchestrator.example#fragment", + "https:///", + ) + for invalid_url in invalid_urls: + + async def invalid( + _value: str | None, + *, + candidate: str = invalid_url, + ) -> ValidatedLLMProviderBaseURL: + return _validated(normalized_url=candidate) + + client = _client(endpoint_validator=invalid) + with pytest.raises( + ContextualOrchestratorError, + match="orchestrator_policy_rejected", + ): + await client._validated_endpoint() + + async def no_addresses( + _value: str | None, + ) -> ValidatedLLMProviderBaseURL: + return _validated(addresses=()) + + client = _client(endpoint_validator=no_addresses) + with pytest.raises( + ContextualOrchestratorError, + match="orchestrator_policy_rejected", + ): + await client._validated_endpoint() + + async def duplicates( + _value: str | None, + ) -> ValidatedLLMProviderBaseURL: + return _validated( + addresses=( + "93.184.216.35", + "93.184.216.34", + "93.184.216.34", + ) + ) + + client = _client(endpoint_validator=duplicates) + result = await client._validated_endpoint() + assert result.addresses == ("93.184.216.34", "93.184.216.35") + + +def test_message_validation_covers_all_rejection_branches() -> None: + """Reject malformed messages, hostile Unicode, and exceeded budgets.""" + client = _client() + invalid_inputs: tuple[Any, ...] = ( + "message", + b"message", + object(), + (), + tuple({"role": "user", "content": "x"} for _ in range(65)), + (1,), + ({"role": "owner", "content": "x"},), + ({"role": 1, "content": "x"},), + ({"role": "user", "content": 1},), + ({"role": "user", "content": "x" * 200_001},), + ({"role": "user", "content": "\ud800"},), + ( + {"role": "user", "content": "x"}, + {"role": "assistant", "content": "y", "extra": "z"}, + ), + tuple( + {"role": "user", "content": "x" * 200_000} + for _ in range(6) + ), + ) + for invalid in invalid_inputs: + with pytest.raises( + ContextualOrchestratorError, + match="orchestrator_policy_rejected", + ): + client._validate_messages(invalid) + + assert client._validate_messages( + ({"role": "tool", "content": ""},) + ) == [{"role": "tool", "content": ""}] + + +def test_strict_json_and_upstream_code_terminal_cases( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Cover strict decoding, safe upstream code extraction, and node bounds.""" + client = _client() + for body in ( + b"\xff", + b'{"value": NaN}', + b"[]", + b'{"value":"\\ud800"}', + ): + with pytest.raises( + ContextualOrchestratorError, + match="orchestrator_malformed_response", + ): + client._strict_json(body) + + assert client._safe_upstream_error_code(b"not-json") is None + assert client._safe_upstream_error_code(b'{"error":"bad"}') is None + assert ( + client._safe_upstream_error_code( + b'{"error":{"code":7}}' + ) + is None + ) + too_long = json.dumps( + {"error": {"code": "x" * 129}} + ).encode() + assert client._safe_upstream_error_code(too_long) is None + assert ( + client._safe_upstream_error_code( + b'{"error":{"code":"stable"}}' + ) + == "stable" + ) + + monkeypatch.setattr(client_module, "_MAX_JSON_DEPTH", 1) + with pytest.raises( + ContextualOrchestratorError, + match="orchestrator_malformed_response", + ): + client._strict_json(b'{"nested":{"value":1}}') + + +@pytest.mark.parametrize( + "document", + ( + {"choices": None, "orchestration": {"mode": "route", "trace": []}}, + {"choices": [1], "orchestration": {"mode": "route", "trace": []}}, + {"choices": [{}], "orchestration": {"mode": "route", "trace": []}}, + { + "choices": [{"message": {"content": 7}}], + "orchestration": {"mode": "route", "trace": []}, + }, + { + "choices": [{"message": {"content": "\ud800"}}], + "orchestration": {"mode": "route", "trace": []}, + }, + { + "choices": [{"message": {"content": "ok"}}], + "orchestration": None, + }, + { + "choices": [{"message": {"content": "ok"}}], + "orchestration": {"mode": "route", "trace": "bad"}, + }, + { + "choices": [{"message": {"content": "ok"}}], + "orchestration": {"mode": "route", "trace": [1]}, + }, + { + "choices": [{"message": {"content": "ok"}}], + "orchestration": { + "mode": "route", + "trace": [{"usage": "bad"}], + }, + }, + ), +) +def test_completion_shape_rejections(document: dict[str, object]) -> None: + """Reject malformed nested response members before admitting evidence.""" + client = _client() + body = json.dumps(document, ensure_ascii=True).encode() + with pytest.raises( + ContextualOrchestratorError, + match="orchestrator_malformed_response", + ): + client._parse_completion(body) + + +def test_completion_accepts_an_empty_trace() -> None: + """Allow a strict completion that contains no orchestration steps.""" + client = _client() + completion = client._parse_completion( + json.dumps(_payload(trace=[])).encode() + ) + assert completion.trace == () + + +class _PortClient: + """Minimal candidate transport used to exercise port lifecycle branches.""" + + def __init__(self) -> None: + self.closed = False + + async def complete( + self, + _messages: object, + *, + mode: str, + ) -> ContextualOrchestratorCompletion: + """Return one strict completion.""" + return ContextualOrchestratorCompletion( + answer="ok", + mode=cast(Any, mode), + trace=(), + ) + + async def aclose(self) -> None: + """Record closure.""" + self.closed = True + + +@pytest.mark.parametrize("capacity", (0, 33)) +def test_port_rejects_invalid_worker_capacity(capacity: int) -> None: + """Reject worker counts outside the bounded production range.""" + with pytest.raises( + ValueError, + match="judge_capacity must be between 1 and 32", + ): + EmailWritingOrchestratorPort( + cast(Any, _PortClient()), + judge_capacity=capacity, + ) + + +@pytest.mark.asyncio +async def test_port_rejects_sync_use_on_event_loop_and_is_idempotent() -> None: + """Keep sync compatibility off the event loop and close only once.""" + transport = _PortClient() + port = EmailWritingOrchestratorPort(cast(Any, transport)) + with pytest.raises( + RuntimeError, + match="sync_completion_on_event_loop", + ): + port.complete(_MESSAGES, mode="route") + + await port.aclose() + await port.aclose() + assert transport.closed is True + + with pytest.raises(RuntimeError, match="orchestrator_port_closed"): + await port.complete_candidate(_MESSAGES, mode="route") + with pytest.raises(RuntimeError, match="orchestrator_port_closed"): + await asyncio.to_thread( + port.complete, + _MESSAGES, + mode="route", + ) + + +class _ScalarResult: + """Minimal scalar result for owner-scoped query tests.""" + + def __init__(self, value: Any) -> None: + self.value = value + + def scalar_one_or_none(self) -> Any: + """Return the stored scalar.""" + return self.value + + +class _Session: + """Record executed SQLAlchemy statements.""" + + def __init__(self, value: Any) -> None: + self.value = value + self.queries: list[Any] = [] + + async def execute(self, query: Any) -> _ScalarResult: + """Record one query and return the configured scalar.""" + self.queries.append(query) + return _ScalarResult(self.value) + + +@pytest.mark.asyncio +async def test_legacy_and_orchestrator_owner_scope_helpers() -> None: + """Cover both legacy and modular owner-scope helper branches.""" + with_organization = scope_module.tenant_config_owner_filters( + "user_alpha", + "organization_alpha", + ) + assert with_organization[1].right.value == "organization_alpha" + personal = scope_module.tenant_config_owner_filters( + "user_alpha", + None, + ) + assert personal[1].operator.__name__ == "is_" + + session = _Session(None) + assert ( + await scope_module.get_scoped_tenant_config( + cast(Any, session), + "user_alpha", + None, + ) + is None + ) + assert len(session.queries) == 1 + created = scope_module.new_scoped_tenant_config( + "user_alpha", + "organization_alpha", + ) + assert created.user_id == "user_alpha" + assert created.organization_id == "organization_alpha" + + orchestrator_personal = ( + scope_module.email_writing_orchestrator_owner_filters( + "user_alpha", + None, + ) + ) + assert orchestrator_personal[1].operator.__name__ == "is_" + assert scope_module._clean_orchestrator_value(None) is None + assert scope_module._clean_orchestrator_value(" value ") == "value" + assert scope_module._clean_orchestrator_value(" ") is None + + +def test_configuration_model_text_validation_is_bounded() -> None: + """Cover null, type, length, control, and valid normalization branches.""" + assert ( + config_api.EmailWritingOrchestratorConfigUpdate( + model_profile_id=None + ).model_profile_id + is None + ) + with pytest.raises(ValidationError): + config_api.EmailWritingOrchestratorConfigUpdate( + model_profile_id=cast(Any, 7) + ) + with pytest.raises(ValidationError): + config_api.EmailWritingOrchestratorConfigUpdate( + model_profile_id="x" * 256 + ) + with pytest.raises(ValidationError): + config_api.EmailWritingOrchestratorConfigUpdate( + model_profile_id="line\nbreak" + ) + assert ( + config_api.EmailWritingOrchestratorConfigUpdate( + model_profile_id=" profile " + ).model_profile_id + == "profile" + ) + + +@pytest.mark.asyncio +async def test_configuration_url_none_and_commit_failures_are_stable( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Cover the explicit-null endpoint and both database error branches.""" + + async def absent( + _value: str | None, + ) -> ValidatedLLMProviderBaseURL | None: + return None + + monkeypatch.setattr( + config_api, + "validate_llm_provider_base_url_details_async", + absent, + ) + assert await config_api._validated_orchestrator_url(None) is None + + async def no_existing( + _session: Any, + _user_id: str, + _organization_id: str | None, + ) -> None: + return None + + monkeypatch.setattr( + config_api, + "get_scoped_email_writing_orchestrator_config", + no_existing, + ) + + class FailingDatabase: + """Fail commits with one configured exception.""" + + def __init__(self, error: Exception) -> None: + self.error = error + self.added: list[object] = [] + + def add(self, value: object) -> None: + """Record the pending configuration.""" + self.added.append(value) + + async def commit(self) -> None: + """Raise the configured persistence failure.""" + raise self.error + + auth = AuthContext( + user_id="user_alpha", + role="member", + organization_id="organization_alpha", + group_ids=(), + workspace_id="workspace_alpha", + ) + update = config_api.EmailWritingOrchestratorConfigUpdate( + orchestrator_enabled=False, + inference_credential="opaque-value", + ) + + with pytest.raises(HTTPException) as encrypted: + await config_api.update_email_writing_orchestrator_config( + update, + cast( + Any, + FailingDatabase( + RuntimeError( + "ENCRYPTION_KEY is required: private configuration" + ) + ), + ), + auth, + ) + assert encrypted.value.status_code == 503 + assert encrypted.value.detail == ( + "Server encryption key is not configured. " + "Contact your workspace administrator." + ) + + with pytest.raises(RuntimeError, match="database unavailable"): + await config_api.update_email_writing_orchestrator_config( + update, + cast( + Any, + FailingDatabase( + RuntimeError("database unavailable") + ), + ), + auth, + ) + + +def test_migration_executes_upgrade_and_downgrade( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Execute the real migration against SQLite without dropping other data.""" + backend_root = Path(__file__).resolve().parents[1] + migration_path = ( + backend_root + / "alembic" + / "versions" + / "20260813_0001_add_email_writing_orchestrator_config.py" + ) + module_name = "task5_email_writing_orchestrator_migration" + spec = importlib.util.spec_from_file_location( + module_name, + migration_path, + ) + assert spec is not None + assert spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[module_name] = module + spec.loader.exec_module(module) + + engine = create_engine("sqlite://") + with engine.begin() as connection: + connection.exec_driver_sql( + "CREATE TABLE unrelated_record " + "(unrelated_record_id INTEGER PRIMARY KEY)" + ) + monkeypatch.setattr(module.op, "get_bind", lambda: connection) + + module.upgrade() + module.upgrade() + inspector = inspect(connection) + assert inspector.has_table("email_writing_orchestrator_config") + assert inspector.has_table("unrelated_record") + + module.downgrade() + module.downgrade() + inspector = inspect(connection) + assert not inspector.has_table( + "email_writing_orchestrator_config" + ) + assert inspector.has_table("unrelated_record") + engine.dispose()