diff --git a/.gitmodules b/.gitmodules index 5098433..7498ad8 100644 --- a/.gitmodules +++ b/.gitmodules @@ -1,3 +1,9 @@ [submodule "vendor/CultLib"] path = vendor/CultLib url = https://github.com/GameCult/CultLib.git +[submodule "vendor/kimodo"] + path = vendor/kimodo + url = https://github.com/nv-tlabs/kimodo.git +[submodule "vendor/SOMA-X"] + path = vendor/SOMA-X + url = https://github.com/NVlabs/SOMA-X.git diff --git a/package-lock.json b/package-lock.json index cbedae2..7eecc03 100644 --- a/package-lock.json +++ b/package-lock.json @@ -8,7 +8,9 @@ "name": "@gamecult/vili", "version": "0.1.0", "dependencies": { - "cultcache-ts": "file:vendor/CultLib/packages/cultcache-ts" + "@gamecult/cultcache-ts": "file:vendor/CultLib/packages/cultcache-ts", + "@msgpack/msgpack": "3.1.3", + "cultnet-ts": "file:vendor/CultLib/packages/cultnet-ts" } }, "../CultLib/packages/cultcache-ts": { @@ -29,6 +31,19 @@ "node": ">=20" } }, + "node_modules/@gamecult/cultcache-ts": { + "version": "0.14.2", + "resolved": "file:vendor/CultLib/packages/cultcache-ts", + "license": "MIT", + "dependencies": { + "@msgpack/msgpack": "^3.1.3", + "koffi": "3.3.0", + "zod": "^3.25.76" + }, + "engines": { + "node": ">=20" + } + }, "node_modules/@koromix/koffi-android-arm64": { "version": "3.3.0", "resolved": "https://registry.npmjs.org/@koromix/koffi-android-arm64/-/koffi-android-arm64-3.3.0.tgz", @@ -358,19 +373,62 @@ "node": ">= 18" } }, - "node_modules/@types/node": { - "version": "24.19.2", - "resolved": "https://registry.npmjs.org/@types/node/-/node-24.19.2.tgz", - "integrity": "sha512-93d49wCWJkCyN87MxLpNN6KXPCwFw7bVUg79mGUV1OS40JGWwVCr8n61DXCSqz7CV6/VfPkb7xec7uP2M+QHWQ==", - "dev": true, + "node_modules/ajv": { + "version": "8.20.0", + "resolved": "https://registry.npmjs.org/ajv/-/ajv-8.20.0.tgz", + "integrity": "sha512-Thbli+OlOj+iMPYFBVBfJ3OmCAnaSyNn4M1vz9T6Gka5Jt9ba/HIR56joy65tY6kx/FCF5VXNB819Y7/GUrBGA==", "license": "MIT", "dependencies": { - "undici-types": ">=7.24.0 <7.24.7" + "fast-deep-equal": "^3.1.3", + "fast-uri": "^3.0.1", + "json-schema-traverse": "^1.0.0", + "require-from-string": "^2.0.2" + }, + "funding": { + "type": "github", + "url": "https://github.com/sponsors/epoberezkin" } }, - "node_modules/cultcache-ts": { - "resolved": "vendor/CultLib/packages/cultcache-ts", - "link": true + "node_modules/cultnet-ts": { + "version": "0.3.0", + "resolved": "file:vendor/CultLib/packages/cultnet-ts", + "license": "MIT", + "dependencies": { + "@gamecult/cultcache-ts": "^0.14.1", + "@msgpack/msgpack": "^3.1.3", + "ajv": "^8.17.1" + }, + "engines": { + "node": ">=20" + } + }, + "node_modules/fast-deep-equal": { + "version": "3.1.3", + "resolved": "https://registry.npmjs.org/fast-deep-equal/-/fast-deep-equal-3.1.3.tgz", + "integrity": "sha512-f3qQ9oQy9j2AhBe/H9VC91wLmKBCCU/gDOnKNAYG5hswO7BLKj09Hc5HYNz9cGI++xlpDCIgDaitVs03ATR84Q==", + "license": "MIT" + }, + "node_modules/fast-uri": { + "version": "3.1.9", + "resolved": "https://registry.npmjs.org/fast-uri/-/fast-uri-3.1.9.tgz", + "integrity": "sha512-hS/0Wt3Ya043oKXs0sIxrMpKQITXhMLIrVAKKXi4gozGC9v7YRWVnXIZzjNPTO02M9BdxLCWqK1yRXgNziVyJQ==", + "funding": [ + { + "type": "github", + "url": "https://github.com/sponsors/fastify" + }, + { + "type": "opencollective", + "url": "https://opencollective.com/fastify" + } + ], + "license": "BSD-3-Clause" + }, + "node_modules/json-schema-traverse": { + "version": "1.0.0", + "resolved": "https://registry.npmjs.org/json-schema-traverse/-/json-schema-traverse-1.0.0.tgz", + "integrity": "sha512-NM8/P9n3XjXhIZn1lLhkFaACTOURQXjWhV4BA/RnOv8xvgqtqpAX9IO4mRQxSx1Rlo4tqzeqb0sOlruaOy3dug==", + "license": "MIT" }, "node_modules/koffi": { "version": "3.3.0", @@ -404,27 +462,15 @@ "@koromix/koffi-win32-x64": "3.3.0" } }, - "node_modules/typescript": { - "version": "5.9.3", - "resolved": "https://registry.npmjs.org/typescript/-/typescript-5.9.3.tgz", - "integrity": "sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw==", - "dev": true, - "license": "Apache-2.0", - "bin": { - "tsc": "bin/tsc", - "tsserver": "bin/tsserver" - }, + "node_modules/require-from-string": { + "version": "2.0.2", + "resolved": "https://registry.npmjs.org/require-from-string/-/require-from-string-2.0.2.tgz", + "integrity": "sha512-Xf0nWe6RseziFMu+Ap9biiUbmplq6S9/p+7w7YXP/JBHhrUDDUhwa+vANyubuqfZWTveU//DYVGsDG7RKL/vEw==", + "license": "MIT", "engines": { - "node": ">=14.17" + "node": ">=0.10.0" } }, - "node_modules/undici-types": { - "version": "7.24.6", - "resolved": "https://registry.npmjs.org/undici-types/-/undici-types-7.24.6.tgz", - "integrity": "sha512-WRNW+sJgj5OBN4/0JpHFqtqzhpbnV0GuB+OozA9gCL7a993SmU+1JBZCzLNxYsbMfIeDL+lTsphD5jN5N+n0zg==", - "dev": true, - "license": "MIT" - }, "node_modules/zod": { "version": "3.25.76", "resolved": "https://registry.npmjs.org/zod/-/zod-3.25.76.tgz", @@ -437,6 +483,7 @@ "vendor/CultLib/packages/cultcache-ts": { "name": "@gamecult/cultcache-ts", "version": "0.14.2", + "extraneous": true, "license": "MIT", "dependencies": { "@msgpack/msgpack": "^3.1.3", diff --git a/package.json b/package.json index 0c718fb..da6fe37 100644 --- a/package.json +++ b/package.json @@ -6,10 +6,14 @@ "description": "Vili is the GameCult Persona animation daemon: speech intent in, embodied motion affordance out.", "scripts": { "start": "node scripts/vili-daemon.mjs", - "health": "node scripts/vili-daemon.mjs --health", "test": "node --test --test-timeout=60000 scripts/test-vili-daemon.mjs" }, "dependencies": { - "cultcache-ts": "file:vendor/CultLib/packages/cultcache-ts" + "@gamecult/cultcache-ts": "file:vendor/CultLib/packages/cultcache-ts", + "@msgpack/msgpack": "3.1.3", + "cultnet-ts": "file:vendor/CultLib/packages/cultnet-ts" + }, + "overrides": { + "@gamecult/cultcache-ts": "file:vendor/CultLib/packages/cultcache-ts" } } diff --git a/scripts/fixtures/fake-idunn-entry.mjs b/scripts/fixtures/fake-idunn-entry.mjs new file mode 100644 index 0000000..7208cbc --- /dev/null +++ b/scripts/fixtures/fake-idunn-entry.mjs @@ -0,0 +1,26 @@ +// Test seam only: the shipped Idunn runtime (scripts/idunn-entry.mjs idunnRuntime) over a signer whose write lease is +// a file, so the lease gate is tested without Idunn's authority files. Present = leased. The daemon's own flags +// arrive on argv as they do for the hand-run daemon. +import { existsSync, appendFileSync } from "node:fs"; +import { idunnRuntime } from "../idunn-entry.mjs"; +import { serve } from "../vili-daemon.mjs"; + +const leasePath = process.env.FAKE_IDUNN_LEASE; +const healthLog = process.env.FAKE_IDUNN_HEALTH_LOG; +const answer = process.env.FAKE_IDUNN_ANSWER_THROWS !== "1"; + +const signer = { + assertWriteLease() { + if (!existsSync(leasePath)) throw new Error("no lease"); + return "lease"; + }, + answerRouteObservation(request) { + if (!answer) throw new Error("signer refused"); + return { schemaId: "fake.presence", recordKey: request.messageId, payload: new Uint8Array([1, 2, 3]) }; + }, + reportHealth(state) { + appendFileSync(healthLog, `${state}\n`); + }, +}; + +await serve(idunnRuntime(signer, { pollMs: 200 })); diff --git a/scripts/fixtures/stub-worker.mjs b/scripts/fixtures/stub-worker.mjs index aaac654..99de4c1 100644 --- a/scripts/fixtures/stub-worker.mjs +++ b/scripts/fixtures/stub-worker.mjs @@ -29,8 +29,8 @@ if (model === "crash-before-ready") process.exit(3); if (model === "stubborn") { process.on("SIGTERM", () => {}); - const grandchild = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { stdio: "ignore" }); - appendFileSync(spawnLog, `grandchild ${grandchild.pid}\n`); + // the grandchild reports itself once its handler is armed, so a test never reads a pid that would die of the default SIGTERM + spawn(process.execPath, ["-e", `process.on('SIGTERM', () => {}); require('fs').appendFileSync(${JSON.stringify(spawnLog)}, 'grandchild ' + process.pid + '\\n'); setInterval(() => {}, 1000)`], { stdio: "ignore" }); } if (model === "graceful") { @@ -66,8 +66,7 @@ if (model === "hang-before-ready") { if (model === "slow-generate") return; if (model === "encoder") { // an encoder child of the real worker: a grandchild in the worker's group, then a request that never answers - const encoder = spawn(process.execPath, ["-e", "process.on('SIGTERM', () => {}); setInterval(() => {}, 1000)"], { stdio: "ignore" }); - appendFileSync(spawnLog, `grandchild ${encoder.pid}\n`); + spawn(process.execPath, ["-e", `process.on('SIGTERM', () => {}); require('fs').appendFileSync(${JSON.stringify(spawnLog)}, 'grandchild ' + process.pid + '\\n'); setInterval(() => {}, 1000)`], { stdio: "ignore" }); return; } send(200, { diff --git a/scripts/idunn-entry.mjs b/scripts/idunn-entry.mjs new file mode 100644 index 0000000..1cbaf8f --- /dev/null +++ b/scripts/idunn-entry.mjs @@ -0,0 +1,123 @@ +#!/usr/bin/env node +// The Idunn entry: the process Idunn's unit runs for target "vili". Idunn owns Vili's lifecycle and hands it one +// writer authority, the process write lease. This file turns Idunn's environment into the daemon's own +// configuration and into the runtime port serve() takes; it adds no routes and decides nothing the daemon decides. +// +// node vili/scripts/idunn-entry.mjs --state-root --kimodo-commit <40 hex> +// +// Nothing from the bundle, the activation or the identity reaches the worker: the daemon's childEnv forwards only +// PATH, HOME and LANG besides its own VILI_* settings. +import { createHash } from "node:crypto"; +import { readFileSync } from "node:fs"; +import path from "node:path"; +import { setTimeout as delay } from "node:timers/promises"; +import { fileURLToPath, pathToFileURL } from "node:url"; +import { decode, encode } from "@msgpack/msgpack"; +import { encodeCultNetMessageForWire, parseCultNetMessage } from "cultnet-ts"; +import { createIdunnRuntimeSigner, loadIdunnRuntimeAuthorityFromEnvironment } from "cultnet-ts/idunn-runtime"; + +const HEALTH_CONTRACT = "gamecult.runtime_presence_health.v2"; +const LEASE_POLL_MS = 5000; +const here = path.dirname(fileURLToPath(import.meta.url)); + +// The runtime port serve() takes, over an Idunn runtime signer. Exported so the lease gate is tested against the +// code that ships, with a signer whose lease is a file (scripts/fixtures/fake-idunn-entry.mjs). +export function idunnRuntime(signer, { pollMs = LEASE_POLL_MS } = {}) { + let held = false; + let lost = false; + + return { + async acquire() { + for (;;) { + try { + signer.assertWriteLease(); + held = true; + return; + } catch { + // Idunn has not leased this incarnation yet + } + await delay(pollMs); + } + }, + held: () => held, + refusal: () => (lost ? "write-lease-lost" : "write-lease-pending"), + watch(onLost) { + if (!held) return; + try { + signer.assertWriteLease(); + } catch { + held = false; + lost = true; + onLost(); + } + }, + recordWorker() {}, + forgetWorker() {}, + answerSnapshot(bytes) { + let request; + try { + request = parseCultNetMessage(decode(bytes), "cultnet.schema.v0"); + if (request.schemaVersion !== "cultnet.snapshot_request.v0") throw new Error("not a snapshot request"); + } catch { + throw Object.assign(new Error("invalid snapshot request"), { code: "invalid-snapshot-request" }); + } + const document = signer.answerRouteObservation(request); + const response = { + schemaVersion: "cultnet.snapshot_response_raw.v0", + messageId: request.messageId, + documents: [document], + }; + return Buffer.from(encode(encodeCultNetMessageForWire(response, "cultnet.schema.v0"))); + }, + reportHealth: (state) => signer.reportHealth(state), + release() {}, + }; +} + +function fail(reason) { + console.error(`vili idunn entry: ${reason}`); + process.exit(2); +} + +// host:port, with an optional http:// scheme, as Idunn writes the candidate bind. +function splitBind(bind) { + const match = /^(?:http:\/\/)?([^:/\s]+):(\d{1,5})\/?$/u.exec(bind ?? ""); + if (!match || Number(match[2]) > 65535) fail("GAMECULT_IDUNN_CANDIDATE_BIND is not host:port"); + return { host: match[1], port: match[2] }; +} + +function parseEntryArgs(argv) { + if (argv.length !== 4 || argv[0] !== "--state-root" || argv[2] !== "--kimodo-commit") { + fail("arguments must be exactly --state-root --kimodo-commit <40 hex>"); + } + if (!path.isAbsolute(argv[1])) fail("--state-root must be absolute"); + if (!/^[0-9a-f]{40}$/u.test(argv[3])) fail("--kimodo-commit must be 40 lowercase hex characters"); +} + +async function main() { + parseEntryArgs(process.argv.slice(2)); + const { host, port } = splitBind(process.env.GAMECULT_IDUNN_CANDIDATE_BIND); + const locks = ["requirements-build.lock", "requirements.lock"].map((name) => + readFileSync(path.resolve(here, "..", "deployment", "idunn", name))); + process.env.VILI_DAEMON_HOST = host; + process.env.VILI_DAEMON_PORT = port; + process.env.VILI_PYTHON = path.resolve(here, "..", "..", "python", "bin", "python3.12"); + process.env.VILI_PYTHON_ENV = `locks-sha256:${createHash("sha256").update(Buffer.concat(locks)).digest("hex")}`; + + const authority = loadIdunnRuntimeAuthorityFromEnvironment("vili", HEALTH_CONTRACT); + const signer = createIdunnRuntimeSigner({ + authority, + capabilities: [{ capability: "vili.motion.api", schema: "http.v1", compatibility: "v1", capacity: 1 }], + }); + if (!signer.requiresWriteLease) fail("Idunn declares no write lease for this runtime"); + + const { serve } = await import("./vili-daemon.mjs"); + await serve(idunnRuntime(signer)); +} + +if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) { + main().catch((error) => { + console.error(`vili idunn entry failed: ${error?.name || "Error"}`); + process.exit(1); + }); +} diff --git a/scripts/kimodo-resident-worker.py b/scripts/kimodo-resident-worker.py index df9feb2..e855a32 100644 --- a/scripts/kimodo-resident-worker.py +++ b/scripts/kimodo-resident-worker.py @@ -3,6 +3,7 @@ import argparse import json import os import shutil +import sys import time import traceback from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer @@ -63,7 +64,7 @@ model, resolved_model = load_model( text_encoder=text_encoder, ) model_revision = text_encoding.read_snapshot_revision(HF_HOME, MODEL_NAMES[resolved_model]) -worker_identity = {"kimodoCommit": kimodo_commit, "pythonEnv": python_env, "modelRevision": model_revision, "threads": threads} +worker_identity = {"kimodoCommit": kimodo_commit, "pythonEnv": python_env, "modelRevision": model_revision, "threads": threads, "python": sys.version} model_load_seconds = time.time() - model_load_started_at model_info = get_model_info(resolved_model) display_model = model_info.display_name if model_info else resolved_model diff --git a/scripts/kimodo-text-encode.py b/scripts/kimodo-text-encode.py index 6ead1a8..f5f7f68 100644 --- a/scripts/kimodo-text-encode.py +++ b/scripts/kimodo-text-encode.py @@ -40,6 +40,20 @@ def load_encoder(hf_home, mntp_id, supervised_id, dtype): return LLM2Vec(model=model, tokenizer=tokenizer) +def encode_texts(model, texts): + """Each text wrapped in the chat header and encoded alone: (vector bytes, seconds) lists.""" + vectors, encode_seconds = [], [] + for text in texts: + began = time.perf_counter() + features = model.tokenize([chat_header(text)]) + with torch.no_grad(): + vector = model.forward(features).detach().cpu().to(torch.float32)[0] + assert vector.shape == (EMBEDDING_DIM,) + vectors.append(vector.numpy().tobytes()) + encode_seconds.append(time.perf_counter() - began) + return vectors, encode_seconds + + def main() -> int: apply_thread_cap(os.environ) texts = json.load(sys.stdin)["texts"] @@ -53,15 +67,7 @@ def main() -> int: p.requires_grad = False load_seconds = time.perf_counter() - started - vectors, encode_seconds = [], [] - for text in texts: - began = time.perf_counter() - features = model.tokenize([chat_header(text)]) - with torch.no_grad(): - vector = model.forward(features).detach().cpu().to(torch.float32)[0] - assert vector.shape == (EMBEDDING_DIM,) - vectors.append(vector.numpy().tobytes()) - encode_seconds.append(time.perf_counter() - began) + vectors, encode_seconds = encode_texts(model, texts) out.write_bytes(msgpack.packb({ "vectors": vectors, @@ -75,4 +81,5 @@ def main() -> int: return 0 -sys.exit(main()) +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/test-vili-daemon.mjs b/scripts/test-vili-daemon.mjs index ac41e59..7fca2e9 100644 --- a/scripts/test-vili-daemon.mjs +++ b/scripts/test-vili-daemon.mjs @@ -2,16 +2,18 @@ // Runs the real daemon against scripts/fixtures/stub-worker.mjs as its --python/--worker-entry. import assert from "node:assert/strict"; import { spawn } from "node:child_process"; -import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { existsSync, mkdtempSync, readFileSync, readdirSync, rmSync, statSync, writeFileSync } from "node:fs"; import http from "node:http"; import os from "node:os"; import path from "node:path"; import test from "node:test"; import { fileURLToPath } from "node:url"; +import { decode, encode } from "@msgpack/msgpack"; const here = path.dirname(fileURLToPath(import.meta.url)); const daemonPath = path.join(here, "vili-daemon.mjs"); const stubPath = path.join(here, "fixtures", "stub-worker.mjs"); +const fakeIdunnEntry = path.join(here, "fixtures", "fake-idunn-entry.mjs"); const THREADS = 5; async function until(condition, what, seconds = 15) { @@ -40,7 +42,7 @@ function isAlive(pid) { async function startDaemon({ omit = [], viaEnv = false, threads = String(THREADS), python = process.execPath, nested = false, host = "127.0.0.1", - relativeModels = false, extraArgs = [], root: givenRoot = null, + relativeModels = false, extraArgs = [], root: givenRoot = null, entry = daemonPath, extraEnv = {}, hold = entry === daemonPath, } = {}) { const root = givenRoot ?? mkdtempSync(path.join(os.tmpdir(), "vili-test-")); const stateRoot = nested ? path.join(root, "a", "b", "state") : root; @@ -57,9 +59,9 @@ async function startDaemon({ }; if (threads !== null) settings.threads = threads; for (const name of omit) delete settings[name]; - const env = { ...process.env, LEAK_CHECK: "leaked", LANG: "xx_TEST.UTF-8" }; + const env = { ...process.env, LEAK_CHECK: "leaked", LANG: "xx_TEST.UTF-8", ...extraEnv }; for (const name of Object.keys(env)) if (name.startsWith("VILI_")) delete env[name]; - const argv = [daemonPath, ...extraArgs]; + const argv = [entry, ...extraArgs]; for (const [name, value] of Object.entries(settings)) { if (viaEnv) env[`VILI_${{ host: "DAEMON_HOST", port: "DAEMON_PORT", "state-root": "STATE_ROOT", "worker-entry": "WORKER_ENTRY" }[name] ?? name.toUpperCase().replace("-", "_")}`] = value; else argv.push(`--${name}`, value); @@ -71,17 +73,21 @@ async function startDaemon({ proc.stderr.on("data", (chunk) => { errors += chunk; }); const listening = await until(() => /listening on http:\/\/([^:]+):(\d+)/.exec(output), "the daemon to listen"); const port = listening[2]; + // the hand-run daemon takes its lock after it listens; a test that wants a served state root waits for it + if (hold) await until(() => /holds the state root/.test(output), "the daemon to hold the state root"); const spawnLog = () => { const log = path.join(stateRoot, "embeddings", "stub-spawns.log"); return existsSync(log) ? readFileSync(log, "utf8").split("\n").filter(Boolean) : []; }; return { + root, stateRoot, modelsRoot, proc, boundHost: listening[1], port: Number(port), stderr: () => errors, + stdout: () => output, spawns: () => spawnLog().filter((line) => /^\d+$/.test(line)).map(Number), grandchildren: () => spawnLog().filter((line) => line.startsWith("grandchild ")).map((line) => Number(line.split(" ")[1])), async stop() { @@ -429,11 +435,14 @@ test("a live worker outside the pid file's group is left alone", async () => { try { assert.equal((await generate(first)).status, 200); const [worker] = first.spawns(); + first.proc.kill("SIGKILL"); + await until(() => first.proc.exitCode !== null || first.proc.signalCode !== null, "the first daemon to die"); writeFileSync(path.join(root, "worker.pid"), `${bystander.pid}\n`); const second = await startDaemon({ root }); assert.ok(isAlive(worker) && isAlive(bystander.pid), "only the group the pid file names, and only if it is ours, is killed"); second.proc.kill("SIGTERM"); await until(() => second.proc.exitCode !== null, "the second daemon to exit", 30); + process.kill(-worker, "SIGKILL"); } finally { bystander.kill("SIGKILL"); await first.stop(); @@ -655,22 +664,6 @@ test("the typed service state is refreshed while the daemon runs", async () => { } }); -test("--health asks a running daemon, and fails when there is none", async () => { - const daemon = await startDaemon(); - const check = (port) => new Promise((resolve) => { - const proc = spawn(process.execPath, [daemonPath, "--health", "--host", "127.0.0.1", "--port", String(port)], { stdio: "ignore" }); - proc.on("exit", resolve); - }); - try { - assert.equal(await check(daemon.port), 0); - const gone = daemon.port; - await daemon.stop(); - assert.equal(await check(gone), 1); - } finally { - await daemon.stop(); - } -}); - test("a thread count outside 1 to 16 stops the daemon without echoing it", async () => { for (const threads of ["0", "17", "99", "100", "8x", "x8", "-1", "1.5"]) { const proc = spawn(process.execPath, [daemonPath, "--threads", threads], { stdio: ["ignore", "pipe", "pipe"] }); @@ -683,3 +676,161 @@ test("a thread count outside 1 to 16 stops the daemon without echoing it", async assert.ok(!output.replace("from 1 to 16", "").includes(threads), `the refusal repeated ${threads}`); } }); + +const msgpackPost = (daemon, body, headers = {}) => + new Promise((resolve, reject) => { + const req = http.request({ host: "127.0.0.1", port: daemon.port, method: "POST", path: "/cultnet/snapshot", headers: { "content-type": "application/msgpack", ...headers } }, (res) => { + const chunks = []; + res.on("data", (chunk) => chunks.push(chunk)); + res.on("end", () => resolve({ status: res.statusCode, headers: res.headers, raw: Buffer.concat(chunks) })); + }); + req.on("error", reject); + req.end(body); + }); +const SECRET = "canary-9f3c-never-echoed"; +const snapshotRequest = (messageId = "m1") => Buffer.from(encode({ schemaVersion: "cultnet.snapshot_request.v0", messageId })); + +// A second daemon on a state root a live daemon holds must refuse, and must not touch the first one's worker. +test("a second daemon on a live state root exits 2 and kills nothing", async () => { + const first = await startDaemon(); + try { + assert.equal((await generate(first)).status, 200); + const [worker] = first.spawns(); + assert.ok(isAlive(worker)); + const second = await startDaemon({ root: first.root, hold: false }); + await until(() => second.proc.exitCode !== null, "the second daemon to refuse", 20); + assert.equal(second.proc.exitCode, 2); + assert.match(second.stderr(), /held by a running daemon/); + assert.ok(isAlive(worker), "the first daemon's worker must survive the refusal"); + assert.equal((await generate(first)).status, 200); + assert.equal(aliveWorkers(first).length, 1); + } finally { + await first.stop(); + } +}); + +const idunnDaemon = async ({ leased = false, extraEnv = {} } = {}) => { + const root = mkdtempSync(path.join(os.tmpdir(), "vili-test-")); + const lease = path.join(root, "lease"); + const healthLog = path.join(root, "health.log"); + if (leased) writeFileSync(lease, ""); + const daemon = await startDaemon({ + root, nested: true, entry: fakeIdunnEntry, extraEnv: { FAKE_IDUNN_LEASE: lease, FAKE_IDUNN_HEALTH_LOG: healthLog, ...extraEnv }, + }); + return Object.assign(daemon, { lease, health: () => (existsSync(healthLog) ? readFileSync(healthLog, "utf8").split("\n").filter(Boolean) : []) }); +}; + +test("before the lease nothing is written, spawned or accepted", async () => { + const daemon = await idunnDaemon(); + try { + const answer = await generate(daemon); + assert.equal(answer.status, 503); + assert.equal(answer.body.code, "write-lease-pending"); + assert.deepEqual(daemon.spawns(), []); + assert.equal(existsSync(daemon.stateRoot), false, "the state root must not exist before the lease"); + assert.deepEqual(daemon.health(), []); + } finally { + await daemon.stop(); + } +}); + +test("with the lease the daemon writes and serves, and no pid file exists", async () => { + const daemon = await idunnDaemon(); + try { + writeFileSync(daemon.lease, ""); + await until(() => existsSync(path.join(daemon.stateRoot, "vili.service.cc")), "the service store", 10); + const answer = await generate(daemon); + assert.equal(answer.status, 200); + assert.equal(existsSync(path.join(daemon.stateRoot, "worker.pid")), false); + assert.equal(existsSync(path.join(daemon.stateRoot, "daemon.lock")), false); + assert.deepEqual(daemon.health(), ["active"]); + } finally { + await daemon.stop(); + } +}); + +test("losing the lease kills the worker tree, refuses, and writes nothing more", async () => { + const daemon = await idunnDaemon({ leased: true }); + try { + await until(() => existsSync(path.join(daemon.stateRoot, "vili.service.cc")), "the service store", 10); + assert.equal((await generate(daemon, { prompt: "x", model: "stubborn" })).status, 200); + const [worker] = daemon.spawns(); + const [grandchild] = await until(() => { const found = daemon.grandchildren(); return found.length ? found : null; }, "the armed grandchild"); + assert.ok(isAlive(worker) && isAlive(grandchild)); + rmSync(daemon.lease); + await until(() => !isAlive(worker) && !isAlive(grandchild), "the worker tree to die", 20); + const answer = await generate(daemon); + assert.equal(answer.status, 503); + assert.equal(answer.body.code, "write-lease-lost"); + await until(() => daemon.health().join() === "active,degraded", "the health to say degraded"); + const store = path.join(daemon.stateRoot, "vili.service.cc"); + const before = statSync(store).mtimeMs; + const jobs = readdirSync(path.join(daemon.stateRoot, "jobs")).length; + await new Promise((resolve) => setTimeout(resolve, 6500)); + assert.equal(statSync(store).mtimeMs, before); + assert.equal(readdirSync(path.join(daemon.stateRoot, "jobs")).length, jobs); + } finally { + await daemon.stop(); + } +}); + +test("a job that finishes after the lease is lost is not recorded", async () => { + const daemon = await idunnDaemon({ leased: true }); + try { + await until(() => existsSync(path.join(daemon.stateRoot, "vili.service.cc")), "the service store", 10); + const slow = generate(daemon, { prompt: "x", model: "slow-generate" }); + await until(() => daemon.spawns().length, "the worker"); + await until(() => request(daemon.port, "GET", "/health").then((r) => r.body?.state?.backend?.state === "ready"), "the worker to be ready"); + rmSync(daemon.lease); + const answer = await slow; + assert.equal(answer.status, 503); + assert.equal(answer.body.code, "write-lease-lost"); + assert.deepEqual(readdirSync(path.join(daemon.stateRoot, "jobs")), []); + } finally { + await daemon.stop(); + } +}); + +test("the route challenge is answered ahead of the cross-site refusal, and an Origin is still refused", async () => { + const daemon = await idunnDaemon(); + try { + // the challenge is how Idunn learns the incarnation is warming, so it is answered before the lease + const answered = await msgpackPost(daemon, snapshotRequest("m-42")); + assert.equal(answered.status, 200); + assert.equal(answered.headers["content-type"], "application/msgpack"); + const message = decode(answered.raw); + assert.equal(message.schemaVersion, "cultnet.snapshot_response_raw.v0"); + assert.equal(message.messageId, "m-42"); + assert.equal(message.documents.length, 1); + assert.equal((await msgpackPost(daemon, snapshotRequest(), { origin: "https://evil.example" })).status, 403); + assert.equal((await msgpackPost(daemon, Buffer.from("not msgpack at all"))).status, 400); + assert.equal((await msgpackPost(daemon, Buffer.from(encode({ schemaVersion: "cultnet.snapshot_response_raw.v0", messageId: "x", documents: [] })))).status, 400); + assert.equal((await msgpackPost(daemon, Buffer.alloc(70 * 1024))).status, 400); + const text = await request(daemon.port, "POST", "/motion/generate", { body: "{}", headers: { "content-type": "text/plain" } }); + assert.equal(text.status, 403); + } finally { + await daemon.stop(); + } +}); + +test("a signer that cannot answer gives 503 and echoes nothing", async () => { + const daemon = await idunnDaemon({ extraEnv: { FAKE_IDUNN_ANSWER_THROWS: "1" } }); + try { + const answered = await msgpackPost(daemon, snapshotRequest(SECRET)); + assert.equal(answered.status, 503); + assert.ok(!answered.raw.toString("utf8").includes(SECRET)); + assert.ok(!daemon.stderr().includes(SECRET)); + } finally { + await daemon.stop(); + } +}); + +test("the hand-run daemon has no route challenge", async () => { + const daemon = await startDaemon(); + try { + assert.equal((await msgpackPost(daemon, snapshotRequest())).status, 404); + assert.equal((await msgpackPost(daemon, snapshotRequest(), { origin: "https://evil.example" })).status, 404); + } finally { + await daemon.stop(); + } +}); diff --git a/scripts/test_text_encoding.py b/scripts/test_text_encoding.py index 49d6fa4..bda389c 100644 --- a/scripts/test_text_encoding.py +++ b/scripts/test_text_encoding.py @@ -55,6 +55,9 @@ class Identity(unittest.TestCase): self.assertNotEqual(key, te.embedding_key("id", "text b")) self.assertNotEqual(te.embedding_key("id", "a" * 300 + "x"), te.embedding_key("id", "a" * 300 + "y")) + def test_key_is_case_sensitive(self): + self.assertNotEqual(te.embedding_key("id", "A person waves."), te.embedding_key("id", "a person waves.")) + def test_chat_header_is_the_llama3_user_turn_pooled_after_the_separator(self): header = "<|start_header_id|>user<|end_header_id|>" + chr(10) * 2 self.assertEqual(te.chat_header("hello "), header + "!@#$%^&*()hello<|eot_id|>") @@ -269,6 +272,51 @@ class ChildBoundary(unittest.TestCase): self.assertEqual(len(answer["vectors"]), 2) +class ChildHeader(unittest.TestCase): + def test_every_text_is_tokenized_with_the_chat_header(self): + import importlib.util + import torch + spec = importlib.util.spec_from_file_location("kimodo_text_encode", Path(te.__file__).with_name("kimodo-text-encode.py")) + child = importlib.util.module_from_spec(spec) + spec.loader.exec_module(child) + + class Recorder: + def __init__(self): + self.seen = [] + + def tokenize(self, batch): + self.seen.extend(batch) + return batch + + def forward(self, features): + return torch.zeros(1, te.EMBEDDING_DIM) + + model = Recorder() + vectors, seconds = child.encode_texts(model, ["a person waves", " padded "]) + self.assertEqual(model.seen, [te.chat_header("a person waves"), te.chat_header(" padded ")]) + for header in model.seen: + self.assertTrue(header.startswith("<|start_header_id|>user<|end_header_id|>" + chr(10) * 2)) + self.assertEqual([len(v) for v in vectors], [te.EMBEDDING_DIM * 4] * 2) + self.assertEqual(len(seconds), 2) + + +# sha256 of the two Kimodo sources the encoder mirrors, at vendor/kimodo 6bb58488 (embedding-identity-complete). +# A Kimodo bump that changes either file must re-read the mirrored loader before these move. +KIMODO_LOADER_SHA256 = { + "model/load_model.py": "008092cd0b7adef1faea9fa9075718a61bea8ca9feaf9acf1b1d6caacaa0380a", + "model/llm2vec/llm2vec.py": "dc6e8b7d38086b4f456e013ab22cd4cc2cd72eb7fd2c13069f037f42aec80857", +} + + +class KimodoPin(unittest.TestCase): + def test_the_mirrored_loader_sources_are_the_pinned_ones(self): + import hashlib + import kimodo + root = Path(kimodo.__file__).resolve().parent + for relative, expected in KIMODO_LOADER_SHA256.items(): + self.assertEqual(hashlib.sha256((root / relative).read_bytes()).hexdigest(), expected, relative) + + REFERENCE_MODELS = os.environ.get("VILI_REFERENCE_MODELS") ROCK = "A person shakes their fist up and down three times, then thrusts a closed fist forward like a rock." REFERENCE_CODE = """ diff --git a/scripts/vili-daemon.mjs b/scripts/vili-daemon.mjs index 4049c4c..408dcf7 100644 --- a/scripts/vili-daemon.mjs +++ b/scripts/vili-daemon.mjs @@ -5,7 +5,8 @@ import { closeSync, openSync, readdirSync, readFileSync, unlinkSync, writeFileSy import { mkdir, writeFile } from "node:fs/promises"; import path from "node:path"; import readline from "node:readline"; -import { fileURLToPath } from "node:url"; +import net from "node:net"; +import { fileURLToPath, pathToFileURL } from "node:url"; import { createRequire } from "node:module"; import { spawn } from "node:child_process"; @@ -16,7 +17,7 @@ const { SingleFileMessagePackBackingStore, defineDocumentRegistry, defineDocumentType, -} = require("cultcache-ts"); +} = require("@gamecult/cultcache-ts"); const projectRoot = path.resolve(__dirname, ".."); const providerId = "vili.animation"; const viliStoreFileName = "vili.service.cc"; @@ -85,7 +86,6 @@ function parseArgs(argv) { threads: env.VILI_THREADS || "8", kimodoCommit: env.VILI_KIMODO_COMMIT || "", pythonEnv: env.VILI_PYTHON_ENV || "", - health: false, }; for (let index = 0; index < argv.length; index += 1) { @@ -99,7 +99,6 @@ function parseArgs(argv) { else if (arg === "--threads") args.threads = argv[++index]; else if (arg === "--kimodo-commit") args.kimodoCommit = argv[++index]; else if (arg === "--python-env") args.pythonEnv = argv[++index]; - else if (arg === "--health") args.health = true; } if (!/^[0-9]{1,2}$/.test(String(args.threads)) || Number(args.threads) < 1 || Number(args.threads) > MAX_THREADS) { console.error(`threads must be an integer from 1 to ${MAX_THREADS}`); @@ -294,11 +293,9 @@ function waitForExit(proc, timeoutMs) { } // The worker leads its own process group (its pid is the group id), so the encoder child it spawns is in the group -// too. The daemon owns the whole group: it is killed when the worker exits for any reason, on a requested stop, and -// at the next start if a crashed daemon left it behind (reclaimStrayWorker). Under systemd, KillMode=control-group -// is the same ownership from outside the process. -const workerPidFile = () => path.join(args.stateRoot, "worker.pid"); - +// too. The daemon owns the whole group: it is killed when the worker exits for any reason and on a requested stop. +// A crashed hand-run daemon's group is reclaimed by the next hand-run daemon (handRunRuntime). Under Idunn the +// unit's cgroup is the same ownership from outside the process. function killGroup(pgid, signal) { try { process.kill(-pgid, signal); @@ -307,46 +304,102 @@ function killGroup(pgid, signal) { } } -function forgetWorkerPid(pid) { - try { - if (readFileSync(workerPidFile(), "utf8").trim() === String(pid)) unlinkSync(workerPidFile()); - } catch { - // no pid file, or another worker's - } -} +// The hand-run way of owning a state root (tests, development): a Unix socket lock, the worker's pid file and the +// /proc reclaim. Idunn's runtime (scripts/idunn-entry.mjs) replaces all three with its process write lease. +export function handRunRuntime(stateRoot) { + const lockPath = path.join(stateRoot, "daemon.lock"); + const pidFile = path.join(stateRoot, "worker.pid"); + let lock = null; -// A worker left by a crashed daemon: the pid file names its group, and the group is ours only if some member -// still carries this state root's embeddings directory in its environment (the pid may have been reused). -function reclaimStrayWorker() { - let pgid; - try { - pgid = Number(readFileSync(workerPidFile(), "utf8").trim()); - } catch { - return; - } - const marker = `VILI_EMBEDDINGS_DIR=${path.join(args.stateRoot, "embeddings")}`; - let ours = false; - try { - for (const entry of readdirSync("/proc")) { - if (!/^\d+$/.test(entry)) continue; - try { - const stat = readFileSync(`/proc/${entry}/stat`, "utf8"); - const fields = stat.slice(stat.lastIndexOf(")") + 2).split(" "); - if (Number(fields[2]) !== pgid) continue; - if (readFileSync(`/proc/${entry}/environ`, "utf8").split("\0").includes(marker)) ours = true; - } catch { - // the process exited while we looked - } + const listen = () => new Promise((resolve, reject) => { + const server = net.createServer((socket) => socket.destroy()); + server.once("error", reject); + server.listen(lockPath, () => { + server.removeListener("error", reject); + lock = server; + resolve(); + }); + }); + + const probe = () => new Promise((resolve) => { + const socket = net.connect(lockPath); + socket.once("connect", () => { socket.destroy(); resolve("live"); }); + socket.once("error", () => resolve("stale")); + }); + + // A worker left by a crashed daemon: the pid file names its group, and the group is ours only if some member + // still carries this state root's embeddings directory in its environment (the pid may have been reused). + function reclaimStrayWorker() { + let pgid; + try { + pgid = Number(readFileSync(pidFile, "utf8").trim()); + } catch { + return; + } + const marker = `VILI_EMBEDDINGS_DIR=${path.join(stateRoot, "embeddings")}`; + let ours = false; + try { + for (const entry of readdirSync("/proc")) { + if (!/^\d+$/.test(entry)) continue; + try { + const stat = readFileSync(`/proc/${entry}/stat`, "utf8"); + const fields = stat.slice(stat.lastIndexOf(")") + 2).split(" "); + if (Number(fields[2]) !== pgid) continue; + if (readFileSync(`/proc/${entry}/environ`, "utf8").split("\0").includes(marker)) ours = true; + } catch { + // the process exited while we looked + } + } + } catch { + // no /proc: nothing can be verified, so nothing is killed + } + if (ours) killGroup(pgid, "SIGKILL"); + try { + unlinkSync(pidFile); + } catch { + // already gone } - } catch { - // no /proc: nothing can be verified, so nothing is killed - } - if (ours) killGroup(pgid, "SIGKILL"); - try { - unlinkSync(workerPidFile()); - } catch { - // already gone } + + return { + async acquire() { + await mkdir(stateRoot, { recursive: true }); + try { + await listen(); + } catch (error) { + if (error?.code !== "EADDRINUSE") throw error; + if ((await probe()) === "live") { + console.error("the state root is held by a running daemon"); + process.exit(2); + } + unlinkSync(lockPath); + await listen(); + } + reclaimStrayWorker(); // synchronous: no request is served before a predecessor's worker is gone + }, + held: () => lock !== null, + refusal: () => "write-lease-pending", + watch() {}, + recordWorker(pid) { + writeFileSync(pidFile, `${pid}\n`); + }, + forgetWorker(pid) { + try { + if (readFileSync(pidFile, "utf8").trim() === String(pid)) unlinkSync(pidFile); + } catch { + // no pid file, or another worker's + } + }, + release() { + if (lock) lock.close(); + lock = null; + try { + unlinkSync(lockPath); + } catch { + // already gone + } + }, + }; } async function stopWorker() { @@ -374,10 +427,10 @@ async function startWorker(model, timeoutMs) { }); closeSync(logFd); if (proc.pid !== undefined) { - writeFileSync(workerPidFile(), `${proc.pid}\n`); + runtime.recordWorker(proc.pid); proc.once("exit", () => { killGroup(proc.pid, "SIGKILL"); // whatever the worker left running, an encoder child included - forgetWorkerPid(proc.pid); + runtime.forgetWorker(proc.pid); }); } Object.assign(worker, { state: "starting", proc, pid: proc.pid ?? null, port: null, servedModels: [], lastError: null }); @@ -463,6 +516,10 @@ async function readRequestBody(request) { } async function generateMotion(request, response) { + if (!runtime.held()) { + json(response, 503, { ok: false, code: runtime.refusal() }); + return; + } if (workerUnconfigured()) { json(response, 503, { ok: false, code: "worker-unconfigured" }); return; @@ -593,6 +650,11 @@ async function generateMotion(request, response) { result = { ok: false, error: `the worker request failed (${error?.name || "Error"})`, body: null }; } + if (!runtime.held()) { + // the writer was lost while the job ran: nothing more is written + json(response, 503, { ok: false, code: runtime.refusal() }); + return; + } const job = { schema: "gamecult.vili.motion_job.v0", jobId, @@ -612,8 +674,48 @@ async function generateMotion(request, response) { json(response, result.ok ? 200 : 502, job); } +const SNAPSHOT_BODY_MAX = 64 * 1024; + +// Idunn's route challenge: a msgpack POST from the host, answered from local authority. A browser page can send +// it too, so an Origin is refused exactly as on every other POST. Only the Idunn runtime answers; the hand-run +// daemon has no such route. +async function answerRouteChallenge(request, response) { + if (!runtime.answerSnapshot) { + text(response, 404, "not found\n"); + return; + } + if (request.headers.origin !== undefined) { + json(response, 403, { ok: false, code: "cross-site" }); + return; + } + const chunks = []; + let size = 0; + for await (const chunk of request) { + size += chunk.length; + if (size > SNAPSHOT_BODY_MAX) { + json(response, 400, { ok: false, code: "invalid-snapshot-request" }); + return; + } + chunks.push(chunk); + } + let answer; + try { + answer = runtime.answerSnapshot(Buffer.concat(chunks)); + } catch (error) { + if (error?.code === "invalid-snapshot-request") json(response, 400, { ok: false, code: error.code }); + else json(response, 503, { ok: false, code: "runtime-presence-unavailable" }); + return; + } + response.writeHead(200, { "content-type": "application/msgpack", "cache-control": "no-store" }); + response.end(answer); +} + async function handleRequest(request, response) { const url = new URL(request.url, `http://${request.headers.host || `${args.host}:${args.port}`}`); + if (url.pathname === "/cultnet/snapshot" && request.method === "POST") { + await answerRouteChallenge(request, response); + return; + } if (request.method === "POST" && refuseCrossSite(request, response)) return; if (url.pathname === "/" || url.pathname === "/health") { @@ -627,23 +729,13 @@ async function handleRequest(request, response) { } } -async function healthCheck() { - const host = args.host === "0.0.0.0" ? "127.0.0.1" : args.host; - const response = await fetch(`http://${host}:${args.port}/health`); - if (!response.ok) throw new Error(`health returned ${response.status}`); - const body = await response.json(); - if (!body.ok) throw new Error("health returned ok=false"); -} - -if (args.health) { - healthCheck().then(() => process.exit(0), (error) => { - console.error(error.message); - process.exit(1); - }); -} else { - await mkdir(path.join(args.stateRoot, "jobs"), { recursive: true }); - await persistSurfaces(); +let runtime; +// The one daemon. `rt` owns the state root: the hand-run runtime (socket lock) or Idunn's (process write lease). +// The server listens first, so Idunn's challenge can be answered while the lease is still pending; nothing is +// written to the state root until the runtime holds it. +export async function serve(rt) { + runtime = rt; const server = http.createServer((request, response) => { handleRequest(request, response).catch((error) => { // The error's message can carry the request; name only its class. @@ -652,20 +744,39 @@ if (args.health) { else response.end(); }); }); + await new Promise((resolve) => server.listen(args.port, args.host, resolve)); + console.log(`Vili listening on http://${args.host}:${server.address().port}`); - server.listen(args.port, args.host, () => { - reclaimStrayWorker(); // synchronous: no request is served before a predecessor's worker is gone - console.log(`Vili listening on http://${args.host}:${server.address().port}`); - }); - - const surfaceTimer = setInterval(() => persistSurfaces().catch((error) => console.error(`persist failed: ${error?.name || "Error"}`)), 5000); - + let surfaceTimer = null; const shutdown = async () => { clearInterval(surfaceTimer); await stopWorker(); + runtime.release(); server.close(); process.exit(0); }; process.on("SIGTERM", shutdown); process.on("SIGINT", shutdown); + + await runtime.acquire(); + await mkdir(path.join(args.stateRoot, "jobs"), { recursive: true }); + await persistSurfaces(); + let lost = false; + const onLost = () => { + if (lost) return; + lost = true; + clearInterval(surfaceTimer); + stopWorker().finally(() => runtime.reportHealth?.("degraded")); + }; + surfaceTimer = setInterval(() => { + runtime.watch(onLost); + if (lost) return; + persistSurfaces().catch((error) => console.error(`persist failed: ${error?.name || "Error"}`)); + }, 5000); + runtime.reportHealth?.("active"); + console.log("Vili holds the state root"); +} + +if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) { + await serve(handRunRuntime(args.stateRoot)); } diff --git a/vendor/SOMA-X b/vendor/SOMA-X new file mode 160000 index 0000000..56a3d95 --- /dev/null +++ b/vendor/SOMA-X @@ -0,0 +1 @@ +Subproject commit 56a3d957a549c66425b500fddf7f3183bc6fbdff diff --git a/vendor/kimodo b/vendor/kimodo new file mode 160000 index 0000000..6bb5848 --- /dev/null +++ b/vendor/kimodo @@ -0,0 +1 @@ +Subproject commit 6bb58488037dd65360ff0c5d1692b403a23309f7