Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion Dockerfile.runtime
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ ARG SOURCE_COMMIT
ENV DIFFEX_SOURCE_COMMIT=$SOURCE_COMMIT

COPY package.json package-lock.json tsconfig.base.json ./
COPY LICENSE ./
COPY LICENSE AGENTS_release.md ./
COPY legal ./legal
COPY docs/release/distribution-policy.json ./docs/release/distribution-policy.json
COPY packages ./packages
Expand Down
9 changes: 7 additions & 2 deletions docs/cloud-runtime.md
Original file line number Diff line number Diff line change
Expand Up @@ -19,8 +19,11 @@ The cloud relay preserves byte order, applies backpressure and limits, and does
The WorkOS access token stays at the control-plane boundary.
Provider credentials, sandbox identifiers, and internal runtime addresses never enter the Diffex wire protocol.

The production model credential is mounted through Daytona Secrets with a provider-host allowlist.
Only Daytona's opaque placeholder enters the sandbox environment, and the real provider credential is substituted into outbound HTTPS headers outside the sandbox.
Hosted inference uses the authenticated cloud model gateway.
The runtime receives a session-scoped gateway credential and URL, while provider credentials remain in the gateway host.
Every model call, including compaction, branch summaries, and child agents, uses the same gateway transport.
The transport saves a private operation UUID and request digest before sending, then disables automatic provider and agent retries after a failure.
Pending records survive restart for diagnosis and are never automatically regenerated or replayed.

## Runtime Image

Expand All @@ -45,6 +48,8 @@ The production runtime service must:
- Convert runtime events through the existing protocol adapters
- Reject conflicting session mutations instead of silently queuing them
- Shut down cleanly when the sandbox receives a termination signal
- Receive `DIFFEX_MODEL_GATEWAY_URL` and `DIFFEX_MODEL_GATEWAY_CREDENTIAL`, and reject direct provider-key bootstrap
- Persist private model-operation records under `DIFFEX_RUNTIME_STATE_DIR`

## Compatibility

Expand Down
9 changes: 7 additions & 2 deletions packages/coding-agent-server/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,12 @@ It does not own WorkOS authentication, billing, cloud policy, sandbox allocation

