diff --git a/README.md b/README.md index 718db6a..6612cb5 100644 --- a/README.md +++ b/README.md @@ -24,6 +24,40 @@ 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. + +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 +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 `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 +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..309d6ec 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. 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). | @@ -408,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 9198b1a..9b3efbb 100644 --- a/helm/codeapi/README.md +++ b/helm/codeapi/README.md @@ -53,6 +53,54 @@ 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.** 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 +`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/helm/codeapi/templates/api-deployment.yaml b/helm/codeapi/templates/api-deployment.yaml index 0c801ab..bf5f91e 100644 --- a/helm/codeapi/templates/api-deployment.yaml +++ b/helm/codeapi/templates/api-deployment.yaml @@ -41,6 +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: {{ . | 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 3484a61..39c2e19 100644 --- a/helm/codeapi/templates/worker-sandbox-deployment.yaml +++ b/helm/codeapi/templates/worker-sandbox-deployment.yaml @@ -147,6 +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: {{ . | 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 2385814..75abaaf 100644 --- a/helm/codeapi/values.yaml +++ b/helm/codeapi/values.yaml @@ -22,6 +22,12 @@ internalServiceAuth: hardenedSandboxMode: true +# 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 # OTLP/HTTP collector endpoint, e.g. "http://opentelemetry-collector.observability:4318". diff --git a/service/openapi.yml b/service/openapi.yml index c1f8f6e..913e780 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,49 @@ 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: + BadRequest: + description: Invalid request or invalid expected execution profile + headers: + X-CodeAPI-Execution-Profile: + $ref: '#/components/headers/ExecutionProfile' + content: + application/json: + schema: + 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: + anyOf: + - $ref: '#/components/schemas/Error' + - $ref: '#/components/schemas/ExecutionProfileError' + schemas: FileRef: type: object @@ -108,6 +154,14 @@ components: type: array items: $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. Required in strict runtime + session mode and ignored by the default stateless profile. FileObject: type: object @@ -155,6 +209,23 @@ components: type: string details: type: string + message: + type: string + + ExecutionProfileError: + type: object + required: [error, message, actual_profile] + properties: + error: + type: string + enum: [invalid_execution_profile, execution_profile_mismatch] + message: + type: string + expected_profile: + type: string + actual_profile: + type: string + enum: [default, stateful] paths: /exec: @@ -162,6 +233,8 @@ paths: summary: Execute code description: Execute code with specified language and parameters operationId: executeCode + parameters: + - $ref: '#/components/parameters/ExpectedExecutionProfile' requestBody: required: true content: @@ -171,6 +244,9 @@ paths: responses: '200': description: Successful execution + headers: + X-CodeAPI-Execution-Profile: + $ref: '#/components/headers/ExecutionProfile' content: application/json: schema: @@ -181,6 +257,10 @@ paths: application/json: schema: $ref: '#/components/schemas/Error' + '400': + $ref: '#/components/responses/BadRequest' + '409': + $ref: '#/components/responses/Conflict' '503': description: Service unavailable content: @@ -192,6 +272,7 @@ paths: get: summary: Download a file parameters: + - $ref: '#/components/parameters/ExpectedExecutionProfile' - name: session_id in: path required: true @@ -205,6 +286,9 @@ paths: responses: '200': description: File content + headers: + X-CodeAPI-Execution-Profile: + $ref: '#/components/headers/ExecutionProfile' content: application/octet-stream: schema: @@ -216,10 +300,16 @@ paths: application/json: schema: $ref: '#/components/schemas/Error' + '400': + $ref: '#/components/responses/BadRequest' + '409': + $ref: '#/components/responses/Conflict' /upload: post: summary: Upload files + parameters: + - $ref: '#/components/parameters/ExpectedExecutionProfile' requestBody: required: true content: @@ -237,6 +327,9 @@ paths: responses: '200': description: Successful upload + headers: + X-CodeAPI-Execution-Profile: + $ref: '#/components/headers/ExecutionProfile' content: application/json: schema: @@ -247,11 +340,16 @@ paths: application/json: schema: $ref: '#/components/schemas/Error' + '400': + $ref: '#/components/responses/BadRequest' + '409': + $ref: '#/components/responses/Conflict' /files/{session_id}: get: summary: Get files information parameters: + - $ref: '#/components/parameters/ExpectedExecutionProfile' - name: session_id in: path required: true @@ -265,17 +363,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/BadRequest' + '409': + $ref: '#/components/responses/Conflict' /files/{session_id}/{fileId}: delete: summary: Delete a file parameters: + - $ref: '#/components/parameters/ExpectedExecutionProfile' - name: session_id in: path required: true @@ -289,9 +395,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/BadRequest' + '409': + $ref: '#/components/responses/Conflict' 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..ecf3566 100644 --- a/service/src/config.ts +++ b/service/src/config.ts @@ -3,6 +3,10 @@ dotenv.config(); import { nanoid } from 'nanoid'; import type * as t from './types'; import { Languages } from './enum'; +import { + resolveExecutionProfile, + resolveExecutionProfileSource, +} from './execution-profile'; export const languageConfig: Record = { [Languages.bash]: { language: 'bash', version: '5.2.0', fileName: 'script.sh' }, @@ -259,6 +263,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 +351,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 +360,20 @@ 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, + ), + 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/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..da20bce --- /dev/null +++ b/service/src/execution-profile.test.ts @@ -0,0 +1,109 @@ +import { describe, expect, test } from 'bun:test'; +import { + checkExecutionProfileExpectation, + queueNamesForExecutionProfile, + resolveExecutionProfile, + resolveExecutionProfileSource, + validateQueuedExecutionProfile, +} 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')).toBe('default'); + expect(resolveExecutionProfile(' ', 'affinity')).toBe('stateful'); + expect(resolveExecutionProfileSource('')).toBe('inferred'); + expect(resolveExecutionProfileSource('stateful')).toBe('explicit'); + }); +}); + +describe('execution profile queue isolation', () => { + test('keeps the legacy queue names for the default profile', () => { + expect(queueNamesForExecutionProfile('default', 'explicit')).toEqual({ + python: 'python-queue', + other: 'other-queue', + }); + }); + + 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', + }); + }); +}); + +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: 'execution_profile_mismatch', + message: 'Expected the stateful execution profile, but reached default', + 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: { + 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 new file mode 100644 index 0000000..c495190 --- /dev/null +++ b/service/src/execution-profile.ts @@ -0,0 +1,135 @@ +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'; + +export interface ExecutionProfileQueueNames { + python: string; + other: string; +} + +export function resolveExecutionProfile( + raw: string | undefined, + runtimeSessionMode: 'stateless' | 'affinity' | 'strict', +): 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(', ')}`, + ); + } + + /* 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 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: 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 + * 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. */ + return source === 'explicit' + ? EXPLICIT_PROFILE_QUEUE_NAMES[profile] + : LEGACY_QUEUE_NAMES; +} + +export type ExecutionProfileExpectation = + | { ok: true } + | { + ok: false; + status: 400 | 409; + body: { + error: 'invalid_execution_profile' | 'execution_profile_mismatch'; + message: string; + 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', + message: `Invalid execution profile: ${rawExpectedProfile}`, + expected_profile: rawExpectedProfile, + actual_profile: actualProfile, + }, + }; + } + + if (rawExpectedProfile !== actualProfile) { + return { + ok: false, + status: 409, + body: { + error: 'execution_profile_mismatch', + message: `Expected the ${rawExpectedProfile} execution profile, but reached ${actualProfile}`, + expected_profile: rawExpectedProfile, + actual_profile: actualProfile, + }, + }; + } + + 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 59a18a8..8fc0adc 100644 --- a/service/src/lifecycle.ts +++ b/service/src/lifecycle.ts @@ -3,14 +3,28 @@ 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'; +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(); @@ -75,12 +89,14 @@ 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 * 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'); @@ -97,7 +113,9 @@ export async function startupApiOnly(): Promise { export async function startupWorkerOnly(): Promise { logger.info('Starting Worker service...'); validateWorkerHardenedConfig(); + validateExecutionProfilePolicy(); validateSandboxBackendPolicy(); + configureProfileMetrics(); // Dynamically import workers to start them const { pyWorker, otherWorker } = await import('./workers'); @@ -131,8 +149,10 @@ async function gracefulStartup(): Promise { logger.info('Starting up service (combined API + Workers)...'); validateApiHardenedConfig(); validateWorkerHardenedConfig(); + 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 b9b280d..701270d 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,6 +20,8 @@ import './workers'; 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'); @@ -28,6 +31,7 @@ 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 @@ -52,6 +56,12 @@ app.use(requestErrorLogger); 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 ba052dc..adfd987 100644 --- a/service/src/metrics.ts +++ b/service/src/metrics.ts @@ -1,8 +1,36 @@ import client, { register, Counter, Histogram, Gauge } from 'prom-client'; import { normalizeMetricPath } from './httpPathNormalize'; +import type { ExecutionProfile } from './execution-profile'; +import type { RuntimeSessionMode } from './types/service'; +import type { SandboxBackend } from './sandbox-backend/types'; 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, +}); + +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', + 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..5a3ebac --- /dev/null +++ b/service/src/middleware/execution-profile.test.ts @@ -0,0 +1,95 @@ +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, rawHeaders?: 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, + ...(rawHeaders ? { rawHeaders } : {}), + } 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: { + error: 'execution_profile_mismatch', + expected_profile: 'stateful', + actual_profile: 'default', + }, + nextCalled: false, + }); + }); + + 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({ + 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..38512dc --- /dev/null +++ b/service/src/middleware/execution-profile.ts @@ -0,0 +1,52 @@ +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'; + +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, + * 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( + expectedExecutionProfile(req), + env.EXECUTION_PROFILE, + ); + if (expectation.ok) { + next(); + return; + } + + executionProfileRequestRejections.inc({ + reason: expectation.body.error === '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..91f97c3 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,22 @@ 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, + env.EXECUTION_PROFILE_SOURCE, +); +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 +116,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..4aa603e 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,8 @@ import { 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, @@ -46,6 +49,8 @@ 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; @@ -78,6 +83,74 @@ function restore(): void { 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(); + }); + + 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(); + 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.EXECUTION_PROFILE_SOURCE = 'explicit'; + 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.EXECUTION_PROFILE_SOURCE = 'explicit'; + 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('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( + '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..a82dabf 100644 --- a/service/src/secure-startup.ts +++ b/service/src/secure-startup.ts @@ -63,6 +63,46 @@ 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') { + const compatibleBackend = env.SANDBOX_BACKEND === 'http' + || ( + env.EXECUTION_PROFILE_SOURCE === 'inferred' + && env.SANDBOX_BACKEND === 'lambda-microvm' + ); + if ( + env.RUNTIME_SESSION_MODE !== 'stateless' + || (requireBackendMatch && !compatibleBackend) + ) { + 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/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 d88dfdc..f355c2d 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, { @@ -245,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, @@ -279,6 +281,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/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 e9c3b2b..ed2a46d 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'; @@ -18,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}`; @@ -32,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')); } @@ -57,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; @@ -238,7 +241,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 +250,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: {