fix(codeapi): run sync validation on supported macOS runner

api/Dockerfile

@@ -255,6 +255,7 @@ RUN dnf install -y --setopt=install_weak_deps=False \
     libkrun \
     socat \
     curl \
+    openssl \
     util-linux \
     grep \
     gawk \
@@ -268,8 +269,9 @@ COPY --from=nsjail-builder /usr/local/bin/sandbox-rootfs-setup /sandbox-rootfs-s
 COPY launcher/entrypoint.sh /usr/local/bin/launcher-entrypoint.sh
 COPY docker/start-direct-sandbox.sh /usr/local/bin/start-direct-sandbox.sh
 COPY docker/sandbox-runner-healthcheck.sh /usr/local/bin/sandbox-runner-healthcheck.sh
+COPY docker/sandbox-runner-drain.sh /usr/local/bin/sandbox-runner-drain.sh
 COPY docker/sandbox-entrypoint.sh /usr/local/bin/entrypoint.sh
-RUN chmod +x /usr/local/bin/launcher-entrypoint.sh /usr/local/bin/start-direct-sandbox.sh /usr/local/bin/sandbox-runner-healthcheck.sh /usr/local/bin/entrypoint.sh
+RUN chmod +x /usr/local/bin/launcher-entrypoint.sh /usr/local/bin/start-direct-sandbox.sh /usr/local/bin/sandbox-runner-healthcheck.sh /usr/local/bin/sandbox-runner-drain.sh /usr/local/bin/entrypoint.sh
 
 EXPOSE 2000/tcp
 ENTRYPOINT ["/usr/local/bin/entrypoint.sh"]

api/src/api/drain.admission.test.ts

