fix: restore codex-api via pod-local forwarder
This commit is contained in:
Codex
@@ -0,0 +1,236 @@
|
||||
#!/usr/bin/env node
|
||||
import { createServer, request as httpRequest } from "node:http";
|
||||
import { request as httpsRequest } from "node:https";
|
||||
|
||||
const serviceId = "hwlab-codex-api-responses-forwarder";
|
||||
const defaultPort = 49280;
|
||||
const port = parsePositiveInteger(process.env.HWLAB_CODE_AGENT_CODEX_API_FORWARDER_PORT || process.env.PORT, defaultPort);
|
||||
const host = firstNonEmpty(process.env.HWLAB_CODE_AGENT_CODEX_API_FORWARDER_HOST, "127.0.0.1");
|
||||
const upstreamBaseUrl = new URL(firstNonEmpty(process.env.HWLAB_CODE_AGENT_CODEX_API_UPSTREAM_BASE_URL, "https://hyueapi.com"));
|
||||
const upstreamNoProxyEntries = noProxyEntriesForHost(upstreamBaseUrl.hostname);
|
||||
const maxBodyBytes = parsePositiveInteger(process.env.HWLAB_CODE_AGENT_CODEX_API_FORWARDER_MAX_BODY_BYTES, 64 * 1024 * 1024);
|
||||
|
||||
const hopByHopHeaders = new Set([
|
||||
"connection",
|
||||
"keep-alive",
|
||||
"proxy-authenticate",
|
||||
"proxy-authorization",
|
||||
"te",
|
||||
"trailer",
|
||||
"transfer-encoding",
|
||||
"upgrade"
|
||||
]);
|
||||
|
||||
ensureNoProxyEntries(upstreamNoProxyEntries);
|
||||
|
||||
const server = createServer(async (clientReq, clientRes) => {
|
||||
try {
|
||||
const url = new URL(clientReq.url || "/", "http://hwlab-codex-api-forwarder.local");
|
||||
if (url.pathname === "/health" || url.pathname === "/health/live" || url.pathname === "/health/liveliness") {
|
||||
sendJson(clientRes, 200, healthPayload("ok"));
|
||||
return;
|
||||
}
|
||||
if (url.pathname === "/health/readiness") {
|
||||
sendJson(clientRes, 200, healthPayload("ready"));
|
||||
return;
|
||||
}
|
||||
await proxyRequest(clientReq, clientRes, url);
|
||||
} catch (error) {
|
||||
sendJson(clientRes, 502, {
|
||||
error: {
|
||||
code: "codex_api_forwarder_failed",
|
||||
message: "Codex API Responses forwarder failed to proxy request",
|
||||
reason: error.message
|
||||
}
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
server.listen(port, host, () => {
|
||||
process.stdout.write(JSON.stringify({
|
||||
serviceId,
|
||||
status: "listening",
|
||||
host,
|
||||
port,
|
||||
upstreamHost: upstreamBaseUrl.hostname,
|
||||
noProxyRequired: upstreamNoProxyEntries
|
||||
}) + "\n");
|
||||
});
|
||||
|
||||
async function proxyRequest(clientReq, clientRes, url) {
|
||||
const startedAt = Date.now();
|
||||
const targetUrl = targetUrlForRequest(url);
|
||||
const headers = filterRequestHeaders(clientReq.headers, targetUrl);
|
||||
const requestImpl = targetUrl.protocol === "https:" ? httpsRequest : httpRequest;
|
||||
|
||||
const upstreamReq = requestImpl({
|
||||
protocol: targetUrl.protocol,
|
||||
hostname: targetUrl.hostname,
|
||||
port: targetUrl.port || (targetUrl.protocol === "https:" ? 443 : 80),
|
||||
method: clientReq.method,
|
||||
path: `${targetUrl.pathname}${targetUrl.search}`,
|
||||
headers
|
||||
}, (upstreamRes) => {
|
||||
const responseHeaders = filterResponseHeaders(upstreamRes.headers);
|
||||
clientRes.writeHead(upstreamRes.statusCode || 502, responseHeaders);
|
||||
let responseBytes = 0;
|
||||
upstreamRes.on("data", (chunk) => {
|
||||
responseBytes += chunk.length;
|
||||
});
|
||||
upstreamRes.on("end", () => {
|
||||
process.stderr.write(JSON.stringify({
|
||||
serviceId,
|
||||
method: clientReq.method,
|
||||
path: url.pathname,
|
||||
upstreamPath: targetUrl.pathname,
|
||||
status: upstreamRes.statusCode || null,
|
||||
responseBytes,
|
||||
elapsedMs: Date.now() - startedAt
|
||||
}) + "\n");
|
||||
});
|
||||
upstreamRes.pipe(clientRes);
|
||||
});
|
||||
|
||||
upstreamReq.on("error", (error) => {
|
||||
if (!clientRes.headersSent) {
|
||||
sendJson(clientRes, 502, {
|
||||
error: {
|
||||
code: "codex_api_forwarder_upstream_error",
|
||||
message: "Codex API forwarder upstream request failed",
|
||||
reason: error.message
|
||||
}
|
||||
});
|
||||
} else {
|
||||
clientRes.end();
|
||||
}
|
||||
});
|
||||
|
||||
await pipeRequestBody(clientReq, upstreamReq);
|
||||
}
|
||||
|
||||
async function pipeRequestBody(clientReq, upstreamReq) {
|
||||
let bytes = 0;
|
||||
for await (const chunk of clientReq) {
|
||||
bytes += chunk.length;
|
||||
if (bytes > maxBodyBytes) {
|
||||
upstreamReq.destroy(new Error(`request body exceeded ${maxBodyBytes} bytes`));
|
||||
return;
|
||||
}
|
||||
if (!upstreamReq.write(chunk)) {
|
||||
await new Promise((resolve) => upstreamReq.once("drain", resolve));
|
||||
}
|
||||
}
|
||||
upstreamReq.end();
|
||||
}
|
||||
|
||||
function targetUrlForRequest(url) {
|
||||
const target = new URL(upstreamBaseUrl.toString());
|
||||
target.pathname = upstreamPathFor(url.pathname);
|
||||
target.search = url.search;
|
||||
target.hash = "";
|
||||
return target;
|
||||
}
|
||||
|
||||
function upstreamPathFor(inboundPath) {
|
||||
const basePath = upstreamBaseUrl.pathname.replace(/\/+$/u, "");
|
||||
const requestPath = normalizeCodexResponsesPath(inboundPath || "/");
|
||||
if (!basePath) return requestPath;
|
||||
if (basePath === "/v1" && requestPath.startsWith("/v1/")) return requestPath;
|
||||
if (basePath === "/v1" && requestPath === "/v1") return requestPath;
|
||||
if (basePath.endsWith("/responses") && (requestPath === "/responses" || requestPath === "/v1/responses")) return basePath;
|
||||
if (basePath.endsWith("/models") && (requestPath === "/models" || requestPath === "/v1/models")) return basePath;
|
||||
if (requestPath.startsWith(`${basePath}/`) || requestPath === basePath) return requestPath;
|
||||
return `${basePath}${requestPath}`;
|
||||
}
|
||||
|
||||
function normalizeCodexResponsesPath(pathname) {
|
||||
if (upstreamBaseUrl.pathname.replace(/\/+$/u, "") === "/v1") return pathname;
|
||||
if (pathname === "/v1/responses") return "/responses";
|
||||
if (pathname === "/v1/models") return "/models";
|
||||
return pathname;
|
||||
}
|
||||
|
||||
function filterRequestHeaders(headers, targetUrl) {
|
||||
const filtered = {};
|
||||
for (const [name, value] of Object.entries(headers)) {
|
||||
const lower = name.toLowerCase();
|
||||
if (hopByHopHeaders.has(lower) || lower === "host") continue;
|
||||
filtered[name] = value;
|
||||
}
|
||||
filtered.host = targetUrl.host;
|
||||
return filtered;
|
||||
}
|
||||
|
||||
function filterResponseHeaders(headers) {
|
||||
const filtered = {};
|
||||
for (const [name, value] of Object.entries(headers)) {
|
||||
if (hopByHopHeaders.has(name.toLowerCase())) continue;
|
||||
filtered[name] = value;
|
||||
}
|
||||
return filtered;
|
||||
}
|
||||
|
||||
function sendJson(res, status, payload) {
|
||||
const body = JSON.stringify(payload);
|
||||
res.writeHead(status, { "content-type": "application/json", "content-length": Buffer.byteLength(body) });
|
||||
res.end(body);
|
||||
}
|
||||
|
||||
function healthPayload(status) {
|
||||
return {
|
||||
serviceId,
|
||||
status,
|
||||
listen: { host, port },
|
||||
upstream: {
|
||||
origin: upstreamBaseUrl.origin,
|
||||
host: upstreamBaseUrl.hostname,
|
||||
directNoProxyRequired: true,
|
||||
noProxyRequired: upstreamNoProxyEntries
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
function ensureNoProxyEntries(entries) {
|
||||
for (const name of ["NO_PROXY", "no_proxy"]) {
|
||||
process.env[name] = mergeNoProxyEntries(process.env[name], entries);
|
||||
}
|
||||
}
|
||||
|
||||
function noProxyEntriesForHost(hostname) {
|
||||
const hostText = String(hostname || "").trim().toLowerCase();
|
||||
if (!hostText) return ["127.0.0.1", "localhost", "::1"];
|
||||
const entries = [hostText, "127.0.0.1", "localhost", "::1"];
|
||||
if (!/^\d+(?:\.\d+){3}$/u.test(hostText) && !hostText.includes(":")) entries.push(`.${hostText}`);
|
||||
return uniqueStrings(entries);
|
||||
}
|
||||
|
||||
function mergeNoProxyEntries(current, additions = []) {
|
||||
return uniqueStrings([...String(current ?? "").split(/[,;\s]+/u), ...additions]).join(",");
|
||||
}
|
||||
|
||||
function uniqueStrings(values) {
|
||||
const seen = new Set();
|
||||
const result = [];
|
||||
for (const value of values) {
|
||||
const text = String(value ?? "").trim();
|
||||
if (!text) continue;
|
||||
const key = text.toLowerCase();
|
||||
if (seen.has(key)) continue;
|
||||
seen.add(key);
|
||||
result.push(text);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
function firstNonEmpty(...values) {
|
||||
for (const value of values) {
|
||||
const text = String(value ?? "").trim();
|
||||
if (text) return text;
|
||||
}
|
||||
return "";
|
||||
}
|
||||
|
||||
function parsePositiveInteger(value, fallback) {
|
||||
const parsed = Number.parseInt(value || "", 10);
|
||||
return Number.isInteger(parsed) && parsed > 0 ? parsed : fallback;
|
||||
}
|
||||
@@ -0,0 +1,118 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { spawn } from "node:child_process";
|
||||
import { createServer } from "node:http";
|
||||
import test from "node:test";
|
||||
|
||||
test("Codex API Responses forwarder maps loopback Codex paths to configurable upstream and preserves stream body", async () => {
|
||||
const upstream = await startUpstream();
|
||||
const forwarder = await startForwarder(upstream.port, { upstreamPath: "/" });
|
||||
try {
|
||||
const response = await fetch(`http://127.0.0.1:${forwarder.port}/v1/responses`, {
|
||||
method: "POST",
|
||||
headers: {
|
||||
"content-type": "application/json",
|
||||
authorization: "Bearer test-secret"
|
||||
},
|
||||
body: JSON.stringify({ model: "gpt-5.5", input: "OK" })
|
||||
});
|
||||
assert.equal(response.status, 200);
|
||||
const payload = await response.json();
|
||||
assert.equal(payload.ok, true);
|
||||
assert.equal(upstream.captured.url, "/responses");
|
||||
assert.equal(upstream.captured.authorization, "Bearer test-secret");
|
||||
assert.deepEqual(JSON.parse(upstream.captured.body), { model: "gpt-5.5", input: "OK" });
|
||||
|
||||
const health = await fetch(`http://127.0.0.1:${forwarder.port}/health/live`).then((item) => item.json());
|
||||
assert.equal(health.upstream.host, "127.0.0.1");
|
||||
assert.equal(health.upstream.directNoProxyRequired, true);
|
||||
} finally {
|
||||
await forwarder.stop();
|
||||
await upstream.stop();
|
||||
}
|
||||
});
|
||||
|
||||
test("Codex API Responses forwarder keeps /v1 prefix when upstream base is /v1", async () => {
|
||||
const upstream = await startUpstream();
|
||||
const forwarder = await startForwarder(upstream.port, { upstreamPath: "/v1" });
|
||||
try {
|
||||
const response = await fetch(`http://127.0.0.1:${forwarder.port}/responses`, {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ input: "OK" })
|
||||
});
|
||||
assert.equal(response.status, 200);
|
||||
assert.equal(upstream.captured.url, "/v1/responses");
|
||||
} finally {
|
||||
await forwarder.stop();
|
||||
await upstream.stop();
|
||||
}
|
||||
});
|
||||
|
||||
async function startUpstream() {
|
||||
let captured = null;
|
||||
const server = createServer(async (request, response) => {
|
||||
const chunks = [];
|
||||
for await (const chunk of request) chunks.push(chunk);
|
||||
captured = {
|
||||
url: request.url,
|
||||
method: request.method,
|
||||
authorization: request.headers.authorization || null,
|
||||
body: Buffer.concat(chunks).toString("utf8")
|
||||
};
|
||||
response.writeHead(200, { "content-type": "application/json" });
|
||||
response.end(JSON.stringify({ ok: true }));
|
||||
});
|
||||
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
|
||||
return {
|
||||
port: server.address().port,
|
||||
get captured() {
|
||||
return captured;
|
||||
},
|
||||
stop: () => new Promise((resolve) => server.close(resolve))
|
||||
};
|
||||
}
|
||||
|
||||
async function startForwarder(upstreamPort, { upstreamPath }) {
|
||||
const port = await freePort();
|
||||
const child = spawn(process.execPath, ["cmd/hwlab-codex-api-responses-forwarder/main.mjs"], {
|
||||
cwd: process.cwd(),
|
||||
env: {
|
||||
...process.env,
|
||||
HWLAB_CODE_AGENT_CODEX_API_FORWARDER_PORT: String(port),
|
||||
HWLAB_CODE_AGENT_CODEX_API_UPSTREAM_BASE_URL: `http://127.0.0.1:${upstreamPort}${upstreamPath}`
|
||||
},
|
||||
stdio: ["ignore", "pipe", "pipe"]
|
||||
});
|
||||
let stderr = "";
|
||||
child.stderr.on("data", (chunk) => {
|
||||
stderr += chunk;
|
||||
});
|
||||
await new Promise((resolve, reject) => {
|
||||
const timer = setTimeout(() => reject(new Error(`forwarder did not start: ${stderr}`)), 5000);
|
||||
child.stdout.on("data", (chunk) => {
|
||||
if (chunk.toString().includes("listening")) {
|
||||
clearTimeout(timer);
|
||||
resolve();
|
||||
}
|
||||
});
|
||||
child.on("exit", (code) => {
|
||||
clearTimeout(timer);
|
||||
reject(new Error(`forwarder exited early code=${code}: ${stderr}`));
|
||||
});
|
||||
});
|
||||
return {
|
||||
port,
|
||||
stop: async () => {
|
||||
child.kill("SIGTERM");
|
||||
await new Promise((resolve) => child.once("exit", resolve));
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
async function freePort() {
|
||||
const server = createServer();
|
||||
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
|
||||
const port = server.address().port;
|
||||
await new Promise((resolve) => server.close(resolve));
|
||||
return port;
|
||||
}
|
||||
Reference in New Issue
Block a user