Skip to content
Merged
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
56 changes: 41 additions & 15 deletions scripts/ci/first-agent-smoke.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -559,7 +559,7 @@ function kubectl(stack, args, options = {}) {
// Answer api.openai.com inside the cluster with the stand-in provider. Agent
// Pods reach it through their ordinary model egress NetworkPolicy and verify
// its certificate against the CA added to the runtime image.
async function routeModelProvider(stack, modelDirectory, runtimeImage) {
async function routeModelProvider(stack, modelDirectory, runtimeImage, credentialDigests) {
await run("docker", ["network", "create", "--subnet", modelNetwork.subnet, modelNetwork.name]);
// Fixed addresses: Docker would otherwise give the node the provider's address.
await run("docker", [
Expand All @@ -570,7 +570,7 @@ async function routeModelProvider(stack, modelDirectory, runtimeImage) {
modelNetwork.name,
`k3d-${stack.state.cluster}-server-0`,
]);
await run("docker", [
const providerArgs = [
"run",
"--detach",
"--name",
Expand All @@ -590,12 +590,17 @@ async function routeModelProvider(stack, modelDirectory, runtimeImage) {
"--env",
`PROBE_ENDPOINT_ECHO_PATTERN=${noncePattern}`,
"--env",
"PROBE_ENDPOINT_CREDENTIAL_DIGESTS",
"--env",
"PROBE_PUBLIC_READ_JOURNEY=1",
"--entrypoint",
"node",
runtimeImage,
"/fixture/endpoint.mjs",
]);
];
await run("docker", providerArgs, {
env: { ...process.env, PROBE_ENDPOINT_CREDENTIAL_DIGESTS: JSON.stringify(credentialDigests) },
});
await waitFor(
"the stand-in provider to listen",
async () => ({ done: (await modelEvents()).some(({ event }) => event === "listening") }),
Expand Down Expand Up @@ -685,12 +690,15 @@ async function modelEvents() {

async function assertModelAnswered(nonce, label) {
const events = await modelEvents();
const answered = events.filter(({ event, text }) => event === "turn-answered" && text === nonce);
const answered = events.find(
({ event, status, text, caller }) =>
event === "turn-answered" && status === 200 && text === nonce && caller === label,
);
assert.ok(
answered.length > 0,
`${label}: the stand-in provider never answered ${nonce}; the reply did not come from a model turn`,
answered,
`${label}: the stand-in provider never answered ${nonce} with this Agent's credential`,
);
return answered[0].transport;
return answered.transport;
}

function cli(stack, args, options = {}) {
Expand Down Expand Up @@ -874,7 +882,7 @@ async function deployAgent(stack, agent) {
return revision.id;
}

async function runFirstAgent(stack) {
async function runFirstAgent(stack, modelCredential) {
const { stdout } = await run(
process.execPath,
[join(root, "scripts/first-agent.mjs"), "smoke-embedded"],
Expand All @@ -884,7 +892,7 @@ async function runFirstAgent(stack) {
OCC_DEVELOPMENT_STATE_DIRECTORY: stack.directory,
OCC_SERVICE_KEY_FILE: stack.keyPath,
OCC_URL: stack.origin,
OPENAI_API_KEY: `sk-oce-smoke-${randomUUID()}`,
OPENAI_API_KEY: modelCredential,
},
timeout: 15 * minute,
label: "first-agent",
Expand All @@ -907,11 +915,11 @@ async function runFirstAgent(stack) {
// The dedicated Codex Agent follows the console's API path: a Namespace Secret
// for the model key, an exact operate grant for the Agent's service principal,
// transport credentials, then deployment.
async function createCodexAgent(stack, reader, label = "codex") {
async function createCodexAgent(stack, reader, label, modelCredential) {
const base = `/namespaces/${stack.namespaceId}`;
const secret = await stack.api("POST", `${base}/secrets`, {
name: `smoke-codex-${randomUUID()}`,
value: `sk-oce-smoke-${randomUUID()}`,
value: modelCredential,
});
const values = createHarnessConfiguration("codex", defaultAgentModel);
if (reader) {
Expand Down Expand Up @@ -1540,6 +1548,15 @@ async function smoke() {
OCC_DEVELOPMENT_STARTUP_TIMEOUT_SECONDS: "900",
};
delete environment.OPENAI_API_KEY;
const modelCredentials = Object.fromEntries(
["embedded", "codex", "reader-control"].map((label) => [label, `sk-oce-smoke-${randomUUID()}`]),
);
const credentialDigests = Object.fromEntries(
Object.entries(modelCredentials).map(([label, value]) => [
label,
createHash("sha256").update(value).digest("hex"),
]),
);
let stack;
try {
stack = await step("Local Setup (occ dev up)", () => startLocalSetup(environment));
Expand All @@ -1551,14 +1568,23 @@ async function smoke() {
enablePublicReader(stack),
);
await step("route api.openai.com to the stand-in provider", () =>
routeModelProvider(stack, modelDirectory, process.env.OCC_KUBERNETES_RUNTIME_IMAGE),
routeModelProvider(
stack,
modelDirectory,
process.env.OCC_KUBERNETES_RUNTIME_IMAGE,
credentialDigests,
),
);
// The two Agents are independent; deploy them together to bound CI time.
const [embedded, codex, control] = await Promise.all([
step("embedded OpenClaw Agent via scripts/first-agent.mjs", () => runFirstAgent(stack)),
step("dedicated Codex Agent via the API", () => createCodexAgent(stack, reader)),
step("embedded OpenClaw Agent via scripts/first-agent.mjs", () =>
runFirstAgent(stack, modelCredentials.embedded),
),
step("dedicated Codex Agent via the API", () =>
createCodexAgent(stack, reader, "codex", modelCredentials.codex),
),
step("selected reader control via the API", () =>
createCodexAgent(stack, reader, "reader-control"),
createCodexAgent(stack, reader, "reader-control", modelCredentials["reader-control"]),
),
]);
await step("dedicated Codex chat turn", () => chatTurn(stack, codex.agent, codex.revisionId));
Expand Down
57 changes: 55 additions & 2 deletions tests/fixtures/runtime-model-probe-endpoint.mjs
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import { createHash } from "node:crypto";
import { readFileSync } from "node:fs";
import { get as httpGet } from "node:http";
import { createServer } from "node:https";
Expand Down Expand Up @@ -26,6 +27,7 @@ const observing = process.env.PROBE_OBSERVE_NATIVE_PORT !== undefined;
const publicReadJourney = process.env.PROBE_PUBLIC_READ_JOURNEY === "1";
const nativePort = Number(process.env.PROBE_OBSERVE_NATIVE_PORT);
const statusPort = Number(process.env.PROBE_OBSERVE_STATUS_PORT);
const credentialLabels = readCredentialLabels(process.env.PROBE_ENDPOINT_CREDENTIAL_DIGESTS);
// Codex opens the Responses API over a WebSocket first and falls back to HTTPS.
// Use the WebSocket implementation the runtime image already ships.
const { WebSocketServer } = createRequire("/app/node_modules/ws/package.json")("ws");
Expand All @@ -37,6 +39,47 @@ const rejection = JSON.stringify({
},
});

function readCredentialLabels(value) {
if (value === undefined) {
return undefined;
}
try {
const parsed = JSON.parse(value);
if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) {
throw new Error();
}
const entries = Object.entries(parsed);
const labels = new Map();
if (entries.length === 0) {
throw new Error();
}
for (const [label, digest] of entries) {
if (
!/^[a-z][a-z0-9-]{0,63}$/.test(label) ||
typeof digest !== "string" ||
!/^[0-9a-f]{64}$/.test(digest) ||
labels.has(digest)
) {
throw new Error();
}
labels.set(digest, label);
}
return labels;
} catch {
throw new Error("Invalid probe endpoint credential digests");
}
}

function callerLabel(request) {
if (credentialLabels === undefined) {
return undefined;
}
const token = /^Bearer (\S+)$/i.exec(request.headers.authorization ?? "")?.[1];
return token === undefined
? null
: (credentialLabels.get(createHash("sha256").update(token).digest("hex")) ?? null);
}

function emit(event) {
process.stdout.write(`${JSON.stringify({ ...event, at: Date.now() })}\n`);
}
Expand Down Expand Up @@ -175,7 +218,7 @@ function publicReadEvents(model, request) {
}

// Answer one model turn after the configured delay, unless the caller left.
function turn(transport, model, request, respond, closed) {
function turn(transport, model, request, respond, closed, caller) {
emit({ event: "request", transport, turn: true });
if (mode === "hang") {
return;
Expand All @@ -190,6 +233,7 @@ function turn(transport, model, request, respond, closed) {
event: "turn-answered",
transport,
status: 200,
...(caller === undefined ? {} : { caller }),
...(echoPattern === undefined ? {} : { text }),
});
}, delayMs);
Expand All @@ -198,6 +242,12 @@ function turn(transport, model, request, respond, closed) {
const server = createServer(
{ key: readFileSync("/fixture/key.pem"), cert: readFileSync("/fixture/cert.pem") },
(request, response) => {
const caller = callerLabel(request);
if (caller === null) {
response.writeHead(401, { "content-type": "application/json" });
response.end(rejection);
return;
}
let body = "";
request.setEncoding("utf8");
request.on("data", (chunk) => {
Expand Down Expand Up @@ -248,6 +298,7 @@ const server = createServer(
body,
(events) => sseResponse(response, events),
() => response.destroyed,
caller,
);
});
},
Expand All @@ -261,7 +312,8 @@ server.on("tlsClientError", (error) => emit({ event: "tls-error", code: error.co

const sockets = new WebSocketServer({ noServer: true });
server.on("upgrade", (request, socket, head) => {
if (mode === "reject") {
const caller = callerLabel(request);
if (caller === null || mode === "reject") {
emit({ event: "request", transport: "websocket", turn: true });
socket.end(
"HTTP/1.1 401 Unauthorized\r\ncontent-type: application/json\r\n" +
Expand Down Expand Up @@ -315,6 +367,7 @@ server.on("upgrade", (request, socket, head) => {
}
},
() => !open,
caller,
);
});
});
Expand Down
Loading