diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index dcae972..832614e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -31,10 +31,10 @@ jobs: - uses: actions/checkout@93cb6efe18208431cddfb8368fd83d5badbf9bfd # v5 - name: Set up Docker Buildx - uses: docker/setup-buildx-action@b5730b14e8b0bc39f62ade545785cb7e6f44b97c # v3 + uses: docker/setup-buildx-action@8d2750c68a42422c14e847fe6c8ac0403b4cbd6f # v3 - name: Build image - uses: docker/build-push-action@471d1dc4e07e5cdedd4c2171150001c434f0b4b5 # v6 + uses: docker/build-push-action@10e90e3645eae34f1e60eeb005ba3a3d33f178e8 # v6 with: context: . platforms: linux/amd64 @@ -70,7 +70,7 @@ jobs: - name: Push to GHCR if: github.event_name == 'push' && github.ref == 'refs/heads/main' - uses: docker/build-push-action@471d1dc4e07e5cdedd4c2171150001c434f0b4b5 # v6 + uses: docker/build-push-action@10e90e3645eae34f1e60eeb005ba3a3d33f178e8 # v6 with: context: . platforms: linux/amd64 diff --git a/docs/architecture.md b/docs/architecture.md index f66d83e..550dd67 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -37,3 +37,10 @@ and principal IDs remain deployment-orchestration outputs and are not applicatio - CAS correlation IDs are attached to workflow spans and canonical events preserve W3C trace context. - Broad Azure SDK and outbound HTTP auto-instrumentation is disabled to avoid capturing prompt or output content. The application records only explicit boundary spans and safe identifiers. +# Loop execution boundaries + +The public HTTP ingress is an Azure Functions Flex Consumption Linux app. It validates the canonical prompt envelope and returns `202` only after handing the message to Azure Queue Storage. It never invokes Foundry, executes tools, or runs sandbox work in the request process. A separately scaled worker owns those long-running operations. + +The ingress uses a system-assigned managed identity. Bicep grants Storage Queue Data Contributor on the workload storage account and Azure AI User on the specific Foundry project. No account keys, API keys, or connection-string credentials are materialized. Local development continues to use `DefaultAzureCredential`; Azure environments use `ManagedIdentityCredential`. + +Loop spans use W3C trace context and the fixed stages `control_plane`, `worker`, `tool`, `verifier`, and `foundry`. Attributes are limited to correlation, goal, and work-item identifiers; prompts and model output are not telemetry attributes. diff --git a/docs/operations.md b/docs/operations.md index 1f37e99..bcf3f43 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -34,3 +34,13 @@ Build a Linux AMD64 image and pass its immutable image reference to the `contain The Docker build context excludes local `.env` files and development artifacts. The application does not consume platform resource IDs; deployment orchestration retains those outputs for RBAC and operations workflows. +# Flex ingress validation + +Build the infrastructure locally before any reviewed deployment: + +```powershell +az bicep build --file infra/main.bicep +az deployment group what-if --resource-group --template-file infra/main.bicep --parameters foundryProjectResourceId= +``` + +Deployment is intentionally not performed by repository validation. A rollout must publish the function package separately, verify the system-assigned principal's project-scoped and storage-scoped role assignments, submit one synthetic envelope, and confirm a queue message and correlated Application Insights spans. Rollback disables ingress or restores the prior function package; storage and identity resources are retained to avoid destructive replacement. diff --git a/functions/ingress/function_app.py b/functions/ingress/function_app.py new file mode 100644 index 0000000..bc65339 --- /dev/null +++ b/functions/ingress/function_app.py @@ -0,0 +1,24 @@ +import azure.functions as func + +from cas_reference_product.ingress import InvalidIngressRequest, create_worker_message + +app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION) + + +@app.route(route="workflows", methods=["POST"]) +@app.queue_output( + arg_name="worker_message", + queue_name="%WORK_QUEUE_NAME%", + connection="WORK_QUEUE_STORAGE", +) +def submit_workflow( + request: func.HttpRequest, + worker_message: func.Out[str], +) -> func.HttpResponse: + try: + message = create_worker_message(request.get_body()) + except InvalidIngressRequest: + return func.HttpResponse("Invalid request.", status_code=400) + + worker_message.set(message) + return func.HttpResponse(status_code=202) diff --git a/functions/ingress/host.json b/functions/ingress/host.json new file mode 100644 index 0000000..369b5be --- /dev/null +++ b/functions/ingress/host.json @@ -0,0 +1,11 @@ +{ + "version": "2.0", + "logging": { + "applicationInsights": { + "samplingSettings": { + "isEnabled": true, + "excludedTypes": "Request" + } + } + } +} diff --git a/functions/ingress/requirements.txt b/functions/ingress/requirements.txt new file mode 100644 index 0000000..66e16d4 --- /dev/null +++ b/functions/ingress/requirements.txt @@ -0,0 +1,2 @@ +azure-functions>=1.23.0 +cas-reference-product diff --git a/infra/main.bicep b/infra/main.bicep new file mode 100644 index 0000000..e7f99aa --- /dev/null +++ b/infra/main.bicep @@ -0,0 +1,70 @@ +targetScope = 'resourceGroup' + +@description('Short lowercase application name used in resource names.') +param applicationName string = 'casloop' + +@description('Deployment environment name.') +@allowed(['dev', 'test', 'staging', 'prod']) +param environment string = 'dev' + +@description('Azure region for all regional resources.') +param location string = resourceGroup().location + +@description('Foundry project resource ID receiving the agent caller role assignment.') +param foundryProjectResourceId string + +var suffix = uniqueString(resourceGroup().id, applicationName, environment) +var foundrySegments = split(foundryProjectResourceId, '/') +var tags = { + application: applicationName + environment: environment + managedBy: 'bicep' + dataClassification: 'internal' +} + +module observability 'modules/observability.bicep' = { + name: 'observability' + params: { + name: '${applicationName}-${environment}' + location: location + tags: tags + } +} + +module storage 'modules/storage.bicep' = { + name: 'storage' + params: { + name: take('st${applicationName}${environment}${suffix}', 24) + location: location + tags: tags + } +} + +module ingress 'modules/function-ingress.bicep' = { + name: 'function-ingress' + params: { + name: '${applicationName}-${environment}-${suffix}' + location: location + storageAccountName: storage.outputs.name + storageAccountId: storage.outputs.id + queueName: storage.outputs.queueName + applicationInsightsConnectionString: observability.outputs.connectionString + tags: tags + } +} + +module foundryRbac 'modules/foundry-rbac.bicep' = { + name: 'foundry-rbac' + scope: resourceGroup(foundrySegments[2], foundrySegments[4]) + params: { + accountName: foundrySegments[8] + projectName: foundrySegments[10] + principalId: ingress.outputs.principalId + } +} + +@description('System-assigned principal used by ingress for queue and Foundry access.') +output ingressPrincipalId string = ingress.outputs.principalId + +@description('Ingress function resource ID.') +output ingressResourceId string = ingress.outputs.id diff --git a/infra/modules/foundry-rbac.bicep b/infra/modules/foundry-rbac.bicep new file mode 100644 index 0000000..9c48e25 --- /dev/null +++ b/infra/modules/foundry-rbac.bicep @@ -0,0 +1,24 @@ +targetScope = 'resourceGroup' + +param accountName string +param projectName string +param principalId string + +resource foundryAccount 'Microsoft.CognitiveServices/accounts@2025-04-01-preview' existing = { + name: accountName +} + +resource foundryProject 'Microsoft.CognitiveServices/accounts/projects@2025-04-01-preview' existing = { + parent: foundryAccount + name: projectName +} + +resource foundryUser 'Microsoft.Authorization/roleAssignments@2022-04-01' = { + name: guid(foundryProject.id, principalId, 'foundry-project-user') + scope: foundryProject + properties: { + principalId: principalId + principalType: 'ServicePrincipal' + roleDefinitionId: subscriptionResourceId('Microsoft.Authorization/roleDefinitions', '53ca6127-db72-4b80-b1b0-d745d6d5456d') + } +} diff --git a/infra/modules/function-ingress.bicep b/infra/modules/function-ingress.bicep new file mode 100644 index 0000000..1f80d2b --- /dev/null +++ b/infra/modules/function-ingress.bicep @@ -0,0 +1,57 @@ +param name string +param location string +param storageAccountName string +param storageAccountId string +param queueName string +param applicationInsightsConnectionString string +param tags object + +resource plan 'Microsoft.Web/serverfarms@2024-04-01' = { + name: 'plan-${name}' + location: location + tags: tags + kind: 'functionapp' + sku: { name: 'FC1', tier: 'FlexConsumption' } + properties: { reserved: true } +} + +resource functionApp 'Microsoft.Web/sites@2024-04-01' = { + name: 'func-${name}' + location: location + tags: tags + kind: 'functionapp,linux' + identity: { type: 'SystemAssigned' } + properties: { + serverFarmId: plan.id + httpsOnly: true + publicNetworkAccess: 'Enabled' + siteConfig: { + linuxFxVersion: 'Python|3.12' + minTlsVersion: '1.2' + ftpsState: 'Disabled' + appSettings: [ + { name: 'FUNCTIONS_WORKER_RUNTIME', value: 'python' } + { name: 'WORK_QUEUE_NAME', value: queueName } + { name: 'WORK_QUEUE_STORAGE__queueServiceUri', value: 'https://${storageAccountName}.queue.${environment().suffixes.storage}' } + { name: 'APPLICATIONINSIGHTS_CONNECTION_STRING', value: applicationInsightsConnectionString } + ] + } + } +} + +resource storageAccount 'Microsoft.Storage/storageAccounts@2023-05-01' existing = { + name: storageAccountName +} + +resource queueContributor 'Microsoft.Authorization/roleAssignments@2022-04-01' = { + name: guid(storageAccountId, functionApp.id, 'queue-contributor') + scope: storageAccount + properties: { + principalId: functionApp.identity.principalId + principalType: 'ServicePrincipal' + roleDefinitionId: subscriptionResourceId('Microsoft.Authorization/roleDefinitions', '974c5e8b-45b9-4653-ba55-5f855dd0fb88') + } +} + +output id string = functionApp.id +output principalId string = functionApp.identity.principalId diff --git a/infra/modules/observability.bicep b/infra/modules/observability.bicep new file mode 100644 index 0000000..376b1f6 --- /dev/null +++ b/infra/modules/observability.bicep @@ -0,0 +1,27 @@ +param name string +param location string +param tags object + +resource workspace 'Microsoft.OperationalInsights/workspaces@2023-09-01' = { + name: 'log-${name}' + location: location + tags: tags + properties: { + retentionInDays: 30 + features: { enableLogAccessUsingOnlyResourcePermissions: true } + } +} + +resource insights 'Microsoft.Insights/components@2020-02-02' = { + name: 'appi-${name}' + location: location + tags: tags + kind: 'web' + properties: { + Application_Type: 'web' + WorkspaceResourceId: workspace.id + IngestionMode: 'LogAnalytics' + } +} + +output connectionString string = insights.properties.ConnectionString diff --git a/infra/modules/storage.bicep b/infra/modules/storage.bicep new file mode 100644 index 0000000..824bfc1 --- /dev/null +++ b/infra/modules/storage.bicep @@ -0,0 +1,32 @@ +@description('Globally unique storage account name.') +param name string +param location string +param tags object + +resource account 'Microsoft.Storage/storageAccounts@2023-05-01' = { + name: name + location: location + tags: tags + sku: { name: 'Standard_LRS' } + kind: 'StorageV2' + properties: { + allowBlobPublicAccess: false + allowSharedKeyAccess: false + minimumTlsVersion: 'TLS1_2' + publicNetworkAccess: 'Enabled' + } +} + +resource queueService 'Microsoft.Storage/storageAccounts/queueServices@2023-05-01' = { + parent: account + name: 'default' +} + +resource workQueue 'Microsoft.Storage/storageAccounts/queueServices/queues@2023-05-01' = { + parent: queueService + name: 'cas-work-items' +} + +output id string = account.id +output name string = account.name +output queueName string = workQueue.name diff --git a/pyproject.toml b/pyproject.toml index 99097f3..80b1810 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,6 +10,7 @@ readme = "README.md" requires-python = ">=3.12,<3.15" license = { text = "MIT" } dependencies = [ + "azure-functions>=1.23.0", "azure-ai-projects>=1.0.0", "azure-identity>=1.19.0", "azure-monitor-opentelemetry>=1.6.0", diff --git a/scripts/validate.ps1 b/scripts/validate.ps1 index ead45af..d6cc5ab 100644 --- a/scripts/validate.ps1 +++ b/scripts/validate.ps1 @@ -2,9 +2,10 @@ $ErrorActionPreference = 'Stop' function Invoke-Checked { param([scriptblock]$Command) + $global:LASTEXITCODE = 0 & $Command - if ($LASTEXITCODE -ne 0) { - throw "Command failed with exit code $LASTEXITCODE" + if ($null -ne $global:LASTEXITCODE -and $global:LASTEXITCODE -ne 0) { + throw "Command failed with exit code $global:LASTEXITCODE" } } @@ -17,4 +18,12 @@ Invoke-Checked { & .\.venv\Scripts\python.exe -m ruff check . } Invoke-Checked { & .\.venv\Scripts\python.exe -m mypy } Invoke-Checked { & .\.venv\Scripts\python.exe -m pytest } Invoke-Checked { & .\.venv\Scripts\python.exe -m cas_reference_product.evidence } -Invoke-Checked { git -c safe.directory="$PWD" diff --check } +$git = Get-Command git -ErrorAction SilentlyContinue +if ($null -eq $git -and (Test-Path 'C:\Program Files\Git\cmd\git.exe')) { + $git = Get-Item 'C:\Program Files\Git\cmd\git.exe' +} +if ($null -eq $git) { + throw 'Git executable was not found.' +} +$gitPath = if ($git.Source) { $git.Source } else { $git.FullName } +Invoke-Checked { & $gitPath -c safe.directory="$PWD" diff --check } diff --git a/src/cas_reference_product/ingress.py b/src/cas_reference_product/ingress.py new file mode 100644 index 0000000..4f0def8 --- /dev/null +++ b/src/cas_reference_product/ingress.py @@ -0,0 +1,18 @@ +import json + +from pydantic import ValidationError + +from .models import PromptEnvelope + + +class InvalidIngressRequest(ValueError): + """The ingress payload does not satisfy the canonical prompt envelope.""" + + +def create_worker_message(payload: bytes) -> str: + try: + envelope = PromptEnvelope.model_validate_json(payload) + except (ValidationError, ValueError) as error: + raise InvalidIngressRequest("invalid CAS prompt envelope") from error + + return json.dumps(envelope.model_dump(mode="json"), separators=(",", ":")) diff --git a/src/cas_reference_product/telemetry.py b/src/cas_reference_product/telemetry.py index 5e0b4a0..54a60b2 100644 --- a/src/cas_reference_product/telemetry.py +++ b/src/cas_reference_product/telemetry.py @@ -1,4 +1,7 @@ from collections.abc import Awaitable, Callable +from contextlib import AbstractContextManager +from enum import StrEnum +from typing import Any from opentelemetry import context, propagate, trace from opentelemetry.propagators.composite import CompositePropagator @@ -13,6 +16,32 @@ tracer = trace.get_tracer(__name__) +class LoopStage(StrEnum): + CONTROL_PLANE = "control_plane" + WORKER = "worker" + TOOL = "tool" + VERIFIER = "verifier" + FOUNDRY = "foundry" + + +def start_loop_span( + stage: LoopStage, + correlation_id: str, + *, + goal_id: str | None = None, + work_item_id: str | None = None, +) -> AbstractContextManager[Any]: + attributes = { + "cas.stage": stage.value, + "cas.correlation_id": correlation_id, + } + if goal_id is not None: + attributes["cas.goal_id"] = goal_id + if work_item_id is not None: + attributes["cas.work_item_id"] = work_item_id + return tracer.start_as_current_span(f"cas.loop.{stage.value}", attributes=attributes) + + # --------------------------------------------------------------------------- # W3C trace-context propagation middleware # --------------------------------------------------------------------------- diff --git a/src/cas_reference_product/workflow.py b/src/cas_reference_product/workflow.py index 04e751b..7354210 100644 --- a/src/cas_reference_product/workflow.py +++ b/src/cas_reference_product/workflow.py @@ -3,14 +3,12 @@ from typing import Literal, Protocol from azure.ai.projects import AIProjectClient -from opentelemetry import trace from .config import Settings from .identity import build_credential from .models import Actor, PromptEnvelope, RunEvent, TraceContext, WorkflowResult -from .telemetry import current_traceparent +from .telemetry import LoopStage, current_traceparent, start_loop_span -tracer = trace.get_tracer(__name__) RunStatus = Literal["queued", "running", "succeeded", "failed", "cancelled"] @@ -47,7 +45,7 @@ def __init__(self, settings: Settings) -> None: ).get_openai_client() def run(self, envelope: PromptEnvelope) -> str: - with tracer.start_as_current_span("foundry.responses.create"): + with start_loop_span(LoopStage.FOUNDRY, envelope.correlationId): try: response = self._client.responses.create( input=envelope.prompt, @@ -81,7 +79,7 @@ def __init__( self._clock = clock def execute(self, envelope: PromptEnvelope) -> WorkflowResult: - with tracer.start_as_current_span("cas.workflow.execute") as span: + with start_loop_span(LoopStage.WORKER, envelope.correlationId) as span: span.set_attribute("cas.correlation_id", envelope.correlationId) events = [self._event(envelope, 0, "workflow.started", "running", "Workflow started.")] try: diff --git a/tests/test_function_boundary.py b/tests/test_function_boundary.py new file mode 100644 index 0000000..4736f5d --- /dev/null +++ b/tests/test_function_boundary.py @@ -0,0 +1,16 @@ +import json + +import pytest + +from cas_reference_product.ingress import InvalidIngressRequest, create_worker_message + + +def test_ingress_validates_and_serializes_canonical_envelope(envelope) -> None: + message = create_worker_message(envelope.model_dump_json().encode()) + + assert json.loads(message)["runId"] == envelope.runId + + +def test_ingress_rejects_malformed_work_without_reasoning() -> None: + with pytest.raises(InvalidIngressRequest, match="invalid CAS prompt envelope"): + create_worker_message(b'{"prompt":"missing canonical fields"}') diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py index f107ea3..c5a4fce 100644 --- a/tests/test_telemetry.py +++ b/tests/test_telemetry.py @@ -2,15 +2,20 @@ from unittest.mock import MagicMock, patch -import pytest from fastapi.testclient import TestClient -from opentelemetry import trace +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from cas_reference_product import telemetry from cas_reference_product.app import create_app from cas_reference_product.config import Settings -from cas_reference_product.telemetry import configure_telemetry, current_traceparent, install_propagator - +from cas_reference_product.telemetry import ( + LoopStage, + configure_telemetry, + current_traceparent, + install_propagator, +) # --------------------------------------------------------------------------- # TEL-04: Application Insights exporter — no-op when connection string absent @@ -73,6 +78,47 @@ def test_invalid_span_preserves_incoming_traceparent() -> None: assert value == incoming +def test_valid_span_returns_formatted_traceparent() -> None: + with patch("cas_reference_product.telemetry.trace.get_current_span") as current: + span = MagicMock() + ctx = MagicMock() + ctx.is_valid = True + ctx.trace_id = 0x4BF92F3577B34DA6A3CE929D0E0E4736 + ctx.span_id = 0x00F067AA0BA902B7 + span.get_span_context.return_value = ctx + current.return_value = span + + value = current_traceparent("fallback") + + assert value == "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01" + + +def test_all_loop_stages_share_one_trace_without_prompt_or_output_attributes() -> None: + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + + with patch.object(telemetry, "tracer", provider.get_tracer("test")): + with telemetry.start_loop_span(LoopStage.CONTROL_PLANE, "corr-1", goal_id="goal-1"): + for stage in ( + LoopStage.WORKER, + LoopStage.TOOL, + LoopStage.VERIFIER, + LoopStage.FOUNDRY, + ): + with telemetry.start_loop_span(stage, "corr-1", work_item_id="work-1"): + pass + + spans = exporter.get_finished_spans() + assert {span.attributes["cas.stage"] for span in spans} == {stage.value for stage in LoopStage} + assert len({span.context.trace_id for span in spans}) == 1 + assert all( + "prompt" not in key and "output" not in key + for span in spans + for key in span.attributes + ) + + # --------------------------------------------------------------------------- # TEL-03: W3C propagator installed # ---------------------------------------------------------------------------