@@ -0,0 +1,133 @@
+import { afterEach, describe, expect, test } from 'bun:test';
+import express from 'express';
+import { request, type Server } from 'node:http';
+import { drainSandbox, resetSandboxDrainForTests, sandboxDrainSnapshot } from '../drain';
+import { withSandboxDrain } from './drain';
+
+afterEach(resetSandboxDrainForTests);
+
+async function serve(handler: Parameters<typeof withSandboxDrain>[1]) {
+  const app = express();
+  app.post('/execute', withSandboxDrain(express.json(), handler));
+  app.use((error: Error, _req: express.Request, res: express.Response, _next: express.NextFunction) => {
+    res.status(400).json({ error: error.message });
+  });
+  const server: Server = app.listen(0, '127.0.0.1');
+  await new Promise<void>((resolve) => server.once('listening', resolve));
+  const address = server.address();
+  if (!address || typeof address === 'string') throw new Error('Missing test port');
+  return { server, url: `http://127.0.0.1:${address.port}/execute` };
+}
+
+async function close(server: Server) {
+  server.closeAllConnections();
+  await new Promise<void>((resolve) => server.close(() => resolve()));
+}
+
+async function until(check: () => boolean) {
+  for (let i = 0; i < 100; i++) {
+    if (check()) return;
+    await Bun.sleep(2);
+  }
+  throw new Error('Condition did not become true');
+}
+
+describe('drain admission across request lifecycle', () => {
+  test('waits for response completion after the async handler returns', async () => {
+    let returned = false;
+    let finishResponse: (() => void) | undefined;
+    const { server, url } = await serve(async (_req, res) => {
+      res.write('partial result');
+      finishResponse = () => res.end(' complete');
+      returned = true;
+    });
+    const pendingResponse = fetch(url, {
+      method: 'POST', headers: { 'Content-Type': 'application/json' }, body: '{}',
+    });
+    try {
+      await until(() => returned);
+      let drained = false;
+      const pendingDrain = drainSandbox(1000, 0).then(result => { drained = true; return result; });
+      await Bun.sleep(10);
+      expect(sandboxDrainSnapshot().activeExecutions).toBe(1);
+      expect(drained).toBe(false);
+      finishResponse?.();
+      expect(await (await pendingResponse).text()).toBe('partial result complete');
+      expect((await pendingDrain).drained).toBe(true);
+    } finally {
+      finishResponse?.();
+      await close(server);
+    }
+  });
+
+  test('counts an unfinished upload before JSON parsing and waits for its handler', async () => {
+    let entered = false;
+    let finish!: () => void;
+    const pendingHandler = new Promise<void>((resolve) => { finish = resolve; });
+    const { server, url } = await serve(async (_req, res) => {
+      entered = true;
+      await pendingHandler;
+      res.json({ ok: true });
+    });
+    const req = request(url, { method: 'POST', headers: { 'Content-Type': 'application/json' } });
+    req.on('error', () => {});
+    try {
+      req.write('{');
+      await until(() => sandboxDrainSnapshot().activeExecutions === 1);
+      let drained = false;
+      const pending = drainSandbox(1000, 0).then(() => { drained = true; });
+      await Bun.sleep(5);
+      expect(entered).toBe(false);
+      expect(drained).toBe(false);
+      req.end('}');
+      await until(() => entered);
+      // Losing the response does not stop an already admitted execution or its cleanup.
+      req.destroy();
+      await Bun.sleep(5);
+      expect(sandboxDrainSnapshot().activeExecutions).toBe(1);
+      expect(drained).toBe(false);
+      finish();
+      await pending;
+      expect(sandboxDrainSnapshot().activeExecutions).toBe(0);
+    } finally {
+      finish();
+      req.destroy();
+      await close(server);
+    }
+  });
+
+  test('releases an aborted upload without invoking the handler', async () => {
+    let calls = 0;
+    const { server, url } = await serve(async (_req, res) => { calls++; res.end(); });
+    const req = request(url, { method: 'POST', headers: { 'Content-Type': 'application/json' } });
+    req.on('error', () => {});
+    try {
+      req.write('{');
+      await until(() => sandboxDrainSnapshot().activeExecutions === 1);
+      const pending = drainSandbox(1000, 0);
+      req.destroy();
+      expect((await pending).drained).toBe(true);
+      expect(calls).toBe(0);
+    } finally {
+      req.destroy();
+      await close(server);
+    }
+  });
+
+  test('parser failures release admission and drain rejection happens before parsing', async () => {
+    let calls = 0;
+    const { server, url } = await serve(async (_req, res) => { calls++; res.end(); });
+    try {
+      const bad = await fetch(url, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: '{' });
+      expect(bad.status).toBe(400);
+      expect(sandboxDrainSnapshot().activeExecutions).toBe(0);
+      await drainSandbox(1000, 0);
+      const rejected = await fetch(url, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: '{' });
+      expect(rejected.status).toBe(503);
+      expect((await rejected.json()).error).toBe('runner_draining');
+      expect(calls).toBe(0);
+    } finally {
+      await close(server);
+    }
+  });
+});

api/src/api/drain.routes.test.ts