The `diffex-runtime-server` executable starts the service with an authenticated WebSocket listener.
The cloud bootstrap supplies `DIFFEX_SESSION_ID`, `DIFFEX_WORKSPACE`, and an internal `DIFFEX_RUNTIME_SHARED_SECRET` of at least 32 bytes.
It also supplies `DIFFEX_MODEL_PROVIDER`, `DIFFEX_MODEL_ID`, and `DIFFEX_MODEL_API_KEY` through the sandbox provider's secret-injection mechanism.
The environment value is an opaque Daytona placeholder, the real provider key never enters the sandbox, and the placeholder is removed from the runtime process environment after the in-memory model client is configured.
It also supplies `DIFFEX_MODEL_PROVIDER`, `DIFFEX_MODEL_ID`, `DIFFEX_MODEL_MAX_OUTPUT_TOKENS`, `DIFFEX_MODEL_GATEWAY_URL`, and `DIFFEX_MODEL_GATEWAY_CREDENTIAL`.
The gateway credential authorizes only this cloud session's model calls and is removed from the process environment after the in-memory transport is configured.
The process rejects a direct provider API key and uses a single configured Anthropic model.
Before each model request, the transport persists its operation UUID and request digest in a private directory under `DIFFEX_RUNTIME_STATE_DIR`.
Provider and agent retries are disabled for gateway failures, including timeouts and dropped streams, because automatically starting another generation could duplicate an already dispatched operation.
Pending operation metadata survives restart for diagnostics without replaying the original request.
Completed model usage is written to a durable sandbox outbox before it is reported to `DIFFEX_USAGE_ENDPOINT`.
These sandbox usage reports are telemetry; the cloud gateway owns authoritative token measurements.
Public WorkOS tokens and cloud attachment tickets are consumed by the control plane and are never passed to this process.
19 changes: 15 additions & 4 deletions packages/coding-agent-server/src/cli.ts
Original file line number Diff line number Diff line change
Expand Up @@ -19,8 +19,20 @@ const path = process.env.DIFFEX_RUNTIME_PATH ?? "/runtime";
const agentDir = process.env.DIFFEX_AGENT_DIR;
const modelProvider = requiredEnvironment("DIFFEX_MODEL_PROVIDER");
const modelId = requiredEnvironment("DIFFEX_MODEL_ID");
const modelApiKey = requiredEnvironment("DIFFEX_MODEL_API_KEY");
const runtimeStateDirectory = requiredEnvironment("DIFFEX_RUNTIME_STATE_DIR");
const gatewayUrl = requiredEnvironment("DIFFEX_MODEL_GATEWAY_URL");
const gatewayCredential = requiredEnvironment("DIFFEX_MODEL_GATEWAY_CREDENTIAL");
const maxOutputTokens = Number(requiredEnvironment("DIFFEX_MODEL_MAX_OUTPUT_TOKENS"));
if (process.env.DIFFEX_MODEL_API_KEY) throw new Error("Hosted runtime cannot receive a direct provider API key");
const modelRuntime = await ModelRuntime.create({
hostedGateway: {
baseUrl: gatewayUrl,
credential: gatewayCredential,
provider: modelProvider,
modelId,
maxOutputTokens,
stateDirectory: runtimeStateDirectory,
},
...(agentDir
? {
authPath: join(agentDir, "auth.json"),
Expand All @@ -29,10 +41,9 @@ const modelRuntime = await ModelRuntime.create({
: {}),
refreshOnCreate: false,
});
await modelRuntime.setRuntimeApiKey(modelProvider, modelApiKey);
delete process.env.DIFFEX_MODEL_API_KEY;
await modelRuntime.refresh({ allowNetwork: false });
delete process.env.DIFFEX_MODEL_GATEWAY_CREDENTIAL;
delete process.env.DIFFEX_RUNTIME_SHARED_SECRET;
const runtimeStateDirectory = requiredEnvironment("DIFFEX_RUNTIME_STATE_DIR");
const usageReporter = createDurableUsageReporter({
endpoint: requiredEnvironment("DIFFEX_USAGE_ENDPOINT"),
sharedSecret,
Expand Down
203 changes: 203 additions & 0 deletions packages/coding-agent-server/test/gateway-runtime-process.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,203 @@
import { type ChildProcess, spawn } from "node:child_process";
import { once } from "node:events";
import { mkdir, mkdtemp, readdir, readFile, realpath, rm } from "node:fs/promises";
import { createServer } from "node:http";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { fileURLToPath } from "node:url";
import { DiffexClient } from "@diffexai/diffex-client";
import { createWebSocketTransportFactory } from "@diffexai/diffex-client/websocket";
import { describe, expect, test } from "vitest";
import { WebSocket } from "ws";

const sessionId = "2f1c9e00-aaaa-4000-8000-000000000001";
const credential = "runtime-gateway-local-fixture";
const secret = "runtime-protocol-local-fixture-32-bytes";

function completion(): string {
const events = [
{
type: "message_start",
message: {
id: "msg_runtime_fixture",
type: "message",
role: "assistant",
model: "claude-sonnet-4-5",
content: [],
stop_reason: null,
stop_sequence: null,
usage: { input_tokens: 10, output_tokens: 0, cache_read_input_tokens: 0, cache_creation_input_tokens: 0 },
},
},
{ type: "content_block_start", index: 0, content_block: { type: "text", text: "" } },
{ type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "gateway-runtime-ok" } },
{ type: "content_block_stop", index: 0 },
{
type: "message_delta",
delta: { stop_reason: "end_turn", stop_sequence: null, stop_details: null },
usage: { output_tokens: 3 },
},
{ type: "message_stop" },
];
return events.map((event) => `event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`).join("");
}

describe("hosted gateway runtime process", () => {
test("uses gateway authentication over native WebSocket, preserves pending calls, and reopens without redispatch", async () => {
const root = await realpath(await mkdtemp(join(tmpdir(), "gateway-runtime-process-")));
for (const name of ["workspace", "agent", "state", "sessions"]) await mkdir(join(root, name));
const requests: Array<{ operationId: string; body: Record<string, unknown> }> = [];
let responseMode = "complete";
const gateway = createServer(async (request, response) => {
try {
if (request.url === "/usage") {
response.writeHead(200, { "content-type": "application/json" });
response.end('{"accepted":true}');
return;
}
expect(request.url).toBe(`/v1/runtime/sessions/${sessionId}/model/v1/messages`);
expect(request.headers["x-api-key"]).toBe(credential);
expect(request.headers.authorization).toBeUndefined();
const operationId = String(request.headers["x-diffex-operation-id"]);
const record = JSON.parse(
await readFile(join(root, "state", "model-operations", `${operationId}.json`), "utf8"),
);
expect(record.operationId).toBe(operationId);
let body = "";
for await (const chunk of request) body += chunk;
requests.push({ operationId, body: JSON.parse(body) });
if (responseMode === "complete") {
response.writeHead(200, { "content-type": "text/event-stream" });
response.end(completion());
} else {
response.writeHead(503, { "content-type": "application/json" });
response.end('{"type":"error","error":{"type":"api_error","message":"gateway_unresolved"}}');
}
} catch (error) {
response.writeHead(500);
response.end(String(error));
}
});
gateway.listen(0, "127.0.0.1");
await once(gateway, "listening");
const gatewayAddress = gateway.address();
if (!gatewayAddress || typeof gatewayAddress === "string") throw new Error("Missing gateway address");
const reservation = createServer();
reservation.listen(0, "127.0.0.1");
await once(reservation, "listening");
const runtimeAddress = reservation.address();
if (!runtimeAddress || typeof runtimeAddress === "string") throw new Error("Missing runtime address");
await new Promise<void>((resolve, reject) => reservation.close((error) => (error ? reject(error) : resolve())));
let child: ChildProcess | undefined;
let exit: Promise<unknown[]> | undefined;
const launch = async (): Promise<void> => {
child = spawn(process.execPath, [fileURLToPath(new URL("../dist/cli.js", import.meta.url))], {
cwd: join(root, "workspace"),
stdio: ["ignore", "pipe", "pipe"],
env: {
PATH: process.env.PATH,
HOME: root,
DIFFEX_OFFLINE: "1",
HOST: "127.0.0.1",
PORT: String(runtimeAddress.port),
DIFFEX_SESSION_ID: sessionId,
DIFFEX_WORKSPACE: join(root, "workspace"),
DIFFEX_AGENT_DIR: join(root, "agent"),
DIFFEX_CODING_AGENT_DIR: join(root, "agent"),
DIFFEX_RUNTIME_STATE_DIR: join(root, "state"),
DIFFEX_SESSION_DIR: join(root, "sessions"),
DIFFEX_RUNTIME_SHARED_SECRET: secret,
DIFFEX_OWNER_USER_ID: "fixture-user",
DIFFEX_ORG_ID: "fixture-org",
DIFFEX_MODEL_PROVIDER: "anthropic",
DIFFEX_MODEL_ID: "claude-sonnet-4-5",
DIFFEX_MODEL_MAX_OUTPUT_TOKENS: "8192",
DIFFEX_MODEL_GATEWAY_URL: `http://127.0.0.1:${gatewayAddress.port}/v1/runtime/sessions/${sessionId}/model`,
DIFFEX_MODEL_GATEWAY_CREDENTIAL: credential,
DIFFEX_USAGE_ENDPOINT: `http://127.0.0.1:${gatewayAddress.port}/usage`,
},
});
exit = once(child, "exit");
await new Promise<void>((resolve, reject) => {
let output = "";
child?.once("error", reject);
child?.once("exit", () => reject(new Error(`Runtime exited before readiness: ${output}`)));
child?.stderr?.on("data", (chunk) => {
output += chunk;
});
child?.stdout?.on("data", (chunk) => {
output += chunk;
if (output.includes("Diffex runtime server listening")) resolve();
});
});
};
const stop = async (): Promise<void> => {
child?.kill("SIGTERM");
expect(await exit).toEqual([0, null]);
};
const connect = (candidate: string) =>
DiffexClient.connect({
transportFactory: createWebSocketTransportFactory({
url: `ws://127.0.0.1:${runtimeAddress.port}/runtime`,
webSocket: class extends WebSocket {
constructor(url: string | URL, protocols?: string | string[]) {
super(url, protocols, { headers: { authorization: `Bearer ${candidate}` } });
}
} as never,
}),
});
try {
await launch();
await expect(connect("wrong")).rejects.toThrow();
const client = await connect(secret);
try {
await expect(client.attachSession("another-session")).rejects.toThrow();
const session = await client.attachSession(sessionId);
await expect(session.prompt("Reply with fixture text")).resolves.toMatchObject({
phase: "idle",
transcript: expect.arrayContaining([
expect.objectContaining({
role: "assistant",
content: [{ type: "text", text: "gateway-runtime-ok" }],
status: "complete",
usage: expect.objectContaining({ input: 10, output: 3 }),
}),
]),
});
expect(requests).toHaveLength(1);
expect(requests[0]?.body).toMatchObject({ model: "claude-sonnet-4-5", stream: true, max_tokens: 8192 });
responseMode = "unresolved";
const failed = await session.prompt("A distinct second call");
expect(failed.transcript.at(-1)).toMatchObject({
status: "error",
errorMessage: expect.stringContaining("gateway_unresolved"),
});
expect(requests).toHaveLength(2);
expect(requests[0]?.operationId).not.toBe(requests[1]?.operationId);
expect(await readdir(join(root, "state", "model-operations"))).toHaveLength(1);
await session.dispose();
} finally {
await client.dispose();
}
await stop();
await launch();
const restored = await connect(secret);
try {
const session = await restored.attachSession(sessionId);
expect(session.snapshot?.transcript.at(-1)).toMatchObject({ status: "error" });
expect(requests).toHaveLength(2);
await session.dispose();
} finally {
await restored.dispose();
}
await stop();
} finally {
if (child && child.exitCode === null && child.signalCode === null) {
child.kill("SIGKILL");
await exit;
}
await new Promise<void>((resolve, reject) => gateway.close((error) => (error ? reject(error) : resolve())));
await rm(root, { recursive: true, force: true });
}
});
});
2 changes: 1 addition & 1 deletion scripts/check-runtime-image.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ def inspect_image(filename):
if config['architecture'] != 'amd64' or config['os'] != 'linux':
raise ValueError('Expected Linux amd64 runtime')
for layer in manifest[0]['Layers']:
with tarfile.open(fileobj=image.extractfile(layer), mode='r|') as archive:
with tarfile.open(fileobj=image.extractfile(layer), mode='r|*') as archive:
for entry in archive:
name = entry.name.removeprefix('./')
if not name.startswith('opt/diffex/') or not entry.isfile():
Expand Down
89 changes: 33 additions & 56 deletions scripts/fixtures/runtime-fixture.mjs
Original file line number Diff line number Diff line change
@@ -1,8 +1,7 @@
// Mounted into acceptance containers only; never included in an image layer.
import { appendFileSync } from "node:fs";
import { appendFileSync, existsSync, readFileSync } from "node:fs";
import { Socket } from "node:net";
import { ModelRuntime } from "/opt/diffex/packages/coding-agent/dist/index.js";
import { createAssistantMessageEventStream } from "/opt/diffex/packages/ai/dist/index.js";
import { join } from "node:path";
import "/opt/diffex/packages/coding-agent/dist/core/http-dispatcher.js";

const connect = Socket.prototype.connect;
Expand All @@ -12,57 +11,35 @@ Socket.prototype.connect = function (...args) {
if (host !== "127.0.0.1" && host !== "::1") throw new Error("Runtime smoke blocked external socket");
return connect.apply(this, args);
};
globalThis.fetch = async (input) => {
if (String(input) !== "http://127.0.0.1:1/usage") throw new Error("Runtime smoke blocked external fetch");
appendFileSync(process.env.DIFFEX_SMOKE_EVENTS, "usage\n");
return Response.json({ accepted: true });
};
const create = ModelRuntime.create.bind(ModelRuntime);
ModelRuntime.create = async (...args) => {
const runtime = await create(...args);
runtime.registerProvider("runtime-fixture", {
baseUrl: "https://fixture.invalid",
apiKey: "local-fixture",
api: "runtime-fixture",
models: [
{
id: "fixture",
name: "Runtime fixture",
reasoning: false,
input: ["text"],
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
contextWindow: 8192,
maxTokens: 128,
},
],
streamSimple(model) {
appendFileSync(process.env.DIFFEX_SMOKE_EVENTS, "inference\n");
const stream = createAssistantMessageEventStream();
const message = {
role: "assistant",
content: [{ type: "text", text: "runtime-fixture-ok" }],
api: model.api,
provider: model.provider,
model: model.id,
usage: {
input: 1,
output: 1,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 2,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
stopReason: "stop",
timestamp: Date.now(),
};
queueMicrotask(() => {
stream.push({ type: "start", partial: message });
stream.push({ type: "text_delta", contentIndex: 0, delta: "runtime-fixture-ok", partial: message });
stream.push({ type: "done", reason: "stop", message });
stream.end(message);
});
return stream;
},
});
return runtime;
globalThis.fetch = async (input, init) => {
const request = new Request(input, init);
if (request.url === "http://127.0.0.1:1/usage") {
appendFileSync(process.env.DIFFEX_SMOKE_EVENTS, "usage\n");
return Response.json({ accepted: true });
}
if (request.url !== `${process.env.DIFFEX_MODEL_GATEWAY_URL}/v1/messages`) {
throw new Error("Runtime smoke blocked external fetch");
}
if (request.headers.get("x-api-key") !== "runtime-local-gateway-fixture") {
throw new Error("Runtime smoke rejected gateway credential");
}
const operationId = request.headers.get("x-diffex-operation-id");
const pending = join(process.env.DIFFEX_RUNTIME_STATE_DIR, "model-operations", `${operationId}.json`);
if (!operationId || !existsSync(pending) || JSON.parse(readFileSync(pending, "utf8")).operationId !== operationId) {
throw new Error("Runtime smoke did not persist its operation before inference");
}
appendFileSync(process.env.DIFFEX_SMOKE_EVENTS, "inference\n");
const events = [
{ type: "message_start", message: { id: "msg_fixture", type: "message", role: "assistant",
model: process.env.DIFFEX_MODEL_ID, content: [], stop_reason: null, stop_sequence: null,
usage: { input_tokens: 1, output_tokens: 0, cache_read_input_tokens: 0, cache_creation_input_tokens: 0 } } },
{ type: "content_block_start", index: 0, content_block: { type: "text", text: "" } },
{ type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "runtime-fixture-ok" } },
{ type: "content_block_stop", index: 0 },
{ type: "message_delta", delta: { stop_reason: "end_turn", stop_sequence: null, stop_details: null },
usage: { output_tokens: 1 } },
{ type: "message_stop" },
];
return new Response(events.map((event) => `event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`).join(""),
{ headers: { "content-type": "text/event-stream" } });
};
Loading
Loading