From 8a40cc50b05a48056194fb52023a3ea59adb76af Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Fri, 14 Aug 2026 12:07:30 -0400 Subject: [PATCH 1/3] feat: isolate default and stateful execution profiles --- README.md | 21 ++++ docs/lambda-microvm/README.md | 2 + helm/codeapi/README.md | 9 ++ helm/codeapi/templates/api-deployment.yaml | 2 + .../templates/worker-sandbox-deployment.yaml | 2 + helm/codeapi/values.yaml | 5 + service/openapi.yml | 107 +++++++++++++++++- service/src/api-server.ts | 2 + service/src/config.ts | 18 ++- service/src/enum/service.ts | 5 - service/src/execution-profile.test.ts | 83 ++++++++++++++ service/src/execution-profile.ts | 97 ++++++++++++++++ service/src/lifecycle.ts | 10 +- service/src/local-api.ts | 4 + service/src/metrics.ts | 19 ++++ .../src/middleware/execution-profile.test.ts | 78 +++++++++++++ service/src/middleware/execution-profile.ts | 37 ++++++ service/src/queue.ts | 20 ++-- service/src/secure-startup.test.ts | 58 ++++++++++ service/src/secure-startup.ts | 35 ++++++ service/src/service-api.ts | 2 + service/src/service/router.ts | 6 +- service/src/workers.ts | 7 +- 23 files changed, 606 insertions(+), 23 deletions(-) create mode 100644 service/src/execution-profile.test.ts create mode 100644 service/src/execution-profile.ts create mode 100644 service/src/middleware/execution-profile.test.ts create mode 100644 service/src/middleware/execution-profile.ts diff --git a/README.md b/README.md index 718db6a..f848b1d 100644 --- a/README.md +++ b/README.md @@ -24,6 +24,27 @@ Code Interpreter (internally `codeapi`, the prefix used by its env vars, images, 4. Files are persisted/retrieved via the **File Server** (backed by S3) 5. Tool calls from within sandboxes are routed through the **Tool Call Server** +## Execution profiles + +Code API can run two isolated deployments at the same time: + +- `default`: the AWS-free HTTP/libkrun path, with stateless executions. +- `stateful`: the AWS Lambda MicroVM path, with runtime-session affinity. + +Set `CODEAPI_EXECUTION_PROFILE` consistently on an API deployment and its +workers. The default profile keeps the existing `python-queue` and +`other-queue`; the stateful profile uses `stateful-python-queue` and +`stateful-other-queue`. This allows both deployments to share Redis without +cross-consuming jobs. + +Trusted callers should send `X-CodeAPI-Expected-Profile: default|stateful` on +every Code API request. A request that reaches the wrong deployment fails +before enqueue with HTTP 409 and `code=execution_profile_mismatch`; every +response advertises the actual deployment in `X-CodeAPI-Execution-Profile`. +Omitting the expected-profile header remains supported for older clients, but +provides no wrong-endpoint protection. There is deliberately no silent +fallback between profiles and no automatic workspace or file migration. + ## Sandbox Isolation Two modes are supported: diff --git a/docs/lambda-microvm/README.md b/docs/lambda-microvm/README.md index adbad83..4bd3e87 100644 --- a/docs/lambda-microvm/README.md +++ b/docs/lambda-microvm/README.md @@ -246,6 +246,7 @@ builds. ```bash CODEAPI_SANDBOX_BACKEND=lambda-microvm +CODEAPI_EXECUTION_PROFILE=stateful CODEAPI_RUNTIME_SESSION_MODE=affinity # warm sessions + checkpoints LAMBDA_MICROVM_IMAGE_ARN= LAMBDA_MICROVM_IMAGE_VERSION= # required for affinity/strict @@ -310,6 +311,7 @@ appear in `api/src/config.ts`. | Env | Default | Meaning | |---|---|---| | `CODEAPI_SANDBOX_BACKEND` | `http` | `http` (byte-identical to today) or `lambda-microvm`. | +| `CODEAPI_EXECUTION_PROFILE` | inferred | `default` for the HTTP/stateless deployment or `stateful` for the Lambda affinity/strict deployment. Stateful API and worker processes consume isolated BullMQ queues. | | `CODEAPI_RUNTIME_SESSION_MODE` | `stateless` | `stateless` \| `affinity` \| `strict`. `affinity` and `strict` require the `lambda-microvm` backend. See [Operating modes](#operating-modes). | | `CODEAPI_RUNTIME_SESSION_LOCK_WAIT_MS` | `15000` | How long a stateful execution waits for the session lock before returning `RUNTIME_SESSION_BUSY` (HTTP 409). | diff --git a/helm/codeapi/README.md b/helm/codeapi/README.md index 9198b1a..83cf2a2 100644 --- a/helm/codeapi/README.md +++ b/helm/codeapi/README.md @@ -53,6 +53,15 @@ platform rather than templated here: external ingress/service mesh, KEDA-style queue-depth autoscaling, and cloud-IAM secret delivery (the env hooks below cover all of them). +**Execution profile.** This chart deploys the AWS-free `default` profile and +sets `CODEAPI_EXECUTION_PROFILE=default` on both API and worker pods. That +profile requires the HTTP sandbox backend in stateless mode and retains the +existing `python-queue` / `other-queue` BullMQ names. A separate stateful +Lambda MicroVM deployment must use `CODEAPI_EXECUTION_PROFILE=stateful`; it +then consumes `stateful-python-queue` / `stateful-other-queue`, so both stacks +may safely share Redis without consuming each other's jobs. Do not mix API +and worker profile values within one deployment. + **Authentication.** Outside local mode the API verifies JWTs. Configure the verifier through environment variables on the api component, e.g.: diff --git a/helm/codeapi/templates/api-deployment.yaml b/helm/codeapi/templates/api-deployment.yaml index 0c801ab..7a62167 100644 --- a/helm/codeapi/templates/api-deployment.yaml +++ b/helm/codeapi/templates/api-deployment.yaml @@ -41,6 +41,8 @@ spec: {{ include "codeapi.otel.env" (dict "root" . "serviceName" "aiml-codeapi-api") | nindent 12 }} - name: CODEAPI_HARDENED_SANDBOX_MODE value: {{ .Values.hardenedSandboxMode | quote }} + - name: CODEAPI_EXECUTION_PROFILE + value: {{ .Values.executionProfile | quote }} # Redis connection - name: REDIS_HOST value: {{ include "codeapi.redis.host" . }} diff --git a/helm/codeapi/templates/worker-sandbox-deployment.yaml b/helm/codeapi/templates/worker-sandbox-deployment.yaml index 3484a61..c1bea5e 100644 --- a/helm/codeapi/templates/worker-sandbox-deployment.yaml +++ b/helm/codeapi/templates/worker-sandbox-deployment.yaml @@ -147,6 +147,8 @@ spec: {{ include "codeapi.otel.env" (dict "root" . "serviceName" "aiml-codeapi-service-worker") | nindent 12 }} - name: CODEAPI_HARDENED_SANDBOX_MODE value: {{ .Values.hardenedSandboxMode | quote }} + - name: CODEAPI_EXECUTION_PROFILE + value: {{ .Values.executionProfile | quote }} - name: SANDBOX_ENDPOINT value: "http://{{ include "codeapi.fullname" . }}-sandbox-runner:{{ .Values.workerSandbox.sandbox.port }}/api/v2" - name: EGRESS_GATEWAY_URL diff --git a/helm/codeapi/values.yaml b/helm/codeapi/values.yaml index 2385814..0f9a225 100644 --- a/helm/codeapi/values.yaml +++ b/helm/codeapi/values.yaml @@ -22,6 +22,11 @@ internalServiceAuth: hardenedSandboxMode: true +# Stable identity advertised by this API/worker deployment. The bundled chart +# is the AWS-free HTTP/libkrun profile. A separate Lambda MicroVM deployment +# must set this to `stateful`; the service then uses isolated BullMQ queues. +executionProfile: default + otel: enabled: false # OTLP/HTTP collector endpoint, e.g. "http://opentelemetry-collector.observability:4318". diff --git a/service/openapi.yml b/service/openapi.yml index c1f8f6e..16d36d7 100644 --- a/service/openapi.yml +++ b/service/openapi.yml @@ -2,7 +2,10 @@ openapi: '3.0.0' info: title: LibreChat Code Interpreter API version: '1.0.0' - description: API for sandbox code execution and file management + description: >- + API for sandbox code execution and file management. Trusted callers should + assert the intended deployment with X-CodeAPI-Expected-Profile on every + request; responses advertise the actual profile. servers: - url: https://api.librechat.ai/v1 description: LibreChat API server @@ -17,6 +20,45 @@ components: scheme: bearer bearerFormat: JWT + parameters: + ExpectedExecutionProfile: + name: X-CodeAPI-Expected-Profile + in: header + required: false + description: >- + Trusted routing assertion. A mismatched endpoint returns HTTP 409 + before any work is enqueued. Optional only for backwards compatibility. + schema: + type: string + enum: [default, stateful] + + headers: + ExecutionProfile: + description: Execution profile served by this deployment. + schema: + type: string + enum: [default, stateful] + + responses: + InvalidExecutionProfile: + description: Invalid expected execution profile + headers: + X-CodeAPI-Execution-Profile: + $ref: '#/components/headers/ExecutionProfile' + content: + application/json: + schema: + $ref: '#/components/schemas/ExecutionProfileError' + ExecutionProfileMismatch: + description: The request reached a different execution profile + headers: + X-CodeAPI-Execution-Profile: + $ref: '#/components/headers/ExecutionProfile' + content: + application/json: + schema: + $ref: '#/components/schemas/ExecutionProfileError' + schemas: FileRef: type: object @@ -108,6 +150,12 @@ components: type: array items: $ref: '#/components/schemas/RequestFile' + runtime_session_hint: + type: string + description: >- + Stable opaque hint for stateful runtime reuse. The server binds it + to the authenticated tenant and user. Ignored by the default + stateless profile. FileObject: type: object @@ -156,12 +204,29 @@ components: details: type: string + ExecutionProfileError: + type: object + required: [error, code, actual_profile] + properties: + error: + type: string + code: + type: string + enum: [invalid_execution_profile, execution_profile_mismatch] + expected_profile: + type: string + actual_profile: + type: string + enum: [default, stateful] + paths: /exec: post: summary: Execute code description: Execute code with specified language and parameters operationId: executeCode + parameters: + - $ref: '#/components/parameters/ExpectedExecutionProfile' requestBody: required: true content: @@ -171,6 +236,9 @@ paths: responses: '200': description: Successful execution + headers: + X-CodeAPI-Execution-Profile: + $ref: '#/components/headers/ExecutionProfile' content: application/json: schema: @@ -181,6 +249,10 @@ paths: application/json: schema: $ref: '#/components/schemas/Error' + '400': + $ref: '#/components/responses/InvalidExecutionProfile' + '409': + $ref: '#/components/responses/ExecutionProfileMismatch' '503': description: Service unavailable content: @@ -192,6 +264,7 @@ paths: get: summary: Download a file parameters: + - $ref: '#/components/parameters/ExpectedExecutionProfile' - name: session_id in: path required: true @@ -205,6 +278,9 @@ paths: responses: '200': description: File content + headers: + X-CodeAPI-Execution-Profile: + $ref: '#/components/headers/ExecutionProfile' content: application/octet-stream: schema: @@ -216,10 +292,16 @@ paths: application/json: schema: $ref: '#/components/schemas/Error' + '400': + $ref: '#/components/responses/InvalidExecutionProfile' + '409': + $ref: '#/components/responses/ExecutionProfileMismatch' /upload: post: summary: Upload files + parameters: + - $ref: '#/components/parameters/ExpectedExecutionProfile' requestBody: required: true content: @@ -237,6 +319,9 @@ paths: responses: '200': description: Successful upload + headers: + X-CodeAPI-Execution-Profile: + $ref: '#/components/headers/ExecutionProfile' content: application/json: schema: @@ -247,11 +332,16 @@ paths: application/json: schema: $ref: '#/components/schemas/Error' + '400': + $ref: '#/components/responses/InvalidExecutionProfile' + '409': + $ref: '#/components/responses/ExecutionProfileMismatch' /files/{session_id}: get: summary: Get files information parameters: + - $ref: '#/components/parameters/ExpectedExecutionProfile' - name: session_id in: path required: true @@ -265,17 +355,25 @@ paths: responses: '200': description: Files information + headers: + X-CodeAPI-Execution-Profile: + $ref: '#/components/headers/ExecutionProfile' content: application/json: schema: type: array items: $ref: '#/components/schemas/FileObject' + '400': + $ref: '#/components/responses/InvalidExecutionProfile' + '409': + $ref: '#/components/responses/ExecutionProfileMismatch' /files/{session_id}/{fileId}: delete: summary: Delete a file parameters: + - $ref: '#/components/parameters/ExpectedExecutionProfile' - name: session_id in: path required: true @@ -289,9 +387,16 @@ paths: responses: '200': description: File deleted successfully + headers: + X-CodeAPI-Execution-Profile: + $ref: '#/components/headers/ExecutionProfile' '500': description: Error deleting file content: application/json: schema: $ref: '#/components/schemas/Error' + '400': + $ref: '#/components/responses/InvalidExecutionProfile' + '409': + $ref: '#/components/responses/ExecutionProfileMismatch' diff --git a/service/src/api-server.ts b/service/src/api-server.ts index 89aba48..7868982 100644 --- a/service/src/api-server.ts +++ b/service/src/api-server.ts @@ -21,6 +21,7 @@ import programmaticRouter from './service/programmatic-router'; import { connection } from './queue'; import { metricsHandler } from './metrics'; import { httpMetricsMiddleware } from './middleware/httpMetrics'; +import { executionProfileMiddleware } from './middleware/execution-profile'; import { traceHttpRequest } from './telemetry'; import { env } from './config'; import logger from './logger'; @@ -32,6 +33,7 @@ app.disable('x-powered-by'); app.set('trust proxy', 1); app.use(traceHttpRequest('codeapi.api.request')); app.use(httpMetricsMiddleware); +app.use(executionProfileMiddleware); const v1 = Router(); diff --git a/service/src/config.ts b/service/src/config.ts index c9dcb68..4309af1 100644 --- a/service/src/config.ts +++ b/service/src/config.ts @@ -3,6 +3,7 @@ dotenv.config(); import { nanoid } from 'nanoid'; import type * as t from './types'; import { Languages } from './enum'; +import { resolveExecutionProfile } from './execution-profile'; export const languageConfig: Record = { [Languages.bash]: { language: 'bash', version: '5.2.0', fileName: 'script.sh' }, @@ -259,6 +260,9 @@ export function resolveRuntimeSessionMode( ); } +const sandboxBackend = resolveSandboxBackend(process.env.CODEAPI_SANDBOX_BACKEND); +const runtimeSessionMode = resolveRuntimeSessionMode(process.env.CODEAPI_RUNTIME_SESSION_MODE); + export const env = { PORT: process.env.SERVICE_PORT ?? 3112, LOCAL_MODE: process.env.LOCAL_MODE === 'true', @@ -344,7 +348,7 @@ export const env = { * (current Kubernetes/libkrun sandbox-runner). * - `lambda-microvm`: AWS Lambda MicroVM backend. */ - SANDBOX_BACKEND: resolveSandboxBackend(process.env.CODEAPI_SANDBOX_BACKEND), + SANDBOX_BACKEND: sandboxBackend, /** * Runtime session affinity for stateful sandbox backends. * - `stateless` (default): no runtime sessions; `runtime_session_hint` ignored. @@ -353,7 +357,17 @@ export const env = { * - `strict`: same serialized session semantics, and a session hint is * required instead of degrading requests without one to stateless. */ - RUNTIME_SESSION_MODE: resolveRuntimeSessionMode(process.env.CODEAPI_RUNTIME_SESSION_MODE), + RUNTIME_SESSION_MODE: runtimeSessionMode, + /** + * Deployment identity used by trusted callers to route each agent to the + * intended execution stack. `default` is HTTP/stateless; `stateful` is + * Lambda MicroVM with session affinity. The startup policy rejects mixed + * tuples so an endpoint cannot claim one profile while running the other. + */ + EXECUTION_PROFILE: resolveExecutionProfile( + process.env.CODEAPI_EXECUTION_PROFILE, + runtimeSessionMode, + ), RUNTIME_SESSION_LOCK_WAIT_MS: configuredNumber( process.env.CODEAPI_RUNTIME_SESSION_LOCK_WAIT_MS, 15_000, diff --git a/service/src/enum/service.ts b/service/src/enum/service.ts index 84e8924..b936404 100644 --- a/service/src/enum/service.ts +++ b/service/src/enum/service.ts @@ -2,11 +2,6 @@ export enum Jobs { execute = 'execute', } -export enum Queues { - python = 'python-queue', - other = 'other-queue', -} - export enum Languages { bash = 'bash', js = 'js', diff --git a/service/src/execution-profile.test.ts b/service/src/execution-profile.test.ts new file mode 100644 index 0000000..32b1f6a --- /dev/null +++ b/service/src/execution-profile.test.ts @@ -0,0 +1,83 @@ +import { describe, expect, test } from 'bun:test'; +import { + checkExecutionProfileExpectation, + queueNamesForExecutionProfile, + resolveExecutionProfile, +} from './execution-profile'; + +describe('execution profile resolution', () => { + test('preserves the HTTP/stateless default when unset', () => { + expect(resolveExecutionProfile(undefined, 'stateless')).toBe('default'); + }); + + test('recognizes an existing stateful deployment when unset', () => { + expect(resolveExecutionProfile(undefined, 'affinity')).toBe('stateful'); + expect(resolveExecutionProfile(undefined, 'strict')).toBe('stateful'); + }); + + test('lets API-only pods infer stateful from session mode without worker config', () => { + expect(resolveExecutionProfile(undefined, 'affinity')).toBe('stateful'); + }); + + test('accepts only the two public execution profiles', () => { + expect(resolveExecutionProfile('default', 'stateless')).toBe('default'); + expect(resolveExecutionProfile('stateful', 'affinity')).toBe('stateful'); + expect(() => resolveExecutionProfile('lambda', 'affinity')).toThrow( + 'CODEAPI_EXECUTION_PROFILE must be one of: default, stateful', + ); + expect(() => resolveExecutionProfile('', 'stateless')).toThrow( + 'CODEAPI_EXECUTION_PROFILE', + ); + }); +}); + +describe('execution profile queue isolation', () => { + test('keeps the legacy queue names for the default profile', () => { + expect(queueNamesForExecutionProfile('default')).toEqual({ + python: 'python-queue', + other: 'other-queue', + }); + }); + + test('uses separate queues for stateful workers', () => { + expect(queueNamesForExecutionProfile('stateful')).toEqual({ + python: 'stateful-python-queue', + other: 'stateful-other-queue', + }); + }); +}); + +describe('execution profile request assertion', () => { + test('allows callers that omit the assertion for backwards compatibility', () => { + expect(checkExecutionProfileExpectation(undefined, 'default')).toEqual({ ok: true }); + }); + + test('allows a matching expected profile', () => { + expect(checkExecutionProfileExpectation('stateful', 'stateful')).toEqual({ ok: true }); + }); + + test('returns a typed conflict before a mismatched request can be routed', () => { + expect(checkExecutionProfileExpectation('stateful', 'default')).toEqual({ + ok: false, + status: 409, + body: { + error: 'Expected the stateful execution profile, but reached default', + code: 'execution_profile_mismatch', + expected_profile: 'stateful', + actual_profile: 'default', + }, + }); + }); + + test('rejects invalid profile names instead of treating them as mismatches', () => { + expect(checkExecutionProfileExpectation('aws', 'default')).toMatchObject({ + ok: false, + status: 400, + body: { + code: 'invalid_execution_profile', + expected_profile: 'aws', + actual_profile: 'default', + }, + }); + }); +}); diff --git a/service/src/execution-profile.ts b/service/src/execution-profile.ts new file mode 100644 index 0000000..177acc2 --- /dev/null +++ b/service/src/execution-profile.ts @@ -0,0 +1,97 @@ +export const EXECUTION_PROFILES = ['default', 'stateful'] as const; + +export type ExecutionProfile = typeof EXECUTION_PROFILES[number]; + +export const EXPECTED_EXECUTION_PROFILE_HEADER = 'X-CodeAPI-Expected-Profile'; +export const EXECUTION_PROFILE_HEADER = 'X-CodeAPI-Execution-Profile'; + +export interface ExecutionProfileQueueNames { + python: string; + other: string; +} + +export function resolveExecutionProfile( + raw: string | undefined, + runtimeSessionMode: 'stateless' | 'affinity' | 'strict', +): ExecutionProfile { + if (raw != null) { + if (EXECUTION_PROFILES.includes(raw as ExecutionProfile)) { + return raw as ExecutionProfile; + } + throw new Error( + `CODEAPI_EXECUTION_PROFILE must be one of: ${EXECUTION_PROFILES.join(', ')}`, + ); + } + + /* Preserve the two supported pre-profile deployments during rollout. The + * common stateless stack remains `default`; a stateful API-only pod can + * infer `stateful` from its session mode even though worker-only backend + * credentials/config are intentionally absent. Worker startup separately + * verifies that this profile is backed by Lambda. */ + return runtimeSessionMode !== 'stateless' + ? 'stateful' + : 'default'; +} + +export function queueNamesForExecutionProfile( + profile: ExecutionProfile, +): ExecutionProfileQueueNames { + if (profile === 'stateful') { + return { + python: 'stateful-python-queue', + other: 'stateful-other-queue', + }; + } + return { + python: 'python-queue', + other: 'other-queue', + }; +} + +export type ExecutionProfileExpectation = + | { ok: true } + | { + ok: false; + status: 400 | 409; + body: { + error: string; + code: 'invalid_execution_profile' | 'execution_profile_mismatch'; + expected_profile?: string; + actual_profile: ExecutionProfile; + }; + }; + +export function checkExecutionProfileExpectation( + rawExpectedProfile: string | undefined, + actualProfile: ExecutionProfile, +): ExecutionProfileExpectation { + if (rawExpectedProfile == null) return { ok: true }; + + if (!EXECUTION_PROFILES.includes(rawExpectedProfile as ExecutionProfile)) { + return { + ok: false, + status: 400, + body: { + error: `Invalid execution profile: ${rawExpectedProfile}`, + code: 'invalid_execution_profile', + expected_profile: rawExpectedProfile, + actual_profile: actualProfile, + }, + }; + } + + if (rawExpectedProfile !== actualProfile) { + return { + ok: false, + status: 409, + body: { + error: `Expected the ${rawExpectedProfile} execution profile, but reached ${actualProfile}`, + code: 'execution_profile_mismatch', + expected_profile: rawExpectedProfile, + actual_profile: actualProfile, + }, + }; + } + + return { ok: true }; +} diff --git a/service/src/lifecycle.ts b/service/src/lifecycle.ts index 59a18a8..e27846e 100644 --- a/service/src/lifecycle.ts +++ b/service/src/lifecycle.ts @@ -3,7 +3,12 @@ import type { Express } from 'express'; import { pyQueue, otherQueue, pyQueueEvents, otherQueueEvents, connection } from './queue'; import { validateStartupAuthConfig } from './auth/startup'; import { env } from './config'; -import { validateApiHardenedConfig, validateSandboxBackendPolicy, validateWorkerHardenedConfig } from './secure-startup'; +import { + validateApiHardenedConfig, + validateExecutionProfilePolicy, + validateSandboxBackendPolicy, + validateWorkerHardenedConfig, +} from './secure-startup'; import logger from './logger'; import { shutdownTelemetry } from './telemetry'; @@ -75,6 +80,7 @@ function setupQueueListeners(queue: Queue, name: string): void { export async function startupApiOnly(): Promise { logger.info('Starting API service (no workers)...'); validateApiHardenedConfig(); + validateExecutionProfilePolicy({ requireBackendMatch: false }); /* No validateSandboxBackendPolicy() here: an API-only pod authenticates and * enqueues jobs, it never constructs the Lambda backend or checkpoint store. * Validating that policy would force worker-only config (LAMBDA_MICROVM_* and @@ -97,6 +103,7 @@ export async function startupApiOnly(): Promise { export async function startupWorkerOnly(): Promise { logger.info('Starting Worker service...'); validateWorkerHardenedConfig(); + validateExecutionProfilePolicy(); validateSandboxBackendPolicy(); // Dynamically import workers to start them @@ -131,6 +138,7 @@ async function gracefulStartup(): Promise { logger.info('Starting up service (combined API + Workers)...'); validateApiHardenedConfig(); validateWorkerHardenedConfig(); + validateExecutionProfilePolicy(); validateSandboxBackendPolicy(); await validateLifecycleAuthConfig(); diff --git a/service/src/local-api.ts b/service/src/local-api.ts index b9b280d..7428346 100644 --- a/service/src/local-api.ts +++ b/service/src/local-api.ts @@ -11,6 +11,7 @@ import express, { json, Router } from 'express'; import serviceRouter from './service/router'; import programmaticRouter from './service/programmatic-router'; import { requestErrorLogger, requestNotFoundLogger } from './middleware/request-error-logger'; +import { executionProfileMiddleware } from './middleware/execution-profile'; import { localAuth } from './auth/local'; import { pyQueue, otherQueue, pyQueueEvents, otherQueueEvents, connection } from './queue'; import { setStartupComplete } from './lifecycle'; @@ -19,10 +20,12 @@ import './workers'; import { env } from './config'; import logger from './logger'; import { shutdownTelemetry, traceHttpRequest } from './telemetry'; +import { validateExecutionProfilePolicy } from './secure-startup'; const app = express(); app.disable('x-powered-by'); app.set('trust proxy', 1); +app.use(executionProfileMiddleware); let localShuttingDown = false; const v1 = Router(); @@ -52,6 +55,7 @@ app.use(requestErrorLogger); async function localStartup(): Promise { logger.info('Starting local development server...'); logger.info('⚠️ LOCAL MODE - No authentication required'); + validateExecutionProfilePolicy(); try { // Set a local user ID for session management diff --git a/service/src/metrics.ts b/service/src/metrics.ts index ba052dc..a2d9865 100644 --- a/service/src/metrics.ts +++ b/service/src/metrics.ts @@ -1,8 +1,27 @@ import client, { register, Counter, Histogram, Gauge } from 'prom-client'; import { normalizeMetricPath } from './httpPathNormalize'; +import { env } from './config'; client.collectDefaultMetrics({ register }); +export const executionProfileInfo = new Gauge({ + name: 'codeapi_execution_profile_info', + help: 'Static identity of this Code API execution deployment', + labelNames: ['profile', 'sandbox_backend', 'runtime_session_mode'] as const, +}); + +executionProfileInfo.set({ + profile: env.EXECUTION_PROFILE, + sandbox_backend: env.SANDBOX_BACKEND, + runtime_session_mode: env.RUNTIME_SESSION_MODE, +}, 1); + +export const executionProfileRequestRejections = new Counter({ + name: 'codeapi_execution_profile_request_rejections_total', + help: 'Requests rejected because the expected execution profile was invalid or mismatched', + labelNames: ['reason'] as const, +}); + // -- HTTP metrics (shared across Express and Bun servers) -- export const httpRequestsTotal = new Counter({ name: 'codeapi_http_requests_total', diff --git a/service/src/middleware/execution-profile.test.ts b/service/src/middleware/execution-profile.test.ts new file mode 100644 index 0000000..43368ef --- /dev/null +++ b/service/src/middleware/execution-profile.test.ts @@ -0,0 +1,78 @@ +import { afterEach, describe, expect, test } from 'bun:test'; +import type { NextFunction, Request, Response } from 'express'; +import { env } from '../config'; +import { executionProfileMiddleware } from './execution-profile'; + +const savedProfile = env.EXECUTION_PROFILE; + +afterEach(() => { + env.EXECUTION_PROFILE = savedProfile; +}); + +function invoke(expectedProfile?: string): { + headers: Record; + status?: number; + body?: unknown; + nextCalled: boolean; +} { + const result: { + headers: Record; + status?: number; + body?: unknown; + nextCalled: boolean; + } = { headers: {}, nextCalled: false }; + const req = { + get: () => expectedProfile, + } as unknown as Request; + const res = { + setHeader: (name: string, value: string) => { + result.headers[name] = value; + }, + status: (status: number) => { + result.status = status; + return res; + }, + json: (body: unknown) => { + result.body = body; + return res; + }, + } as unknown as Response; + const next = (() => { + result.nextCalled = true; + }) as NextFunction; + + executionProfileMiddleware(req, res, next); + return result; +} + +describe('execution profile middleware', () => { + test('advertises the actual profile and allows matching requests', () => { + env.EXECUTION_PROFILE = 'stateful'; + expect(invoke('stateful')).toEqual({ + headers: { 'X-CodeAPI-Execution-Profile': 'stateful' }, + nextCalled: true, + }); + }); + + test('rejects a mismatched endpoint before routing', () => { + env.EXECUTION_PROFILE = 'default'; + expect(invoke('stateful')).toMatchObject({ + headers: { 'X-CodeAPI-Execution-Profile': 'default' }, + status: 409, + body: { + code: 'execution_profile_mismatch', + expected_profile: 'stateful', + actual_profile: 'default', + }, + nextCalled: false, + }); + }); + + test('keeps older callers working when they omit the assertion', () => { + env.EXECUTION_PROFILE = 'default'; + expect(invoke()).toEqual({ + headers: { 'X-CodeAPI-Execution-Profile': 'default' }, + nextCalled: true, + }); + }); +}); diff --git a/service/src/middleware/execution-profile.ts b/service/src/middleware/execution-profile.ts new file mode 100644 index 0000000..2e3bdae --- /dev/null +++ b/service/src/middleware/execution-profile.ts @@ -0,0 +1,37 @@ +import type { NextFunction, Request, Response } from 'express'; +import { env } from '../config'; +import { + checkExecutionProfileExpectation, + EXECUTION_PROFILE_HEADER, + EXPECTED_EXECUTION_PROFILE_HEADER, +} from '../execution-profile'; +import { executionProfileRequestRejections } from '../metrics'; + +/** + * Advertise this deployment's profile and fail closed when a trusted caller + * reaches the wrong endpoint. Apply before routing so no file, programmatic, + * or ordinary execution request can enqueue work on a mismatched stack. + */ +export function executionProfileMiddleware( + req: Request, + res: Response, + next: NextFunction, +): void { + res.setHeader(EXECUTION_PROFILE_HEADER, env.EXECUTION_PROFILE); + + const expectation = checkExecutionProfileExpectation( + req.get(EXPECTED_EXECUTION_PROFILE_HEADER), + env.EXECUTION_PROFILE, + ); + if (expectation.ok) { + next(); + return; + } + + executionProfileRequestRejections.inc({ + reason: expectation.body.code === 'execution_profile_mismatch' + ? 'mismatch' + : 'invalid', + }); + res.status(expectation.status).json(expectation.body); +} diff --git a/service/src/queue.ts b/service/src/queue.ts index c6f3226..3410968 100644 --- a/service/src/queue.ts +++ b/service/src/queue.ts @@ -5,8 +5,9 @@ import { setMaxListeners } from 'events'; import type { CommonRedisOptions } from 'ioredis'; import type * as tls from 'tls'; import type * as t from './types'; -import { Jobs, Queues } from './enum'; +import { Jobs } from './enum'; import { env } from './config'; +import { queueNamesForExecutionProfile } from './execution-profile'; import logger from './logger'; import { redisKeepAliveOptions } from './redis-options'; import { bullmqQueueJobs, registerBullmqQueueMetricsCollector } from './metrics'; @@ -54,16 +55,19 @@ const connection = new IORedis({ // Global queues - no INSTANCE_ID prefix // This enables horizontal scaling where any worker can process any job -const pyQueue = new Queue(Queues.python, { connection }); -const otherQueue = new Queue(Queues.other, { connection }); +// while the execution-profile prefix prevents HTTP and Lambda workers from +// consuming each other's jobs when they share Redis. +const queueNames = queueNamesForExecutionProfile(env.EXECUTION_PROFILE); +const pyQueue = new Queue(queueNames.python, { connection }); +const otherQueue = new Queue(queueNames.other, { connection }); -const pyQueueEvents = new QueueEvents(Queues.python, { connection }); -const otherQueueEvents = new QueueEvents(Queues.other, { connection }); +const pyQueueEvents = new QueueEvents(queueNames.python, { connection }); +const otherQueueEvents = new QueueEvents(queueNames.other, { connection }); const queueMetricStates = ['waiting', 'active', 'delayed'] as const; const queueMetricSources = [ - { name: Queues.python, queue: pyQueue }, - { name: Queues.other, queue: otherQueue }, + { name: queueNames.python, queue: pyQueue }, + { name: queueNames.other, queue: otherQueue }, ] as const; const QUEUE_METRICS_TIMEOUT_MS = 1000; @@ -109,4 +113,4 @@ registerBullmqQueueMetricsCollector(async () => { * BullMQ coordination objects. */ setMaxListeners(0, pyQueue, otherQueue, pyQueueEvents, otherQueueEvents); -export { pyQueue, otherQueue, pyQueueEvents, otherQueueEvents, connection }; +export { pyQueue, otherQueue, pyQueueEvents, otherQueueEvents, queueNames, connection }; diff --git a/service/src/secure-startup.test.ts b/service/src/secure-startup.test.ts index 720809b..7e94960 100644 --- a/service/src/secure-startup.test.ts +++ b/service/src/secure-startup.test.ts @@ -3,6 +3,7 @@ import { env } from './config'; import { validateApiHardenedConfig, validateEgressGatewayHardenedConfig, + validateExecutionProfilePolicy, validateSandboxBackendPolicy, validateWorkerHardenedConfig, } from './secure-startup'; @@ -10,6 +11,7 @@ import { const savedEnv = { ...process.env }; const saved = { hardened: env.HARDENED_SANDBOX_MODE, + executionProfile: env.EXECUTION_PROFILE, sandboxBackend: env.SANDBOX_BACKEND, ptcMode: env.PTC_MODE, runtimeSessionMode: env.RUNTIME_SESSION_MODE, @@ -46,6 +48,7 @@ function restore(): void { } Object.assign(process.env, savedEnv); env.HARDENED_SANDBOX_MODE = saved.hardened; + env.EXECUTION_PROFILE = saved.executionProfile; env.SANDBOX_BACKEND = saved.sandboxBackend; env.PTC_MODE = saved.ptcMode; env.RUNTIME_SESSION_MODE = saved.runtimeSessionMode; @@ -78,6 +81,61 @@ function restore(): void { afterEach(restore); +describe('execution profile policy', () => { + test('accepts the AWS-free default profile', () => { + env.EXECUTION_PROFILE = 'default'; + env.SANDBOX_BACKEND = 'http'; + env.RUNTIME_SESSION_MODE = 'stateless'; + expect(() => validateExecutionProfilePolicy()).not.toThrow(); + }); + + test('accepts affinity and strict stateful profiles', () => { + env.EXECUTION_PROFILE = 'stateful'; + env.SANDBOX_BACKEND = 'lambda-microvm'; + env.RUNTIME_SESSION_MODE = 'affinity'; + expect(() => validateExecutionProfilePolicy()).not.toThrow(); + env.RUNTIME_SESSION_MODE = 'strict'; + expect(() => validateExecutionProfilePolicy()).not.toThrow(); + }); + + test('does not require worker-only backend config on API-only pods', () => { + env.EXECUTION_PROFILE = 'stateful'; + env.SANDBOX_BACKEND = 'http'; + env.RUNTIME_SESSION_MODE = 'affinity'; + expect(() => validateExecutionProfilePolicy({ requireBackendMatch: false })).not.toThrow(); + }); + + test('rejects a default profile backed by AWS or stateful sessions', () => { + env.EXECUTION_PROFILE = 'default'; + env.SANDBOX_BACKEND = 'lambda-microvm'; + env.RUNTIME_SESSION_MODE = 'stateless'; + expect(() => validateExecutionProfilePolicy()).toThrow( + 'CODEAPI_EXECUTION_PROFILE=default requires', + ); + + env.SANDBOX_BACKEND = 'http'; + env.RUNTIME_SESSION_MODE = 'affinity'; + expect(() => validateExecutionProfilePolicy()).toThrow( + 'CODEAPI_EXECUTION_PROFILE=default requires', + ); + }); + + test('rejects a stateful profile without Lambda affinity', () => { + env.EXECUTION_PROFILE = 'stateful'; + env.SANDBOX_BACKEND = 'http'; + env.RUNTIME_SESSION_MODE = 'affinity'; + expect(() => validateExecutionProfilePolicy()).toThrow( + 'CODEAPI_EXECUTION_PROFILE=stateful requires', + ); + + env.SANDBOX_BACKEND = 'lambda-microvm'; + env.RUNTIME_SESSION_MODE = 'stateless'; + expect(() => validateExecutionProfilePolicy()).toThrow( + 'CODEAPI_EXECUTION_PROFILE=stateful requires', + ); + }); +}); + describe('hardened CodeAPI startup config', () => { test('rejects grant secrets in API and worker processes', () => { env.HARDENED_SANDBOX_MODE = true; diff --git a/service/src/secure-startup.ts b/service/src/secure-startup.ts index 8d4729d..df9409e 100644 --- a/service/src/secure-startup.ts +++ b/service/src/secure-startup.ts @@ -63,6 +63,41 @@ export function validateWorkerHardenedConfig(): void { requireValue('CODEAPI_EXECUTION_MANIFEST_PRIVATE_KEY', env.EXECUTION_MANIFEST_PRIVATE_KEY); } +/** + * Make the endpoint identity trustworthy. Callers route by execution profile, + * so accepting a contradictory backend/session tuple would silently send work + * to the wrong infrastructure and could lose workspace continuity. + */ +export function validateExecutionProfilePolicy(options: { + requireBackendMatch?: boolean; +} = {}): void { + const requireBackendMatch = options.requireBackendMatch ?? true; + if (env.EXECUTION_PROFILE === 'default') { + if ( + env.RUNTIME_SESSION_MODE !== 'stateless' + || (requireBackendMatch && env.SANDBOX_BACKEND !== 'http') + ) { + throw new SecureStartupConfigError( + 'CODEAPI_EXECUTION_PROFILE=default requires ' + + (requireBackendMatch ? 'CODEAPI_SANDBOX_BACKEND=http and ' : '') + + 'CODEAPI_RUNTIME_SESSION_MODE=stateless', + ); + } + return; + } + + if ( + env.RUNTIME_SESSION_MODE === 'stateless' + || (requireBackendMatch && env.SANDBOX_BACKEND !== 'lambda-microvm') + ) { + throw new SecureStartupConfigError( + 'CODEAPI_EXECUTION_PROFILE=stateful requires ' + + (requireBackendMatch ? 'CODEAPI_SANDBOX_BACKEND=lambda-microvm and ' : '') + + 'CODEAPI_RUNTIME_SESSION_MODE=affinity or strict', + ); + } +} + /** * Backend-selection policy. Unlike the hardened-mode validators, this runs * unconditionally: a misconfigured backend must never half-start. diff --git a/service/src/service-api.ts b/service/src/service-api.ts index c664f14..ec15b1a 100644 --- a/service/src/service-api.ts +++ b/service/src/service-api.ts @@ -2,6 +2,7 @@ import express, { json, Router } from 'express'; import { startServer, gracefulShutdown } from './lifecycle'; import { apiKeyAuth } from './middleware/auth'; import { requestErrorLogger, requestNotFoundLogger } from './middleware/request-error-logger'; +import { executionProfileMiddleware } from './middleware/execution-profile'; import serviceRouter from './service/router'; import programmaticRouter from './service/programmatic-router'; import { connection } from './queue'; @@ -11,6 +12,7 @@ import logger from './logger'; const app = express(); app.disable('x-powered-by'); app.set('trust proxy', 1); +app.use(executionProfileMiddleware); const v1 = Router(); diff --git a/service/src/service/router.ts b/service/src/service/router.ts index d88dfdc..4c844fb 100644 --- a/service/src/service/router.ts +++ b/service/src/service/router.ts @@ -10,7 +10,7 @@ import { sessionAuth } from '../middleware/auth'; import { executionLimiter, uploadLimiter, downloadLimiter, fetchLimiter } from '../middleware/limits'; import { internalServiceHeaders } from '../internal-service-auth'; import { resolveSessionKey, resolveOutputBucketSessionKey, SessionKeyResolutionError, parseUploadSessionKeyInput, type SessionKeyInput } from '../session-key'; -import { pyQueue, otherQueue, pyQueueEvents, otherQueueEvents, connection } from '../queue'; +import { pyQueue, otherQueue, pyQueueEvents, otherQueueEvents, queueNames, connection } from '../queue'; import { sleep, getAxiosErrorDetails, publicExecutionFailure } from '../utils'; import { env, jobCompletionWaitTimeoutMs, planLimits, resolveLanguage } from '../config'; import { createPayload } from '../payload'; @@ -226,12 +226,13 @@ router.post('/exec', executionLimiter, async (req: t.AuthenticatedRequest, res) const queue = language === Languages.py ? pyQueue : otherQueue; const queueEvents = language === Languages.py ? pyQueueEvents : otherQueueEvents; - const queueName = language === Languages.py ? 'python' : 'other'; + const queueName = language === Languages.py ? queueNames.python : queueNames.other; const job = await withSpan('codeapi.job.enqueue', { 'messaging.system': 'bullmq', 'messaging.destination.name': queueName, 'codeapi.language': language, + 'codeapi.execution_profile': env.EXECUTION_PROFILE, }, () => { const traceCarrier = captureTraceCarrier(); return queue.add(Jobs.execute, { @@ -279,6 +280,7 @@ router.post('/exec', executionLimiter, async (req: t.AuthenticatedRequest, res) 'messaging.system': 'bullmq', 'messaging.destination.name': queueName, 'codeapi.language': language, + 'codeapi.execution_profile': env.EXECUTION_PROFILE, }, () => job.waitUntilFinished(queueEvents, JOB_COMPLETION_WAIT_TIMEOUT_MS), 'CONSUMER'); if (!isSyntheticRequest) { diff --git a/service/src/workers.ts b/service/src/workers.ts index e9c3b2b..206542b 100644 --- a/service/src/workers.ts +++ b/service/src/workers.ts @@ -3,8 +3,7 @@ import { Worker } from 'bullmq'; import type * as t from './types'; import { filterSystemLogs, applySystemReplacements, getAxiosErrorDetails, sandboxErrorMessageFromAxios } from './utils'; import { jobProcessingDuration, jobsCompleted, jobsFailed, activeJobs, workerRunning } from './metrics'; -import { Queues } from './enum'; -import { connection } from './queue'; +import { connection, queueNames } from './queue'; import { env, jobDeadlineAtMs } from './config'; import { summarizeSandboxResponse, summarizeText } from './execution-log'; import { createGatewayEgressGrant, restoreGatewaySandboxResult, revokeGatewayEgressGrant } from './egress-gateway-client'; @@ -238,7 +237,7 @@ async function processJobInner(job: t.ExecuteJob): Promise { // Global workers - no INSTANCE_ID prefix // This enables horizontal scaling where any worker can process any job from the shared queue // Each worker respects its own concurrency limit based on its co-located sandbox capacity -export const pyWorker = new Worker(Queues.python, processJob, { +export const pyWorker = new Worker(queueNames.python, processJob, { connection, concurrency: env.PYTHON_CONCURRENCY, limiter: { @@ -247,7 +246,7 @@ export const pyWorker = new Worker(Queues.python, processJob, { }, }); -export const otherWorker = new Worker(Queues.other, processJob, { +export const otherWorker = new Worker(queueNames.other, processJob, { connection, concurrency: env.OTHER_CONCURRENCY, limiter: { From 33b5f8836d40b51e704d98a308d037b5978f6a6b Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Sun, 16 Aug 2026 11:40:25 -0400 Subject: [PATCH 2/3] fix: preserve queue compatibility during profile rollout --- README.md | 11 +++++++++++ docs/lambda-microvm/README.md | 2 +- helm/codeapi/README.md | 11 +++++++++++ service/src/execution-profile.test.ts | 13 ++++++++++--- service/src/execution-profile.ts | 9 ++++++++- service/src/queue.ts | 5 ++++- 6 files changed, 45 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index f848b1d..a53efc0 100644 --- a/README.md +++ b/README.md @@ -37,6 +37,17 @@ workers. The default profile keeps the existing `python-queue` and `stateful-other-queue`. This allows both deployments to share Redis without cross-consuming jobs. +An affinity/strict deployment upgraded from a pre-profile release may leave +`CODEAPI_EXECUTION_PROFILE` unset for its first binary rollout. It still +identifies itself as `stateful`, but temporarily keeps the legacy queue names +so separately deployed APIs and workers remain compatible with old binaries. +Move that deployment to the isolated stateful queues with a blue/green cutover: +start replacement API and worker pools with the profile explicitly set to +`stateful`, verify them together, switch the stateful endpoint, and drain the +legacy pool. For rollback, switch the endpoint back before stopping the +replacement pool. Do not run the inferred stateful compatibility mode beside a +default deployment on the same Redis because both use the legacy queues. + Trusted callers should send `X-CodeAPI-Expected-Profile: default|stateful` on every Code API request. A request that reaches the wrong deployment fails before enqueue with HTTP 409 and `code=execution_profile_mismatch`; every diff --git a/docs/lambda-microvm/README.md b/docs/lambda-microvm/README.md index 4bd3e87..b6c64a5 100644 --- a/docs/lambda-microvm/README.md +++ b/docs/lambda-microvm/README.md @@ -311,7 +311,7 @@ appear in `api/src/config.ts`. | Env | Default | Meaning | |---|---|---| | `CODEAPI_SANDBOX_BACKEND` | `http` | `http` (byte-identical to today) or `lambda-microvm`. | -| `CODEAPI_EXECUTION_PROFILE` | inferred | `default` for the HTTP/stateless deployment or `stateful` for the Lambda affinity/strict deployment. Stateful API and worker processes consume isolated BullMQ queues. | +| `CODEAPI_EXECUTION_PROFILE` | inferred | `default` for the HTTP/stateless deployment or `stateful` for the Lambda affinity/strict deployment. An explicit `stateful` value selects isolated BullMQ queues. An inferred affinity/strict profile keeps the legacy queues only for a pre-profile binary rollout and must not share Redis with the default deployment. | | `CODEAPI_RUNTIME_SESSION_MODE` | `stateless` | `stateless` \| `affinity` \| `strict`. `affinity` and `strict` require the `lambda-microvm` backend. See [Operating modes](#operating-modes). | | `CODEAPI_RUNTIME_SESSION_LOCK_WAIT_MS` | `15000` | How long a stateful execution waits for the session lock before returning `RUNTIME_SESSION_BUSY` (HTTP 409). | diff --git a/helm/codeapi/README.md b/helm/codeapi/README.md index 83cf2a2..37ccd6a 100644 --- a/helm/codeapi/README.md +++ b/helm/codeapi/README.md @@ -62,6 +62,17 @@ then consumes `stateful-python-queue` / `stateful-other-queue`, so both stacks may safely share Redis without consuming each other's jobs. Do not mix API and worker profile values within one deployment. +For an existing affinity/strict deployment from before execution profiles, +first roll the new binary to API and worker pods with +`CODEAPI_EXECUTION_PROFILE` still unset. The inferred stateful compatibility +mode deliberately retains the legacy queues, so old and new binaries can +overlap. Then create a replacement deployment with the profile explicitly set +to `stateful`, verify its API and workers together, switch the stateful ingress, +and drain the legacy deployment. Roll back by switching ingress to the legacy +deployment before removing the replacement. Never share Redis between the +inferred compatibility deployment and a default deployment: both consume the +legacy queues. + **Authentication.** Outside local mode the API verifies JWTs. Configure the verifier through environment variables on the api component, e.g.: diff --git a/service/src/execution-profile.test.ts b/service/src/execution-profile.test.ts index 32b1f6a..c086a02 100644 --- a/service/src/execution-profile.test.ts +++ b/service/src/execution-profile.test.ts @@ -33,14 +33,21 @@ describe('execution profile resolution', () => { describe('execution profile queue isolation', () => { test('keeps the legacy queue names for the default profile', () => { - expect(queueNamesForExecutionProfile('default')).toEqual({ + expect(queueNamesForExecutionProfile('default', 'explicit')).toEqual({ python: 'python-queue', other: 'other-queue', }); }); - test('uses separate queues for stateful workers', () => { - expect(queueNamesForExecutionProfile('stateful')).toEqual({ + test('keeps inferred stateful deployments on legacy queues during binary rollout', () => { + expect(queueNamesForExecutionProfile('stateful', 'inferred')).toEqual({ + python: 'python-queue', + other: 'other-queue', + }); + }); + + test('uses separate queues only for an explicit stateful deployment', () => { + expect(queueNamesForExecutionProfile('stateful', 'explicit')).toEqual({ python: 'stateful-python-queue', other: 'stateful-other-queue', }); diff --git a/service/src/execution-profile.ts b/service/src/execution-profile.ts index 177acc2..354e04c 100644 --- a/service/src/execution-profile.ts +++ b/service/src/execution-profile.ts @@ -35,8 +35,15 @@ export function resolveExecutionProfile( export function queueNamesForExecutionProfile( profile: ExecutionProfile, + source: 'explicit' | 'inferred', ): ExecutionProfileQueueNames { - if (profile === 'stateful') { + /* A pre-profile affinity/strict deployment used the legacy queues. Keep + * inferred profiles on those names so API and worker Deployments can roll + * or roll back independently without temporarily losing their consumers. + * Queue isolation is an explicit cutover: operators bring up the stateful + * stack with CODEAPI_EXECUTION_PROFILE=stateful on both sides, then switch + * callers to its endpoint. */ + if (profile === 'stateful' && source === 'explicit') { return { python: 'stateful-python-queue', other: 'stateful-other-queue', diff --git a/service/src/queue.ts b/service/src/queue.ts index 3410968..a93ea81 100644 --- a/service/src/queue.ts +++ b/service/src/queue.ts @@ -57,7 +57,10 @@ const connection = new IORedis({ // This enables horizontal scaling where any worker can process any job // while the execution-profile prefix prevents HTTP and Lambda workers from // consuming each other's jobs when they share Redis. -const queueNames = queueNamesForExecutionProfile(env.EXECUTION_PROFILE); +const queueNames = queueNamesForExecutionProfile( + env.EXECUTION_PROFILE, + process.env.CODEAPI_EXECUTION_PROFILE == null ? 'inferred' : 'explicit', +); const pyQueue = new Queue(queueNames.python, { connection }); const otherQueue = new Queue(queueNames.other, { connection }); From 04469d1144f3db334f5bc67bb019bc8f80b37be9 Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Sun, 16 Aug 2026 16:38:08 -0400 Subject: [PATCH 3/3] fix: harden execution profile upgrades --- README.md | 12 ++-- docs/lambda-microvm/README.md | 10 +-- helm/codeapi/README.md | 44 +++++++++--- helm/codeapi/templates/api-deployment.yaml | 4 +- .../templates/worker-sandbox-deployment.yaml | 4 +- helm/codeapi/values.yaml | 9 +-- service/openapi.yml | 50 +++++++------ service/src/config.ts | 8 ++- service/src/execution-profile.test.ts | 31 ++++++-- service/src/execution-profile.ts | 71 +++++++++++++------ service/src/lifecycle.ts | 12 ++++ service/src/local-api.ts | 8 ++- service/src/metrics.test.ts | 18 +++++ service/src/metrics.ts | 21 ++++-- .../src/middleware/execution-profile.test.ts | 21 +++++- service/src/middleware/execution-profile.ts | 19 ++++- service/src/queue.ts | 2 +- service/src/secure-startup.test.ts | 15 ++++ service/src/secure-startup.ts | 7 +- service/src/service/programmatic-router.ts | 2 + service/src/service/router.ts | 1 + service/src/types/service.ts | 3 + service/src/workers.ts | 4 ++ 23 files changed, 292 insertions(+), 84 deletions(-) diff --git a/README.md b/README.md index a53efc0..6612cb5 100644 --- a/README.md +++ b/README.md @@ -37,10 +37,12 @@ workers. The default profile keeps the existing `python-queue` and `stateful-other-queue`. This allows both deployments to share Redis without cross-consuming jobs. -An affinity/strict deployment upgraded from a pre-profile release may leave -`CODEAPI_EXECUTION_PROFILE` unset for its first binary rollout. It still -identifies itself as `stateful`, but temporarily keeps the legacy queue names -so separately deployed APIs and workers remain compatible with old binaries. +An existing Lambda MicroVM deployment upgraded from a pre-profile release may +leave `CODEAPI_EXECUTION_PROFILE` unset for its first binary rollout. An +affinity/strict deployment still identifies itself as `stateful`; a stateless +Lambda deployment identifies itself as `default`. Both temporarily keep the +legacy queue names so separately deployed APIs and workers remain compatible +with old binaries. Move that deployment to the isolated stateful queues with a blue/green cutover: start replacement API and worker pools with the profile explicitly set to `stateful`, verify them together, switch the stateful endpoint, and drain the @@ -50,7 +52,7 @@ default deployment on the same Redis because both use the legacy queues. Trusted callers should send `X-CodeAPI-Expected-Profile: default|stateful` on every Code API request. A request that reaches the wrong deployment fails -before enqueue with HTTP 409 and `code=execution_profile_mismatch`; every +before enqueue with HTTP 409 and `error=execution_profile_mismatch`; every response advertises the actual deployment in `X-CodeAPI-Execution-Profile`. Omitting the expected-profile header remains supported for older clients, but provides no wrong-endpoint protection. There is deliberately no silent diff --git a/docs/lambda-microvm/README.md b/docs/lambda-microvm/README.md index b6c64a5..309d6ec 100644 --- a/docs/lambda-microvm/README.md +++ b/docs/lambda-microvm/README.md @@ -311,7 +311,7 @@ appear in `api/src/config.ts`. | Env | Default | Meaning | |---|---|---| | `CODEAPI_SANDBOX_BACKEND` | `http` | `http` (byte-identical to today) or `lambda-microvm`. | -| `CODEAPI_EXECUTION_PROFILE` | inferred | `default` for the HTTP/stateless deployment or `stateful` for the Lambda affinity/strict deployment. An explicit `stateful` value selects isolated BullMQ queues. An inferred affinity/strict profile keeps the legacy queues only for a pre-profile binary rollout and must not share Redis with the default deployment. | +| `CODEAPI_EXECUTION_PROFILE` | inferred | `default` for the HTTP/stateless deployment or `stateful` for the Lambda affinity/strict deployment. An explicit `stateful` value selects isolated BullMQ queues. Inferred affinity/strict and legacy Lambda/stateless deployments keep the legacy queues only for a pre-profile binary rollout and must not share Redis with the default deployment. | | `CODEAPI_RUNTIME_SESSION_MODE` | `stateless` | `stateless` \| `affinity` \| `strict`. `affinity` and `strict` require the `lambda-microvm` backend. See [Operating modes](#operating-modes). | | `CODEAPI_RUNTIME_SESSION_LOCK_WAIT_MS` | `15000` | How long a stateful execution waits for the session lock before returning `RUNTIME_SESSION_BUSY` (HTTP 409). | @@ -410,9 +410,11 @@ You do not have to adopt the whole stack at once. The knobs compose: **No AWS at all.** Leave `CODEAPI_SANDBOX_BACKEND` unset (`http`). Today's behavior, no MicroVMs, no changes needed anywhere. -**MicroVM isolation without sessions.** `lambda-microvm` + `stateless`. Every -execution gets a fresh, strongly-isolated Firecracker VM. No registry, no -checkpoints, no session workspace. Simplest way to get the isolation boundary. +**MicroVM isolation without sessions.** `lambda-microvm` + `stateless`, with +`CODEAPI_EXECUTION_PROFILE` unset for compatibility. Every execution gets a +fresh, strongly-isolated Firecracker VM. No registry, no checkpoints, no +session workspace. This legacy profile uses the shared queue names and must +not share Redis with a separate default deployment. **Base container image and snapshot boundary.** The default runner uses a stock `oven/bun` base and is **hookless** — session mode arrives per request via the diff --git a/helm/codeapi/README.md b/helm/codeapi/README.md index 37ccd6a..9b3efbb 100644 --- a/helm/codeapi/README.md +++ b/helm/codeapi/README.md @@ -53,14 +53,42 @@ platform rather than templated here: external ingress/service mesh, KEDA-style queue-depth autoscaling, and cloud-IAM secret delivery (the env hooks below cover all of them). -**Execution profile.** This chart deploys the AWS-free `default` profile and -sets `CODEAPI_EXECUTION_PROFILE=default` on both API and worker pods. That -profile requires the HTTP sandbox backend in stateless mode and retains the -existing `python-queue` / `other-queue` BullMQ names. A separate stateful -Lambda MicroVM deployment must use `CODEAPI_EXECUTION_PROFILE=stateful`; it -then consumes `stateful-python-queue` / `stateful-other-queue`, so both stacks -may safely share Redis without consuming each other's jobs. Do not mix API -and worker profile values within one deployment. +**Execution profile.** By default this chart leaves +`CODEAPI_EXECUTION_PROFILE` unset. Its bundled HTTP/stateless configuration is +inferred as the AWS-free `default` profile and retains the existing +`python-queue` / `other-queue` BullMQ names. Set `executionProfile: default` +explicitly when deploying it beside a stateful stack. A separate stateful +Lambda MicroVM deployment must use `executionProfile: stateful`; it then +consumes `stateful-python-queue` / `stateful-other-queue`, so both stacks may +safely share Redis without consuming each other's jobs. Do not mix API and +worker profile values within one deployment. + +The chart does not provision Lambda MicroVM infrastructure. Supply its +runtime settings to both the API and worker (and AWS credentials or workload +identity to the worker) through the existing environment hooks, for example: + +```yaml +executionProfile: stateful +api: + extraEnv: + - name: CODEAPI_RUNTIME_SESSION_MODE + value: affinity +workerSandbox: + extraEnv: + - name: CODEAPI_SANDBOX_BACKEND + value: lambda-microvm + - name: CODEAPI_RUNTIME_SESSION_MODE + value: affinity + - name: LAMBDA_MICROVM_IMAGE_ARN + value: arn:aws:lambda:REGION:ACCOUNT:microvm-image:NAME + - name: LAMBDA_MICROVM_IMAGE_VERSION + value: "VERSION" +``` + +The worker also needs the remaining Lambda networking, checkpoint-store, and +hardening variables documented in `docs/lambda-microvm/README.md`. This chart +still renders its bundled sandbox-runner, though a Lambda worker does not call +it; a platform-specific stateful deployment may omit that component. For an existing affinity/strict deployment from before execution profiles, first roll the new binary to API and worker pods with diff --git a/helm/codeapi/templates/api-deployment.yaml b/helm/codeapi/templates/api-deployment.yaml index 7a62167..bf5f91e 100644 --- a/helm/codeapi/templates/api-deployment.yaml +++ b/helm/codeapi/templates/api-deployment.yaml @@ -41,8 +41,10 @@ spec: {{ include "codeapi.otel.env" (dict "root" . "serviceName" "aiml-codeapi-api") | nindent 12 }} - name: CODEAPI_HARDENED_SANDBOX_MODE value: {{ .Values.hardenedSandboxMode | quote }} + {{- with .Values.executionProfile }} - name: CODEAPI_EXECUTION_PROFILE - value: {{ .Values.executionProfile | quote }} + value: {{ . | quote }} + {{- end }} # Redis connection - name: REDIS_HOST value: {{ include "codeapi.redis.host" . }} diff --git a/helm/codeapi/templates/worker-sandbox-deployment.yaml b/helm/codeapi/templates/worker-sandbox-deployment.yaml index c1bea5e..39c2e19 100644 --- a/helm/codeapi/templates/worker-sandbox-deployment.yaml +++ b/helm/codeapi/templates/worker-sandbox-deployment.yaml @@ -147,8 +147,10 @@ spec: {{ include "codeapi.otel.env" (dict "root" . "serviceName" "aiml-codeapi-service-worker") | nindent 12 }} - name: CODEAPI_HARDENED_SANDBOX_MODE value: {{ .Values.hardenedSandboxMode | quote }} + {{- with .Values.executionProfile }} - name: CODEAPI_EXECUTION_PROFILE - value: {{ .Values.executionProfile | quote }} + value: {{ . | quote }} + {{- end }} - name: SANDBOX_ENDPOINT value: "http://{{ include "codeapi.fullname" . }}-sandbox-runner:{{ .Values.workerSandbox.sandbox.port }}/api/v2" - name: EGRESS_GATEWAY_URL diff --git a/helm/codeapi/values.yaml b/helm/codeapi/values.yaml index 0f9a225..75abaaf 100644 --- a/helm/codeapi/values.yaml +++ b/helm/codeapi/values.yaml @@ -22,10 +22,11 @@ internalServiceAuth: hardenedSandboxMode: true -# Stable identity advertised by this API/worker deployment. The bundled chart -# is the AWS-free HTTP/libkrun profile. A separate Lambda MicroVM deployment -# must set this to `stateful`; the service then uses isolated BullMQ queues. -executionProfile: default +# Stable identity advertised by this API/worker deployment. Leave empty for a +# backwards-compatible inferred profile. Set explicitly to `default` or +# `stateful` when deploying both stacks against shared Redis; explicit +# `stateful` selects isolated BullMQ queues. +executionProfile: "" otel: enabled: false diff --git a/service/openapi.yml b/service/openapi.yml index 16d36d7..913e780 100644 --- a/service/openapi.yml +++ b/service/openapi.yml @@ -40,24 +40,28 @@ components: enum: [default, stateful] responses: - InvalidExecutionProfile: - description: Invalid expected execution profile + BadRequest: + description: Invalid request or invalid expected execution profile headers: X-CodeAPI-Execution-Profile: $ref: '#/components/headers/ExecutionProfile' content: application/json: schema: - $ref: '#/components/schemas/ExecutionProfileError' - ExecutionProfileMismatch: - description: The request reached a different execution profile + anyOf: + - $ref: '#/components/schemas/Error' + - $ref: '#/components/schemas/ExecutionProfileError' + Conflict: + description: Request conflict or execution-profile mismatch headers: X-CodeAPI-Execution-Profile: $ref: '#/components/headers/ExecutionProfile' content: application/json: schema: - $ref: '#/components/schemas/ExecutionProfileError' + anyOf: + - $ref: '#/components/schemas/Error' + - $ref: '#/components/schemas/ExecutionProfileError' schemas: FileRef: @@ -152,10 +156,12 @@ components: $ref: '#/components/schemas/RequestFile' runtime_session_hint: type: string + maxLength: 128 + pattern: '^[A-Za-z0-9._:-]+$' description: >- Stable opaque hint for stateful runtime reuse. The server binds it - to the authenticated tenant and user. Ignored by the default - stateless profile. + to the authenticated tenant and user. Required in strict runtime + session mode and ignored by the default stateless profile. FileObject: type: object @@ -203,16 +209,18 @@ components: type: string details: type: string + message: + type: string ExecutionProfileError: type: object - required: [error, code, actual_profile] + required: [error, message, actual_profile] properties: error: type: string - code: - type: string enum: [invalid_execution_profile, execution_profile_mismatch] + message: + type: string expected_profile: type: string actual_profile: @@ -250,9 +258,9 @@ paths: schema: $ref: '#/components/schemas/Error' '400': - $ref: '#/components/responses/InvalidExecutionProfile' + $ref: '#/components/responses/BadRequest' '409': - $ref: '#/components/responses/ExecutionProfileMismatch' + $ref: '#/components/responses/Conflict' '503': description: Service unavailable content: @@ -293,9 +301,9 @@ paths: schema: $ref: '#/components/schemas/Error' '400': - $ref: '#/components/responses/InvalidExecutionProfile' + $ref: '#/components/responses/BadRequest' '409': - $ref: '#/components/responses/ExecutionProfileMismatch' + $ref: '#/components/responses/Conflict' /upload: post: @@ -333,9 +341,9 @@ paths: schema: $ref: '#/components/schemas/Error' '400': - $ref: '#/components/responses/InvalidExecutionProfile' + $ref: '#/components/responses/BadRequest' '409': - $ref: '#/components/responses/ExecutionProfileMismatch' + $ref: '#/components/responses/Conflict' /files/{session_id}: get: @@ -365,9 +373,9 @@ paths: items: $ref: '#/components/schemas/FileObject' '400': - $ref: '#/components/responses/InvalidExecutionProfile' + $ref: '#/components/responses/BadRequest' '409': - $ref: '#/components/responses/ExecutionProfileMismatch' + $ref: '#/components/responses/Conflict' /files/{session_id}/{fileId}: delete: @@ -397,6 +405,6 @@ paths: schema: $ref: '#/components/schemas/Error' '400': - $ref: '#/components/responses/InvalidExecutionProfile' + $ref: '#/components/responses/BadRequest' '409': - $ref: '#/components/responses/ExecutionProfileMismatch' + $ref: '#/components/responses/Conflict' diff --git a/service/src/config.ts b/service/src/config.ts index 4309af1..ecf3566 100644 --- a/service/src/config.ts +++ b/service/src/config.ts @@ -3,7 +3,10 @@ dotenv.config(); import { nanoid } from 'nanoid'; import type * as t from './types'; import { Languages } from './enum'; -import { resolveExecutionProfile } from './execution-profile'; +import { + resolveExecutionProfile, + resolveExecutionProfileSource, +} from './execution-profile'; export const languageConfig: Record = { [Languages.bash]: { language: 'bash', version: '5.2.0', fileName: 'script.sh' }, @@ -368,6 +371,9 @@ export const env = { process.env.CODEAPI_EXECUTION_PROFILE, runtimeSessionMode, ), + EXECUTION_PROFILE_SOURCE: resolveExecutionProfileSource( + process.env.CODEAPI_EXECUTION_PROFILE, + ), RUNTIME_SESSION_LOCK_WAIT_MS: configuredNumber( process.env.CODEAPI_RUNTIME_SESSION_LOCK_WAIT_MS, 15_000, diff --git a/service/src/execution-profile.test.ts b/service/src/execution-profile.test.ts index c086a02..da20bce 100644 --- a/service/src/execution-profile.test.ts +++ b/service/src/execution-profile.test.ts @@ -3,6 +3,8 @@ import { checkExecutionProfileExpectation, queueNamesForExecutionProfile, resolveExecutionProfile, + resolveExecutionProfileSource, + validateQueuedExecutionProfile, } from './execution-profile'; describe('execution profile resolution', () => { @@ -25,9 +27,10 @@ describe('execution profile resolution', () => { expect(() => resolveExecutionProfile('lambda', 'affinity')).toThrow( 'CODEAPI_EXECUTION_PROFILE must be one of: default, stateful', ); - expect(() => resolveExecutionProfile('', 'stateless')).toThrow( - 'CODEAPI_EXECUTION_PROFILE', - ); + expect(resolveExecutionProfile('', 'stateless')).toBe('default'); + expect(resolveExecutionProfile(' ', 'affinity')).toBe('stateful'); + expect(resolveExecutionProfileSource('')).toBe('inferred'); + expect(resolveExecutionProfileSource('stateful')).toBe('explicit'); }); }); @@ -68,8 +71,8 @@ describe('execution profile request assertion', () => { ok: false, status: 409, body: { - error: 'Expected the stateful execution profile, but reached default', - code: 'execution_profile_mismatch', + error: 'execution_profile_mismatch', + message: 'Expected the stateful execution profile, but reached default', expected_profile: 'stateful', actual_profile: 'default', }, @@ -81,10 +84,26 @@ describe('execution profile request assertion', () => { ok: false, status: 400, body: { - code: 'invalid_execution_profile', + error: 'invalid_execution_profile', expected_profile: 'aws', actual_profile: 'default', }, }); }); }); + +describe('queued execution profile validation', () => { + test('accepts matching and legacy jobs', () => { + expect(() => validateQueuedExecutionProfile('stateful', 'stateful')).not.toThrow(); + expect(() => validateQueuedExecutionProfile(undefined, 'default')).not.toThrow(); + }); + + test('rejects invalid and cross-profile jobs', () => { + expect(() => validateQueuedExecutionProfile('invalid', 'default')).toThrow( + 'Queued job has invalid execution profile', + ); + expect(() => validateQueuedExecutionProfile('stateful', 'default')).toThrow( + 'Queued job targets the stateful execution profile, but worker serves default', + ); + }); +}); diff --git a/service/src/execution-profile.ts b/service/src/execution-profile.ts index 354e04c..c495190 100644 --- a/service/src/execution-profile.ts +++ b/service/src/execution-profile.ts @@ -1,6 +1,7 @@ export const EXECUTION_PROFILES = ['default', 'stateful'] as const; export type ExecutionProfile = typeof EXECUTION_PROFILES[number]; +export type ExecutionProfileSource = 'explicit' | 'inferred'; export const EXPECTED_EXECUTION_PROFILE_HEADER = 'X-CodeAPI-Expected-Profile'; export const EXECUTION_PROFILE_HEADER = 'X-CodeAPI-Execution-Profile'; @@ -14,9 +15,10 @@ export function resolveExecutionProfile( raw: string | undefined, runtimeSessionMode: 'stateless' | 'affinity' | 'strict', ): ExecutionProfile { - if (raw != null) { - if (EXECUTION_PROFILES.includes(raw as ExecutionProfile)) { - return raw as ExecutionProfile; + const configuredChoice = raw?.trim(); + if (configuredChoice) { + if (EXECUTION_PROFILES.includes(configuredChoice as ExecutionProfile)) { + return configuredChoice as ExecutionProfile; } throw new Error( `CODEAPI_EXECUTION_PROFILE must be one of: ${EXECUTION_PROFILES.join(', ')}`, @@ -33,9 +35,28 @@ export function resolveExecutionProfile( : 'default'; } +export function resolveExecutionProfileSource( + raw: string | undefined, +): ExecutionProfileSource { + return raw?.trim() ? 'explicit' : 'inferred'; +} + +const LEGACY_QUEUE_NAMES: ExecutionProfileQueueNames = { + python: 'python-queue', + other: 'other-queue', +}; + +const EXPLICIT_PROFILE_QUEUE_NAMES: Record = { + default: LEGACY_QUEUE_NAMES, + stateful: { + python: 'stateful-python-queue', + other: 'stateful-other-queue', + }, +}; + export function queueNamesForExecutionProfile( profile: ExecutionProfile, - source: 'explicit' | 'inferred', + source: ExecutionProfileSource, ): ExecutionProfileQueueNames { /* A pre-profile affinity/strict deployment used the legacy queues. Keep * inferred profiles on those names so API and worker Deployments can roll @@ -43,16 +64,9 @@ export function queueNamesForExecutionProfile( * Queue isolation is an explicit cutover: operators bring up the stateful * stack with CODEAPI_EXECUTION_PROFILE=stateful on both sides, then switch * callers to its endpoint. */ - if (profile === 'stateful' && source === 'explicit') { - return { - python: 'stateful-python-queue', - other: 'stateful-other-queue', - }; - } - return { - python: 'python-queue', - other: 'other-queue', - }; + return source === 'explicit' + ? EXPLICIT_PROFILE_QUEUE_NAMES[profile] + : LEGACY_QUEUE_NAMES; } export type ExecutionProfileExpectation = @@ -61,8 +75,8 @@ export type ExecutionProfileExpectation = ok: false; status: 400 | 409; body: { - error: string; - code: 'invalid_execution_profile' | 'execution_profile_mismatch'; + error: 'invalid_execution_profile' | 'execution_profile_mismatch'; + message: string; expected_profile?: string; actual_profile: ExecutionProfile; }; @@ -79,8 +93,8 @@ export function checkExecutionProfileExpectation( ok: false, status: 400, body: { - error: `Invalid execution profile: ${rawExpectedProfile}`, - code: 'invalid_execution_profile', + error: 'invalid_execution_profile', + message: `Invalid execution profile: ${rawExpectedProfile}`, expected_profile: rawExpectedProfile, actual_profile: actualProfile, }, @@ -92,8 +106,8 @@ export function checkExecutionProfileExpectation( ok: false, status: 409, body: { - error: `Expected the ${rawExpectedProfile} execution profile, but reached ${actualProfile}`, - code: 'execution_profile_mismatch', + error: 'execution_profile_mismatch', + message: `Expected the ${rawExpectedProfile} execution profile, but reached ${actualProfile}`, expected_profile: rawExpectedProfile, actual_profile: actualProfile, }, @@ -102,3 +116,20 @@ export function checkExecutionProfileExpectation( return { ok: true }; } + +/** Reject producer/consumer profile drift before a worker invokes a sandbox. + * Missing profile is accepted only for jobs queued by pre-profile binaries. */ +export function validateQueuedExecutionProfile( + jobProfile: unknown, + workerProfile: ExecutionProfile, +): void { + if (jobProfile == null) return; + if (!EXECUTION_PROFILES.includes(jobProfile as ExecutionProfile)) { + throw new Error(`Queued job has invalid execution profile: ${String(jobProfile)}`); + } + if (jobProfile !== workerProfile) { + throw new Error( + `Queued job targets the ${jobProfile} execution profile, but worker serves ${workerProfile}`, + ); + } +} diff --git a/service/src/lifecycle.ts b/service/src/lifecycle.ts index e27846e..8fc0adc 100644 --- a/service/src/lifecycle.ts +++ b/service/src/lifecycle.ts @@ -11,11 +11,20 @@ import { } from './secure-startup'; import logger from './logger'; import { shutdownTelemetry } from './telemetry'; +import { configureExecutionProfileMetrics } from './metrics'; const { INSTANCE_ID } = env; let isShuttingDown = false; let isStartingUp = true; +function configureProfileMetrics(): void { + configureExecutionProfileMetrics({ + profile: env.EXECUTION_PROFILE, + sandboxBackend: env.SANDBOX_BACKEND, + runtimeSessionMode: env.RUNTIME_SESSION_MODE, + }); +} + async function shutdownTracing(): Promise { try { await shutdownTelemetry(); @@ -87,6 +96,7 @@ export async function startupApiOnly(): Promise { * the MINIO_* checkpoint creds) into API pods just to boot. The worker and * combined startups own that validation. */ await validateLifecycleAuthConfig(); + configureProfileMetrics(); // Set up queue listeners for monitoring (optional, for observability) setupQueueListeners(pyQueue, 'Python'); @@ -105,6 +115,7 @@ export async function startupWorkerOnly(): Promise { validateWorkerHardenedConfig(); validateExecutionProfilePolicy(); validateSandboxBackendPolicy(); + configureProfileMetrics(); // Dynamically import workers to start them const { pyWorker, otherWorker } = await import('./workers'); @@ -141,6 +152,7 @@ async function gracefulStartup(): Promise { validateExecutionProfilePolicy(); validateSandboxBackendPolicy(); await validateLifecycleAuthConfig(); + configureProfileMetrics(); try { logger.info('Setting up queues...'); diff --git a/service/src/local-api.ts b/service/src/local-api.ts index 7428346..701270d 100644 --- a/service/src/local-api.ts +++ b/service/src/local-api.ts @@ -21,16 +21,17 @@ import { env } from './config'; import logger from './logger'; import { shutdownTelemetry, traceHttpRequest } from './telemetry'; import { validateExecutionProfilePolicy } from './secure-startup'; +import { configureExecutionProfileMetrics } from './metrics'; const app = express(); app.disable('x-powered-by'); app.set('trust proxy', 1); -app.use(executionProfileMiddleware); let localShuttingDown = false; const v1 = Router(); app.use(traceHttpRequest('codeapi.local_api.request')); +app.use(executionProfileMiddleware); app.use(json({ limit: env.HTTP_JSON_LIMIT })); // Health check @@ -56,6 +57,11 @@ async function localStartup(): Promise { logger.info('Starting local development server...'); logger.info('⚠️ LOCAL MODE - No authentication required'); validateExecutionProfilePolicy(); + configureExecutionProfileMetrics({ + profile: env.EXECUTION_PROFILE, + sandboxBackend: env.SANDBOX_BACKEND, + runtimeSessionMode: env.RUNTIME_SESSION_MODE, + }); try { // Set a local user ID for session management diff --git a/service/src/metrics.test.ts b/service/src/metrics.test.ts index d5c48aa..0758fd6 100644 --- a/service/src/metrics.test.ts +++ b/service/src/metrics.test.ts @@ -1,6 +1,8 @@ import { afterEach, expect, test } from 'bun:test'; import { bullmqQueueJobs, + configureExecutionProfileMetrics, + executionProfileInfo, metricsResponse, registerBullmqQueueMetricsCollector, } from './metrics'; @@ -8,6 +10,22 @@ import { afterEach(() => { registerBullmqQueueMetricsCollector(undefined); bullmqQueueJobs.set({ queue: 'other-queue', state: 'waiting' }, 0); + executionProfileInfo.reset(); +}); + +test('execution identity is published only when an API or worker configures it', async () => { + executionProfileInfo.reset(); + expect((await metricsResponse()).body).not.toContain('codeapi_execution_profile_info{'); + + configureExecutionProfileMetrics({ + profile: 'stateful', + sandboxBackend: 'lambda-microvm', + runtimeSessionMode: 'affinity', + }); + + expect((await metricsResponse()).body).toContain( + 'codeapi_execution_profile_info{profile="stateful",sandbox_backend="lambda-microvm",runtime_session_mode="affinity"} 1', + ); }); test('metricsResponse collects BullMQ queue gauges on scrape', async () => { diff --git a/service/src/metrics.ts b/service/src/metrics.ts index a2d9865..adfd987 100644 --- a/service/src/metrics.ts +++ b/service/src/metrics.ts @@ -1,6 +1,8 @@ import client, { register, Counter, Histogram, Gauge } from 'prom-client'; import { normalizeMetricPath } from './httpPathNormalize'; -import { env } from './config'; +import type { ExecutionProfile } from './execution-profile'; +import type { RuntimeSessionMode } from './types/service'; +import type { SandboxBackend } from './sandbox-backend/types'; client.collectDefaultMetrics({ register }); @@ -10,11 +12,18 @@ export const executionProfileInfo = new Gauge({ labelNames: ['profile', 'sandbox_backend', 'runtime_session_mode'] as const, }); -executionProfileInfo.set({ - profile: env.EXECUTION_PROFILE, - sandbox_backend: env.SANDBOX_BACKEND, - runtime_session_mode: env.RUNTIME_SESSION_MODE, -}, 1); +export function configureExecutionProfileMetrics(identity: { + profile: ExecutionProfile; + sandboxBackend: SandboxBackend['name']; + runtimeSessionMode: RuntimeSessionMode; +}): void { + executionProfileInfo.reset(); + executionProfileInfo.set({ + profile: identity.profile, + sandbox_backend: identity.sandboxBackend, + runtime_session_mode: identity.runtimeSessionMode, + }, 1); +} export const executionProfileRequestRejections = new Counter({ name: 'codeapi_execution_profile_request_rejections_total', diff --git a/service/src/middleware/execution-profile.test.ts b/service/src/middleware/execution-profile.test.ts index 43368ef..5a3ebac 100644 --- a/service/src/middleware/execution-profile.test.ts +++ b/service/src/middleware/execution-profile.test.ts @@ -9,7 +9,7 @@ afterEach(() => { env.EXECUTION_PROFILE = savedProfile; }); -function invoke(expectedProfile?: string): { +function invoke(expectedProfile?: string, rawHeaders?: string[]): { headers: Record; status?: number; body?: unknown; @@ -23,6 +23,7 @@ function invoke(expectedProfile?: string): { } = { headers: {}, nextCalled: false }; const req = { get: () => expectedProfile, + ...(rawHeaders ? { rawHeaders } : {}), } as unknown as Request; const res = { setHeader: (name: string, value: string) => { @@ -60,7 +61,7 @@ describe('execution profile middleware', () => { headers: { 'X-CodeAPI-Execution-Profile': 'default' }, status: 409, body: { - code: 'execution_profile_mismatch', + error: 'execution_profile_mismatch', expected_profile: 'stateful', actual_profile: 'default', }, @@ -68,6 +69,22 @@ describe('execution profile middleware', () => { }); }); + test('rejects duplicate expected-profile headers', () => { + env.EXECUTION_PROFILE = 'default'; + expect(invoke('default', [ + 'X-CodeAPI-Expected-Profile', 'stateful', + 'x-codeapi-expected-profile', 'default', + ])).toMatchObject({ + status: 400, + body: { + error: 'invalid_execution_profile', + expected_profile: 'stateful,default', + actual_profile: 'default', + }, + nextCalled: false, + }); + }); + test('keeps older callers working when they omit the assertion', () => { env.EXECUTION_PROFILE = 'default'; expect(invoke()).toEqual({ diff --git a/service/src/middleware/execution-profile.ts b/service/src/middleware/execution-profile.ts index 2e3bdae..38512dc 100644 --- a/service/src/middleware/execution-profile.ts +++ b/service/src/middleware/execution-profile.ts @@ -7,6 +7,21 @@ import { } from '../execution-profile'; import { executionProfileRequestRejections } from '../metrics'; +function expectedExecutionProfile(req: Request): string | undefined { + const values: string[] = []; + const rawHeaders = Array.isArray(req.rawHeaders) ? req.rawHeaders : []; + for (let index = 0; index < rawHeaders.length; index += 2) { + if (rawHeaders[index]?.toLowerCase() === EXPECTED_EXECUTION_PROFILE_HEADER.toLowerCase()) { + values.push(rawHeaders[index + 1] ?? ''); + } + } + /* Joining makes duplicate assertions invalid even when an HTTP runtime + * would otherwise collapse them with last-value-wins semantics. */ + return values.length > 0 + ? values.join(',') + : req.get(EXPECTED_EXECUTION_PROFILE_HEADER); +} + /** * Advertise this deployment's profile and fail closed when a trusted caller * reaches the wrong endpoint. Apply before routing so no file, programmatic, @@ -20,7 +35,7 @@ export function executionProfileMiddleware( res.setHeader(EXECUTION_PROFILE_HEADER, env.EXECUTION_PROFILE); const expectation = checkExecutionProfileExpectation( - req.get(EXPECTED_EXECUTION_PROFILE_HEADER), + expectedExecutionProfile(req), env.EXECUTION_PROFILE, ); if (expectation.ok) { @@ -29,7 +44,7 @@ export function executionProfileMiddleware( } executionProfileRequestRejections.inc({ - reason: expectation.body.code === 'execution_profile_mismatch' + reason: expectation.body.error === 'execution_profile_mismatch' ? 'mismatch' : 'invalid', }); diff --git a/service/src/queue.ts b/service/src/queue.ts index a93ea81..91f97c3 100644 --- a/service/src/queue.ts +++ b/service/src/queue.ts @@ -59,7 +59,7 @@ const connection = new IORedis({ // consuming each other's jobs when they share Redis. const queueNames = queueNamesForExecutionProfile( env.EXECUTION_PROFILE, - process.env.CODEAPI_EXECUTION_PROFILE == null ? 'inferred' : 'explicit', + env.EXECUTION_PROFILE_SOURCE, ); const pyQueue = new Queue(queueNames.python, { connection }); const otherQueue = new Queue(queueNames.other, { connection }); diff --git a/service/src/secure-startup.test.ts b/service/src/secure-startup.test.ts index 7e94960..4aa603e 100644 --- a/service/src/secure-startup.test.ts +++ b/service/src/secure-startup.test.ts @@ -12,6 +12,7 @@ const savedEnv = { ...process.env }; const saved = { hardened: env.HARDENED_SANDBOX_MODE, executionProfile: env.EXECUTION_PROFILE, + executionProfileSource: env.EXECUTION_PROFILE_SOURCE, sandboxBackend: env.SANDBOX_BACKEND, ptcMode: env.PTC_MODE, runtimeSessionMode: env.RUNTIME_SESSION_MODE, @@ -49,6 +50,7 @@ function restore(): void { Object.assign(process.env, savedEnv); env.HARDENED_SANDBOX_MODE = saved.hardened; env.EXECUTION_PROFILE = saved.executionProfile; + env.EXECUTION_PROFILE_SOURCE = saved.executionProfileSource; env.SANDBOX_BACKEND = saved.sandboxBackend; env.PTC_MODE = saved.ptcMode; env.RUNTIME_SESSION_MODE = saved.runtimeSessionMode; @@ -84,6 +86,7 @@ afterEach(restore); describe('execution profile policy', () => { test('accepts the AWS-free default profile', () => { env.EXECUTION_PROFILE = 'default'; + env.EXECUTION_PROFILE_SOURCE = 'explicit'; env.SANDBOX_BACKEND = 'http'; env.RUNTIME_SESSION_MODE = 'stateless'; expect(() => validateExecutionProfilePolicy()).not.toThrow(); @@ -91,6 +94,7 @@ describe('execution profile policy', () => { test('accepts affinity and strict stateful profiles', () => { env.EXECUTION_PROFILE = 'stateful'; + env.EXECUTION_PROFILE_SOURCE = 'explicit'; env.SANDBOX_BACKEND = 'lambda-microvm'; env.RUNTIME_SESSION_MODE = 'affinity'; expect(() => validateExecutionProfilePolicy()).not.toThrow(); @@ -100,6 +104,7 @@ describe('execution profile policy', () => { test('does not require worker-only backend config on API-only pods', () => { env.EXECUTION_PROFILE = 'stateful'; + env.EXECUTION_PROFILE_SOURCE = 'explicit'; env.SANDBOX_BACKEND = 'http'; env.RUNTIME_SESSION_MODE = 'affinity'; expect(() => validateExecutionProfilePolicy({ requireBackendMatch: false })).not.toThrow(); @@ -107,6 +112,7 @@ describe('execution profile policy', () => { test('rejects a default profile backed by AWS or stateful sessions', () => { env.EXECUTION_PROFILE = 'default'; + env.EXECUTION_PROFILE_SOURCE = 'explicit'; env.SANDBOX_BACKEND = 'lambda-microvm'; env.RUNTIME_SESSION_MODE = 'stateless'; expect(() => validateExecutionProfilePolicy()).toThrow( @@ -120,8 +126,17 @@ describe('execution profile policy', () => { ); }); + test('preserves a pre-profile Lambda/stateless deployment when inferred', () => { + env.EXECUTION_PROFILE = 'default'; + env.EXECUTION_PROFILE_SOURCE = 'inferred'; + env.SANDBOX_BACKEND = 'lambda-microvm'; + env.RUNTIME_SESSION_MODE = 'stateless'; + expect(() => validateExecutionProfilePolicy()).not.toThrow(); + }); + test('rejects a stateful profile without Lambda affinity', () => { env.EXECUTION_PROFILE = 'stateful'; + env.EXECUTION_PROFILE_SOURCE = 'explicit'; env.SANDBOX_BACKEND = 'http'; env.RUNTIME_SESSION_MODE = 'affinity'; expect(() => validateExecutionProfilePolicy()).toThrow( diff --git a/service/src/secure-startup.ts b/service/src/secure-startup.ts index df9409e..a82dabf 100644 --- a/service/src/secure-startup.ts +++ b/service/src/secure-startup.ts @@ -73,9 +73,14 @@ export function validateExecutionProfilePolicy(options: { } = {}): void { const requireBackendMatch = options.requireBackendMatch ?? true; if (env.EXECUTION_PROFILE === 'default') { + const compatibleBackend = env.SANDBOX_BACKEND === 'http' + || ( + env.EXECUTION_PROFILE_SOURCE === 'inferred' + && env.SANDBOX_BACKEND === 'lambda-microvm' + ); if ( env.RUNTIME_SESSION_MODE !== 'stateless' - || (requireBackendMatch && env.SANDBOX_BACKEND !== 'http') + || (requireBackendMatch && !compatibleBackend) ) { throw new SecureStartupConfigError( 'CODEAPI_EXECUTION_PROFILE=default requires ' diff --git a/service/src/service/programmatic-router.ts b/service/src/service/programmatic-router.ts index 82e4d07..eade27f 100644 --- a/service/src/service/programmatic-router.ts +++ b/service/src/service/programmatic-router.ts @@ -407,6 +407,7 @@ async function runReplayIteration( executionId: state.execution_id, tenantId: state.tenantId, canonicalUserId: state.canonicalUserId, + executionProfile: env.EXECUTION_PROFILE, runtimeSessionMode: 'stateless', runtimeSessionExemption: PROGRAMMATIC_RUNTIME_SESSION_EXEMPTION, executionManifestClaims: sandboxSecurity.executionManifestClaims, @@ -1376,6 +1377,7 @@ async function handleBlocking( executionId: execution_id, tenantId: identity.storageNamespace, canonicalUserId: identity.canonicalUserId, + executionProfile: env.EXECUTION_PROFILE, runtimeSessionMode: 'stateless', runtimeSessionExemption: PROGRAMMATIC_RUNTIME_SESSION_EXEMPTION, executionManifestClaims: sandboxSecurity.executionManifestClaims, diff --git a/service/src/service/router.ts b/service/src/service/router.ts index 4c844fb..f355c2d 100644 --- a/service/src/service/router.ts +++ b/service/src/service/router.ts @@ -246,6 +246,7 @@ router.post('/exec', executionLimiter, async (req: t.AuthenticatedRequest, res) executionId: execution_id, tenantId: identity.storageNamespace, canonicalUserId: identity.canonicalUserId, + executionProfile: env.EXECUTION_PROFILE, ...(runtimeSessionId != null ? { runtimeSessionId } : {}), runtimeSessionMode, executionManifestClaims: sandboxSecurity.executionManifestClaims, diff --git a/service/src/types/service.ts b/service/src/types/service.ts index d298298..a642dda 100644 --- a/service/src/types/service.ts +++ b/service/src/types/service.ts @@ -3,6 +3,7 @@ import type { Request } from 'express'; import type { ExecutionManifestClaims } from '../execution-manifest'; import type { ExecutionIdentity } from '../execution-identity'; import type { CodeApiPrincipal } from '../auth/principal'; +import type { ExecutionProfile } from '../execution-profile'; import { Jobs } from '@/enum/service'; /** @@ -250,6 +251,8 @@ export type JobData = { executionId?: string; tenantId?: string; canonicalUserId?: string; + /** Producer deployment identity. Optional only for pre-profile queued jobs. */ + executionProfile?: ExecutionProfile; /** * Server-derived runtime session identity. Absence is stateless unless * strict mode requires it; explicit exemptions document intentional gaps. diff --git a/service/src/workers.ts b/service/src/workers.ts index 206542b..ed2a46d 100644 --- a/service/src/workers.ts +++ b/service/src/workers.ts @@ -17,6 +17,7 @@ import { isSyntheticPrincipalSource } from './auth/synthetic'; import { withSpan, withTraceContext } from './telemetry'; import { workerDeadlineFailure } from './worker-error'; import logger from './logger'; +import { validateQueuedExecutionProfile } from './execution-profile'; const { INSTANCE_ID } = env; const WORKER_ID = `${INSTANCE_ID}-${process.pid}`; @@ -31,6 +32,8 @@ async function processJob(job: t.ExecuteJob): Promise { 'messaging.operation.name': 'process', 'messaging.message.id': typeof job.id === 'string' ? job.id : String(job.id ?? ''), 'codeapi.language': job.data.payload?.language ?? 'unknown', + 'codeapi.execution_profile': job.data.executionProfile ?? 'legacy', + 'codeapi.worker_execution_profile': env.EXECUTION_PROFILE, }, () => processJobInner(job), 'CONSUMER')); } @@ -56,6 +59,7 @@ async function processJobInner(job: t.ExecuteJob): Promise { if (controller.signal.aborted) { throw new Error(`Job timed out after ${env.JOB_TIMEOUT}ms`); } + validateQueuedExecutionProfile(job.data.executionProfile, env.EXECUTION_PROFILE); let sandboxPayload = payload; let executionManifestClaims = job.data.executionManifestClaims; let egressGrantToken = job.data.egressGrantToken;