@@ -0,0 +1,115 @@
+import { afterAll, afterEach, beforeAll, describe, expect, test } from 'bun:test';
+import express from 'express';
+import type { Server } from 'node:http';
+import { generateKeyPairSync, sign } from 'node:crypto';
+import { mkdtemp, readFile, rm } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { config } from '../config';
+import { resetSandboxDrainForTests, sandboxDrainSnapshot, startSandboxDrain } from '../drain';
+import { validateDrainConfiguration } from './drain';
+import { RUNNER_DRAIN_MESSAGE } from '../../../shared/drain-contract';
+import v2Router from './v2';
+
+let server: Server;
+let baseUrl: string;
+const originalMinDelay = config.drain_min_delay_ms;
+const originalEnabled = config.graceful_drain_enabled;
+const originalPublicKey = config.drain_public_key;
+let stateDirectory: string;
+const hostScript = join(__dirname, '../../../docker/sandbox-runner-drain.sh');
+
+async function runHostHook(mode?: string): Promise<string> {
+  const child = Bun.spawn(['bash', hostScript, ...(mode ? [mode] : [])], {
+    env: { ...process.env, SANDBOX_DRAIN_STATE_DIR: stateDirectory, PORT: new URL(baseUrl || 'http://localhost:2000').port },
+    stdout: 'pipe', stderr: 'pipe',
+  });
+  const output = await new Response(child.stdout).text();
+  const error = await new Response(child.stderr).text();
+  if (await child.exited !== 0) throw new Error(error);
+  return output;
+}
+
+beforeAll(async () => {
+  config.drain_min_delay_ms = 0;
+  config.graceful_drain_enabled = true;
+  stateDirectory = await mkdtemp(join(tmpdir(), 'runner-drain-'));
+  await runHostHook('init');
+  config.drain_public_key = await readFile(join(stateDirectory, 'public.b64'), 'utf8');
+  validateDrainConfiguration();
+  const app = express();
+  app.use('/api/v2', v2Router);
+  server = app.listen(0, '127.0.0.1');
+  await new Promise<void>((resolve) => server.once('listening', resolve));
+  const address = server.address();
+  if (!address || typeof address === 'string') throw new Error('Expected TCP test server');
+  baseUrl = `http://127.0.0.1:${address.port}/api/v2`;
+});
+
+afterEach(() => {
+  resetSandboxDrainForTests();
+  config.graceful_drain_enabled = true;
+  config.drain_min_delay_ms = 0;
+});
+
+afterAll(async () => {
+  config.drain_min_delay_ms = originalMinDelay;
+  config.graceful_drain_enabled = originalEnabled;
+  config.drain_public_key = originalPublicKey;
+  resetSandboxDrainForTests();
+  await new Promise<void>((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
+  await rm(stateDirectory, { recursive: true, force: true });
+});
+
+describe('sandbox drain routes', () => {
+  test('does not activate drain when the feature is disabled', async () => {
+    config.graceful_drain_enabled = false;
+    expect((await fetch(`${baseUrl}/drain`, { method: 'POST' })).status).toBe(404);
+    const response = await fetch(`${baseUrl}/execute`, {
+      method: 'POST', headers: { 'Content-Type': 'application/json' }, body: '{}',
+    });
+    expect(response.status).not.toBe(503);
+  });
+  test('rejects unsigned and foreign-pod drain requests without changing admission', async () => {
+    expect((await fetch(`${baseUrl}/drain`)).status).toBe(404);
+    expect((await fetch(`${baseUrl}/drain`, { method: 'POST' })).status).toBe(403);
+    const foreign = generateKeyPairSync('ed25519');
+    const signature = sign(null, Buffer.from(RUNNER_DRAIN_MESSAGE), foreign.privateKey).toString('base64');
+    expect((await fetch(`${baseUrl}/drain`, {
+      method: 'POST', headers: { 'X-CodeAPI-Drain-Signature': signature },
+    })).status).toBe(403);
+    expect(sandboxDrainSnapshot().draining).toBe(false);
+  });
+  test('advertises enough delay for endpoint propagation and validates the retry bound', async () => {
+    config.drain_min_delay_ms = 5000;
+    validateDrainConfiguration();
+    startSandboxDrain();
+    const response = await fetch(`${baseUrl}/execute`, {
+      method: 'POST', headers: { 'Content-Type': 'application/json' }, body: '{}',
+    });
+    expect(response.headers.get('retry-after')).toBe('5');
+    config.drain_min_delay_ms = 5001;
+    expect(validateDrainConfiguration).toThrow('5-second');
+  });
+  test('fails readiness and rejects execution before validation after drain begins', async () => {
+    // Run the actual host signer and curl hook against the real guest route.
+    expect(JSON.parse(await runHostHook())).toEqual({ status: 'drained', activeExecutions: 0 });
+
+    const readyResponse = await fetch(`${baseUrl}/ready`);
+    expect(readyResponse.status).toBe(503);
+    expect(await readyResponse.json()).toEqual({ status: 'draining', activeExecutions: 0 });
+
+    const executeResponse = await fetch(`${baseUrl}/execute`, {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: '{}',
+    });
+    expect(executeResponse.status).toBe(503);
+    expect(executeResponse.headers.get('retry-after')).toBe(String(config.drain_retry_after_seconds));
+    expect(executeResponse.headers.get('connection')).toBe('close');
+    expect(await executeResponse.json()).toEqual({
+      error: 'runner_draining',
+      message: 'Sandbox runner is draining for shutdown',
+    });
+  });
+});