f3e21c6b98
Accumulated uncommitted work: GHOST Web backend, WebGL/GPU web layer, web workflow engine and protocol, tooling, docs and task records. Large generated evidence, user-testing archives, build snapshots and imported test assets remain untracked. Co-Authored-By: Claude Code <noreply@anthropic.com>
236 lines
18 KiB
JavaScript
236 lines
18 KiB
JavaScript
import fs from "node:fs";
|
|
import os from "node:os";
|
|
import path from "node:path";
|
|
import crypto from "node:crypto";
|
|
import http from "node:http";
|
|
import { fileURLToPath } from "node:url";
|
|
import { startBoundedServerJobProcess, cancelServerJobProcess, validateServerJobBudget } from "./server-job-process.mjs";
|
|
import { handleNativeWorkflowRequest } from "./native-workflow-server.mjs";
|
|
|
|
const ROOT = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "../..");
|
|
const RUNNER = path.join(ROOT, "tools/web/server-operator-job.py");
|
|
const MAGIC = Buffer.from("WBJ1");
|
|
const MAX_ENVELOPE = 513 * 1024 * 1024;
|
|
const jobs = new Map();
|
|
const operatorSchemaCache = new Map();
|
|
|
|
function fail(message, status = 400) { const error = new Error(message); error.statusCode = status; error.errorCode = message; throw error; }
|
|
function digest(bytes) { return crypto.createHash("sha256").update(bytes).digest("hex"); }
|
|
function executableFingerprint(binary) { try { const resolved = path.isAbsolute(binary) ? binary : (process.env.PATH ?? "").split(path.delimiter).map((dir) => path.join(dir, binary)).find((candidate) => { try { fs.accessSync(candidate, fs.constants.X_OK); return true; } catch { return false; } }) ?? binary; const stat = fs.statSync(resolved); return `${resolved}:${stat.size}:${stat.mtimeMs}`; } catch { return String(binary); } }
|
|
function stableJSON(value) {
|
|
if (value === null || typeof value === "string" || typeof value === "boolean" || typeof value === "number") return JSON.stringify(value);
|
|
if (Array.isArray(value)) return `[${value.map(stableJSON).join(",")}]`;
|
|
return `{${Object.keys(value).sort().map((key) => `${JSON.stringify(key)}:${stableJSON(value[key])}`).join(",")}}`;
|
|
}
|
|
function validateBindings(request) {
|
|
const allowed = new Set(["schemaVersion", "route", "jobId", "sourceBlendSha256", "sourceBlendByteLength", "sourceRevision", "blenderBuild", "operator", "properties", "propertiesSha256", "timeoutMs", "memoryBytes", "resultBinding", "context", "requestSha256"]);
|
|
if (Object.keys(request).some((key) => !allowed.has(key))) fail("SERVER_OPERATOR_HTTP_REQUEST_INVALID");
|
|
if (request.schemaVersion !== 1 || request.route !== "SERVER_NATIVE" || typeof request.jobId !== "string" || !/^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$/u.test(request.jobId) || typeof request.operator !== "string" || !/^[A-Za-z_][A-Za-z0-9_.-]{0,127}$/u.test(request.operator) || typeof request.requestSha256 !== "string" || !request.properties || typeof request.properties !== "object" || Array.isArray(request.properties)) fail("SERVER_OPERATOR_HTTP_REQUEST_INVALID");
|
|
if (!request.blenderBuild || typeof request.blenderBuild !== "object" || !/^[a-f0-9]{64}$/iu.test(String(request.blenderBuild.buildSha256 ?? "")) || !request.resultBinding || typeof request.resultBinding !== "object" || !["json", "snapshot", "scene-delta"].includes(request.resultBinding.kind) || typeof request.resultBinding.key !== "string" || !/^[A-Za-z][A-Za-z0-9_.-]{0,127}$/u.test(request.resultBinding.key)) fail("SERVER_OPERATOR_HTTP_REQUEST_INVALID");
|
|
if (!Number.isSafeInteger(request.timeoutMs) || request.timeoutMs < 1000 || request.timeoutMs > 600000 || !Number.isSafeInteger(request.memoryBytes) || request.memoryBytes < 64 * 1024 * 1024 || request.memoryBytes > 4 * 1024 * 1024 * 1024 || Object.keys(request.properties).length > 128 || Object.keys(request.properties).some((key) => !/^[A-Za-z][A-Za-z0-9_.-]{0,127}$/u.test(key))) fail("SERVER_OPERATOR_HTTP_REQUEST_INVALID");
|
|
const propertiesJSON = stableJSON(request.properties);
|
|
if (Buffer.byteLength(propertiesJSON) > 64 * 1024 || Object.values(request.properties).some((value) => value === null || (value && typeof value === "object" && !Array.isArray(value)) || (typeof value === "string" && Buffer.byteLength(value) > 4096) || (Array.isArray(value) && (value.length > 64 || value.some((item) => item === null || (item && typeof item === "object")) || value.length > 0 && value.some((item) => typeof item !== typeof value[0]))))) fail("SERVER_OPERATOR_HTTP_PROPERTIES_INVALID");
|
|
const propertiesSha256 = digest(Buffer.from(propertiesJSON));
|
|
if (propertiesSha256 !== request.propertiesSha256) fail("SERVER_OPERATOR_HTTP_PROPERTIES_HASH_MISMATCH", 422);
|
|
const { requestSha256, ...unsigned } = request;
|
|
const canonical = stableJSON({ ...unsigned, context: unsigned.context ?? "BACKGROUND" });
|
|
if (digest(Buffer.from(canonical)) !== requestSha256) fail("SERVER_OPERATOR_HTTP_REQUEST_HASH_MISMATCH", 422);
|
|
if (request.context !== undefined && request.context !== "BACKGROUND") fail("SERVER_OPERATOR_CONTEXT_BLOCKED", 422);
|
|
if (/^(?:wm|window|screen|file|outliner|view3d)\./iu.test(String(request.operator ?? ""))) fail("SERVER_OPERATOR_CONTEXT_BLOCKED", 422);
|
|
}
|
|
function readEnvelope(buffer) {
|
|
if (buffer.length < 16 || !buffer.subarray(0, 4).equals(MAGIC) || buffer.readUInt32LE(4) !== 1) fail("SERVER_OPERATOR_HTTP_INVALID_ENVELOPE");
|
|
const requestBytes = buffer.readUInt32LE(8);
|
|
const sourceBytes = buffer.readUInt32LE(12);
|
|
if (buffer.length > MAX_ENVELOPE || requestBytes < 1 || sourceBytes < 1 || 16 + requestBytes + sourceBytes !== buffer.length) fail("SERVER_OPERATOR_HTTP_INVALID_LENGTH");
|
|
const request = JSON.parse(buffer.subarray(16, 16 + requestBytes).toString("utf8"));
|
|
if (!request || typeof request !== "object" || Array.isArray(request)) fail("SERVER_OPERATOR_HTTP_REQUEST_INVALID");
|
|
const source = buffer.subarray(16 + requestBytes);
|
|
if (source.length !== request.sourceBlendByteLength || digest(source) !== request.sourceBlendSha256) fail("SERVER_OPERATOR_HTTP_SOURCE_HASH_MISMATCH", 422);
|
|
validateBindings(request);
|
|
return { request, source };
|
|
}
|
|
function writeJson(response, status, value) { response.statusCode = status; response.setHeader("Content-Type", "application/json"); response.setHeader("Cache-Control", "no-store"); response.end(JSON.stringify(value)); }
|
|
function collect(request, maxBytes) {
|
|
return new Promise((resolve, reject) => { const chunks = []; let total = 0; request.on("data", (chunk) => { total += chunk.length; if (total > maxBytes) { reject(Object.assign(new Error("SERVER_OPERATOR_HTTP_BODY_TOO_LARGE"), { statusCode: 413 })); request.destroy(); return; } chunks.push(chunk); }); request.on("end", () => resolve(Buffer.concat(chunks))); request.on("error", reject); });
|
|
}
|
|
function pythonPayload(request, sourcePath, workspaceRoot, blenderBinary) { return JSON.stringify({ action: "submit", request, sourceBlendPath: sourcePath, workspaceRoot, blenderBinary }); }
|
|
|
|
async function describeServerOperator(operator, options, clientRequest) {
|
|
const adapterHash = digest(Buffer.concat([fs.readFileSync(RUNNER),
|
|
fs.readFileSync(path.join(ROOT, "tools/web/server_operator_properties.py")),
|
|
fs.readFileSync(path.join(ROOT, "tools/web/server_operator_remesh.py"))]));
|
|
const binary = options.blenderBinary ?? "blender";
|
|
const cachePrefix = `${binary}:${executableFingerprint(binary)}:${adapterHash}:${operator}`;
|
|
const cached = [...operatorSchemaCache.entries()].find(([key]) => key.startsWith(`${cachePrefix}:`));
|
|
if (cached) return cached[1];
|
|
const workspace = fs.mkdtempSync(path.join(options.workspaceRoot ?? os.tmpdir(), "server-operator-describe-"));
|
|
let abort = null;
|
|
try {
|
|
const handle = startBoundedServerJobProcess(options.pythonBinary ?? "python3", [RUNNER], { cwd: workspace, timeoutMs: options.describeTimeoutMs ?? 120000, stdio: ["pipe", "pipe", "pipe"], maxOutputBytes: 2 * 1024 * 1024 });
|
|
const chunks = [];
|
|
let processError = null;
|
|
handle.child.on("error", (error) => { processError = error; });
|
|
handle.child.stdin.on("error", (error) => { processError = error; });
|
|
abort = () => { void cancelServerJobProcess(handle, async () => {}, { graceMs: 250 }).catch(() => undefined); };
|
|
clientRequest?.once("aborted", abort);
|
|
handle.child.stdout.on("data", (chunk) => chunks.push(Buffer.from(chunk)));
|
|
handle.child.stdin.end(JSON.stringify({ action: "describe", operator, blenderBinary: options.blenderBinary ?? "blender" }));
|
|
await handle.completion;
|
|
if (handle.receipt.cancelRequested) fail("SERVER_OPERATOR_SCHEMA_TIMEOUT", 504);
|
|
if (processError) fail("SERVER_OPERATOR_SCHEMA_PROCESS_FAILED", 502);
|
|
const output = Buffer.concat(chunks).toString("utf8").trim();
|
|
if (!output) fail("SERVER_OPERATOR_SCHEMA_EMPTY", 502);
|
|
const result = JSON.parse(output);
|
|
if (result.status === "FAILED") fail(result.errorMessage ?? "SERVER_OPERATOR_SCHEMA_FAILED", 422);
|
|
if (handle.child.exitCode !== 0) fail("SERVER_OPERATOR_SCHEMA_PROCESS_FAILED", 502);
|
|
const buildKey = result.blenderBuild?.buildSha256 ?? "unknown";
|
|
operatorSchemaCache.set(`${cachePrefix}:${buildKey}`, result);
|
|
return result;
|
|
}
|
|
finally {
|
|
if (abort) clientRequest?.off("aborted", abort);
|
|
fs.rmSync(workspace, { recursive: true, force: true });
|
|
}
|
|
}
|
|
|
|
function jobStatus(entry, status, errorMessage) {
|
|
const { jobId, sourceBlendSha256, sourceRevision, blenderBuild, propertiesSha256, requestSha256 } = entry.request;
|
|
const errorCode = typeof errorMessage === "string" ? errorMessage.match(/^([A-Z][A-Z0-9_.-]*)(?::|$)/u)?.[1] : undefined;
|
|
return { schemaVersion: 1, route: "SERVER_NATIVE", jobId, status, sourceBlendSha256, sourceRevision, blenderBuild, propertiesSha256, requestSha256, resultSha256: digest(Buffer.from(status)), ...(errorMessage ? { errorMessage, ...(errorCode ? { errorCode } : {}) } : {}) };
|
|
}
|
|
|
|
function cleanupJob(entry) {
|
|
if (entry.result && !entry.retentionTimer) {
|
|
// Cover the client timeout + 2s deadline and its 2s cancellation request.
|
|
entry.retentionTimer = setTimeout(() => {
|
|
if (jobs.get(entry.request.jobId) === entry) jobs.delete(entry.request.jobId);
|
|
}, entry.request.timeoutMs + 4_000);
|
|
entry.retentionTimer.unref();
|
|
}
|
|
if (entry.workspace) {
|
|
fs.rmSync(entry.workspace, { recursive: true, force: true });
|
|
entry.workspace = null;
|
|
}
|
|
}
|
|
|
|
async function finishJob(entry) {
|
|
try {
|
|
await entry.handle.completion;
|
|
// Cancellation publishes its terminal state before waiting for process teardown.
|
|
if (entry.result) return;
|
|
if (entry.handle.receipt.cancelRequested) throw new Error("SERVER_OPERATOR_TIMEOUT");
|
|
if (entry.processError || entry.handle.child.exitCode !== 0) throw new Error("SERVER_OPERATOR_PROCESS_FAILED");
|
|
const result = JSON.parse(fs.readFileSync(path.join(entry.workspace, "result.json"), "utf8"));
|
|
for (const key of ["schemaVersion", "route", "jobId", "sourceBlendSha256", "sourceRevision", "propertiesSha256", "requestSha256"]) {
|
|
if (result[key] !== entry.request[key]) throw new Error("SERVER_OPERATOR_RESULT_BINDING_MISMATCH");
|
|
}
|
|
if (typeof result.resultSha256 !== "string" || !/^[a-f0-9]{64}$/iu.test(result.resultSha256)) throw new Error("SERVER_OPERATOR_RESULT_HASH_INVALID");
|
|
if (stableJSON(result.blenderBuild) !== stableJSON(entry.request.blenderBuild)) throw new Error("SERVER_OPERATOR_RESULT_BUILD_MISMATCH");
|
|
if (result.status !== "SUCCEEDED" && result.status !== "FAILED") throw new Error("SERVER_OPERATOR_RESULT_STATUS_INVALID");
|
|
if (result.status === "SUCCEEDED") {
|
|
const blendPath = path.join(entry.workspace, "result.blend");
|
|
const size = fs.statSync(blendPath).size;
|
|
if (size < 1 || size > 512 * 1024 * 1024 || size !== result.resultBlendByteLength) throw new Error("SERVER_OPERATOR_RESULT_LENGTH_MISMATCH");
|
|
const bytes = fs.readFileSync(blendPath);
|
|
if (digest(bytes) !== result.resultBlendSha256) throw new Error("SERVER_OPERATOR_RESULT_HASH_MISMATCH");
|
|
result.resultBlendBase64 = bytes.toString("base64");
|
|
}
|
|
entry.result = result;
|
|
} catch (error) {
|
|
entry.result ??= jobStatus(entry, "FAILED", String(error.message ?? error));
|
|
} finally {
|
|
cleanupJob(entry);
|
|
}
|
|
}
|
|
|
|
export function serverOperatorJobOptions() {
|
|
return {
|
|
blenderBinary: process.env.BLENDER_BIN ?? "blender",
|
|
pythonBinary: process.env.PYTHON_BIN ?? "python3",
|
|
workspaceRoot: process.env.SERVER_OPERATOR_WORKSPACE_ROOT,
|
|
};
|
|
}
|
|
|
|
async function handle(request, response, options = serverOperatorJobOptions()) {
|
|
if (await handleNativeWorkflowRequest(request, response, options)) return true;
|
|
const pathname = (request.url ?? "/").split("?", 1)[0];
|
|
if (pathname !== (options.path ?? "/api/server-operator-job")) return false;
|
|
const headerAction = request.headers["x-server-operator-action"];
|
|
const action = headerAction ?? ((new URL(request.url, "http://vite.local")).searchParams.has("jobId") ? "read" : "submit");
|
|
try {
|
|
if (!["describe", "submit", "read", "cancel"].includes(action)) fail("SERVER_OPERATOR_HTTP_ACTION_INVALID");
|
|
if (request.method !== "POST" && !(request.method === "GET" && action === "read")) {
|
|
response.setHeader("Allow", action === "read" ? "GET, POST" : "POST");
|
|
fail("SERVER_OPERATOR_HTTP_METHOD_INVALID", 405);
|
|
}
|
|
if (action === "describe") {
|
|
const body = JSON.parse((await collect(request, 64 * 1024)).toString("utf8") || "{}");
|
|
if (!body || typeof body !== "object" || Array.isArray(body)) fail("SERVER_OPERATOR_HTTP_BODY_INVALID");
|
|
const operator = String(body.operator ?? "").trim();
|
|
if (!/^[A-Za-z_][A-Za-z0-9_.-]*$/.test(operator) || Buffer.byteLength(operator) > 128) fail("SERVER_OPERATOR_SCHEMA_OPERATOR_INVALID");
|
|
const result = await describeServerOperator(operator, options, request);
|
|
writeJson(response, 200, { status: { ready: true }, result });
|
|
return true;
|
|
}
|
|
if (action === "submit") {
|
|
const { request: jobRequest, source } = readEnvelope(await collect(request, MAX_ENVELOPE));
|
|
if (jobs.has(jobRequest.jobId)) fail("SERVER_OPERATOR_HTTP_DUPLICATE_JOB", 409);
|
|
if ([...jobs.values()].filter((entry) => !entry.result).length >= 2) fail("SERVER_OPERATOR_HTTP_CAPACITY", 429);
|
|
validateServerJobBudget({ timeoutMs: jobRequest.timeoutMs });
|
|
const workspace = fs.mkdtempSync(path.join(options.workspaceRoot ?? os.tmpdir(), "server-operator-http-"));
|
|
const sourcePath = path.join(workspace, "source.blend");
|
|
fs.writeFileSync(sourcePath, source, { mode: 0o600 });
|
|
const handle = startBoundedServerJobProcess(options.pythonBinary ?? "python3", [RUNNER], { cwd: workspace, timeoutMs: jobRequest.timeoutMs, stdio: ["pipe", "ignore", "pipe"], maxOutputBytes: 8 * 1024 * 1024 });
|
|
const entry = { request: jobRequest, workspace, handle, result: null };
|
|
jobs.set(jobRequest.jobId, entry);
|
|
handle.child.on("error", (error) => { entry.processError = error; });
|
|
handle.child.stdin.on("error", (error) => { entry.processError = error; });
|
|
const abort = () => { if (!entry.result) { entry.result = jobStatus(entry, "CANCELLED"); void cancelServerJobProcess(entry.handle, async () => cleanupJob(entry), { graceMs: 500 }).catch(() => undefined); } };
|
|
request.once("aborted", abort);
|
|
handle.child.stdin.write(pythonPayload(jobRequest, sourcePath, workspace, options.blenderBinary ?? "blender"));
|
|
handle.child.stdin.end();
|
|
entry.completion = finishJob(entry).catch((error) => { entry.result = jobStatus(entry, "FAILED", String(error.message ?? error)); }).finally(() => request.off("aborted", abort));
|
|
writeJson(response, 202, { status: { ready: true }, result: jobStatus(entry, "RUNNING") });
|
|
return true;
|
|
}
|
|
const body = request.method === "GET" ? {} : JSON.parse((await collect(request, 64 * 1024)).toString("utf8") || "{}");
|
|
if (!body || typeof body !== "object" || Array.isArray(body)) fail("SERVER_OPERATOR_HTTP_BODY_INVALID");
|
|
const url = new URL(request.url, "http://vite.local");
|
|
const jobId = String(body.jobId ?? url.searchParams.get("jobId") ?? "");
|
|
const requestSha256 = String(body.requestSha256 ?? url.searchParams.get("requestSha256") ?? "");
|
|
const entry = jobs.get(jobId);
|
|
if (!entry || entry.request.requestSha256 !== requestSha256) fail("SERVER_OPERATOR_HTTP_JOB_NOT_FOUND", 404);
|
|
if (action === "cancel") {
|
|
if (!entry.result) {
|
|
entry.result = jobStatus(entry, "CANCELLED");
|
|
try { await cancelServerJobProcess(entry.handle, async () => cleanupJob(entry), { graceMs: 500 }); }
|
|
catch (error) { cleanupJob(entry); entry.result.errorMessage = `SERVER_OPERATOR_CANCEL_CLEANUP_FAILED: ${String(error.message ?? error)}`; }
|
|
}
|
|
writeJson(response, 200, { status: { ready: true }, result: entry.result });
|
|
return true;
|
|
}
|
|
const result = entry.result ?? jobStatus(entry, "RUNNING");
|
|
writeJson(response, 200, { status: { ready: true }, result });
|
|
return true;
|
|
} catch (error) { if (error.statusCode === 429) response.setHeader("Retry-After", "1"); writeJson(response, error.statusCode ?? (error instanceof SyntaxError ? 400 : 500), { status: { ready: false }, result: { status: "FAILED", errorMessage: String(error.message ?? error), ...(typeof error.errorCode === "string" ? { errorCode: error.errorCode } : {}) } }); return true; }
|
|
}
|
|
|
|
export function serverOperatorJobProvider(options = serverOperatorJobOptions()) {
|
|
return { name: "server-operator-http-provider", configureServer(server) { server.middlewares.use((request, response, next) => { void handle(request, response, options).then((handled) => { if (!handled) next(); }).catch(() => next()); }); }, configurePreviewServer(server) { server.middlewares.use((request, response, next) => { void handle(request, response, options).then((handled) => { if (!handled) next(); }).catch(() => next()); }); } };
|
|
}
|
|
|
|
export { handle as handleServerOperatorJobRequest };
|
|
|
|
if (process.argv[1] && path.resolve(process.argv[1]) === fileURLToPath(import.meta.url)) {
|
|
const port = Number.parseInt(process.env.SERVER_OPERATOR_PORT ?? "8787", 10);
|
|
const host = process.env.SERVER_OPERATOR_HOST ?? "127.0.0.1";
|
|
if (!Number.isSafeInteger(port) || port < 1 || port > 65535) throw new Error("SERVER_OPERATOR_PORT_INVALID");
|
|
const options = serverOperatorJobOptions();
|
|
http.createServer((request, response) => {
|
|
void handle(request, response, options).then((handled) => {
|
|
if (!handled) writeJson(response, 404, { status: { ready: false }, result: { status: "FAILED", errorMessage: "NOT_FOUND", errorCode: "SERVER_OPERATOR_HTTP_NOT_FOUND" } });
|
|
}).catch((error) => writeJson(response, 500, { status: { ready: false }, result: { status: "FAILED", errorMessage: String(error?.message ?? error), ...(typeof error?.errorCode === "string" ? { errorCode: error.errorCode } : {}) } }));
|
|
}).listen(port, host, () => process.stdout.write(`server-operator-provider-ready http://${host}:${port}/api/server-operator-job\n`));
|
|
}
|