Merge cut/vili-cpu-body: Vili as one Linux CPU process tree, encoder header-on (Soul s1-s2)

WSL, Docker, GPU and Raven launchers leave main. Not deployed: admission on Yggdrasil is the next cut. s2 residue in motion-pipeline:follow_up:vili-cpu-body-s2-residue.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-23 18:31:49 +02:00
commit f2b35e3837
23 changed files with 1885 additions and 946 deletions

3
.gitmodules vendored Normal file
View file

@ -0,0 +1,3 @@
[submodule "vendor/CultLib"]
path = vendor/CultLib
url = https://github.com/GameCult/CultLib.git

136
README.md
View file

@ -3,68 +3,72 @@
Vili is the GameCult Persona animation daemon.
It owns the bridge from Persona performance intent to body-motion generation. Its
first backend is Kimodo on Raven, running through Docker Engine inside
`Ubuntu-24.04` WSL with NVIDIA GPU passthrough.
backend is Kimodo, run by Vili's own daemon as a CPU-only Python process tree on
Linux. Admission to Idunn on Yggdrasil is the next cut; until then nothing
supervises Vili.
## Front Door
- **Value:** a warm, supervised path from a text prompt to generated body motion. Vili runs Kimodo as a resident GPU worker on Raven and returns motion artifacts (`motion.npz`, optional BVH) for a job.
- **Problem:** Personas and games need generated motion, but cold per-request container runs took about 195 seconds per 1 second clip, versus about 54 warm. One daemon owns job identity, worker lifecycle and artifact references so callers do not touch Docker, WSL or model state.
- **Entry points:** `scripts/vili-daemon.mjs` (the daemon; `npm start`, `npm run health`, `npm run smoke`, `npm run provider`), `scripts/kimodo-resident-worker.py` (the resident worker), `POST /motion/generate` (see Motion Generation), `docs/verse-service-contract.md` and `docs/motion-pipeline-target.md`.
- **Live:** the resident Kimodo worker on Raven, `POST /motion/generate`, `GET /smoke`, and the CultCache service store. It runs under a Raven scheduled task, not under Idunn. It has `motion.npz` and optional BVH output on the SOMA skeleton; there is no FBX export.
- **Planned (target, 2026-10-20):** FBX animation export on the SOMA skeleton and a text-prompt-to-Unity-clip path with no hand steps, with each artifact recording prompt, constraints, seed and model. Retargeting stays with the consuming engine; see `docs/motion-pipeline-target.md`.
- **Value:** a supervised path from a text prompt to generated body motion. The daemon spawns one resident Kimodo worker, caps its threads, and returns motion artifacts (`motion.npz`, optional BVH) for a job.
- **Problem:** Personas and games need generated motion, but cold per-request model loads cost minutes. One daemon owns job identity, worker lifecycle, the thread budget and artifact references, so callers touch no model state.
- **Entry points:** `scripts/vili-daemon.mjs` (the daemon; `npm start`, `npm run health`, `npm test`), `scripts/kimodo-resident-worker.py` (the worker it spawns), `scripts/kimodo-text-encode.py` (the one-shot encoder child), `POST /motion/generate` (see Motion Generation), `docs/verse-service-contract.md` and `docs/motion-pipeline-target.md`.
- **Live:** the process tree below, run by hand: `POST /motion/generate`, `GET /health`, `GET /operator-state`, and the CultCache service store. It produces `motion.npz` and optional BVH on the SOMA skeleton; there is no FBX export.
- **Planned:** admission to Idunn on Yggdrasil with a locked virtual environment, then FBX export and a text-prompt-to-Unity-clip path with no hand steps (`docs/motion-pipeline-target.md`).
## Authority
- Vili owns animation job intake, Kimodo runtime checks, Kimodo resident worker
lifecycle, generated motion artifact references, and its operator/provider
surface.
- Vili owns animation job intake, the worker process lifecycle, and generated
motion artifact references.
- Personas and Weksa may propose performance intent, gesture prompts, and
constraints.
- Kimodo owns motion synthesis.
- Future renderers own retargeting, playback, facial performance, camera, and
lip sync.
- Odin discovers Vili; Odin does not own Vili's state.
- Vili is started at boot by its scheduled task on Raven today; nothing restarts it after a crash, and Idunn does not manage it until its admission is ruled (deferred until RPS has its clips). Idunn would not own animation decisions.
- Operator surfaces wait for Thing 2; today Vili publishes typed state and plain HTTP.
## Local Runtime
On Raven:
```powershell
scripts\restart-vili-daemon.cmd
```
Default compatibility endpoint:
One Linux CPU process tree:
```text
http://10.77.0.4:8824
vili-daemon.mjs job intake, worker lifecycle, thread cap, nice 10
kimodo-resident-worker.py resident Kimodo model, one generation at a time
kimodo-text-encode.py one child per request with cache misses, on CPU
```
Vili is not an Idunn target today. Its lifecycle is the `GameCult\Vili`
scheduled task on Raven, and `GET /health` is an operator lowering. Admission
to Idunn is motion-pipeline:question:vili-under-idunn.
Vili also writes daemon-owned service state to:
Setup: `git submodule update --init`; in `vendor/CultLib`, `npm ci --ignore-scripts` and
`npm run build --workspace=packages/cultcache-ts`; then `npm ci` here. The worker
needs a Python environment with Kimodo's pins and Kimodo itself importable, and a
Hugging Face cache (`<models root>/hub`) holding the Llama 3 base, both LLM2Vec
adapters and the Kimodo checkpoint. Everything runs offline.
```text
.vili\vili.service.cc
node scripts/vili-daemon.mjs --python <python> --models-root <hf home> --kimodo-commit <sha> --python-env <env id> [--threads 8]
```
That CultCache store owns the provider advertisement, operator state, Eve
surface, command boundary, and transport profile. The adjacent JSON files are
compatibility exports for tools that have not learned the `.cc` store yet.
Flags, each with an environment twin: `--host` (`VILI_DAEMON_HOST`, 127.0.0.1),
`--port` (`VILI_DAEMON_PORT`, 8824), `--state-root` (`VILI_STATE_ROOT`, `.vili`),
`--python` (`VILI_PYTHON`), `--worker-entry` (`VILI_WORKER_ENTRY`, the test seam),
`--models-root` (`VILI_MODELS_ROOT`), `--threads` (`VILI_THREADS`, 1 to 16, default
8), `--kimodo-commit` (`VILI_KIMODO_COMMIT`), `--python-env` (`VILI_PYTHON_ENV`, the
environment's identity). A generate request without a models root, Kimodo commit
and Python environment id gets `503 worker-unconfigured` before anything spawns.
Important routes:
The daemon starts the worker on the first job, on a port the worker binds itself
and reports in its ready line, and respawns it after an exit. The worker runs at
nice 10 with `torch` limited to `--threads` (also `OMP_NUM_THREADS` and
`MKL_NUM_THREADS`), so a shared host keeps its other services. Its log is
`<state>/logs/worker.log`. The daemon owns the whole worker process group: it is killed when the worker exits for any reason, on SIGTERM or SIGINT, and at start-up if a crashed daemon left one (`<state>/worker.pid`). A service unit should also own the tree (systemd `KillMode=control-group`).
- `GET /health`
- `GET /provider-advertisement`
- `GET /operator-state`
- `GET /eve/operator`
- `GET /eve/deck/providers`
- `WS /eve/deck/vili.animation`
- `GET /smoke`
- `POST /motion/generate`
`POST /motion/generate` refuses any POST that carries an `Origin` or is not
`application/json` with `403 cross-site`; no route answers a preflight.
Vili writes daemon-owned state to `.vili/vili.service.cc`: operator state, the
command boundary and the transport profile. Jobs are `.vili/jobs/<jobId>.json`,
artifacts `.vili/artifacts/<jobId>/`.
Routes: `GET /health`, `GET /operator-state`, `POST /motion/generate`.
## Motion Generation
@ -111,14 +115,14 @@ Supported request fields:
- `prompt` / `utterance` plus `duration` / `durationSeconds`
- `texts` plus `durations`, or a full `meta` object
- `inputFolder` / `input_folder` for a container-visible Kimodo input folder
- `inputFolder` / `input_folder` for a Kimodo input folder on the worker host
with `meta.json` and optional `constraints.json`
- `model`; changing the requested model restarts the single resident worker
- `diffusionSteps` / `diffusion_steps`
- `numSamples` / `num_samples`
- `numTransitionFrames` / `num_transition_frames`
- `constraints` as an object/list, or `constraints`, `constraintsPath`, or
`constraints_path` as a container-visible path
`constraints_path` as a path on the worker host
- `seed`
- `cfg`, `cfgType` / `cfg_type`, `cfgWeight` / `cfg_weight`
- `bvh`, `bvhStandardTpose` / `bvh_standard_tpose`
@ -134,41 +138,25 @@ exports CSV when that model is resident.
## Current Runtime State
Raven has a Hugging Face token installed at
`/root/.cache/huggingface/token` inside `Ubuntu-24.04`. Vili mounts that token
read-only into Kimodo containers and passes the standard Hugging Face token
environment variables.
Vili keeps Kimodo warm through a resident worker container named
`vili-kimodo-worker`. The worker loads the model once, exposes `/health` and
`/generate` inside WSL on `127.0.0.1:8825`, and serializes generation through
the loaded model. Because Raven's Docker Engine runs inside WSL, Vili also
keeps a tiny WSL sleep-loop alive while the worker is resident so the Docker
substrate does not disappear underneath the model. `GET /health` observes
worker state without launching model load. `GET /smoke` verifies the backend
and starts the resident worker when it is absent.
Kimodo's 8B text encoder (LLM2Vec, Llama 3) never lives in the worker. The
worker answers Kimodo's encoder call from a CultCache store of float32 vectors,
`.vili/embeddings/text-embeddings.cc` (`gamecult.vili.text_embedding.v0`, written
through the cultcache-py checkout mounted from `E:\Projects\CultLib`). A vector's
key is `sha256(encoderIdentity + newline + text)`, and the identity names the three
Hugging Face snapshot revisions, bfloat16, the transformers and peft versions and
the Kimodo commit, so a changed input misses instead of serving a stale vector.
Kimodo normalizes the prompt before encoding (`A person shrugs sadly.`), and the
cache keys on that text. A request's misses go to one transient child,
`scripts/kimodo-text-encode.py`: it loads the encoder in bf16 on the GPU, with the
layers that do not fit in free VRAM held in RAM, encodes, writes its answer and
exits, so the OS reclaims its RAM and VRAM. It refuses with a typed `gpu-busy`
below 2 GiB of free VRAM and has no CPU fallback; hits never touch it. Each
`/motion/generate` job carries the encoder record (hits, misses, seconds, vector
hashes) and the worker identity. `GET /health` shows the identity and the stored
count.
`<state>/embeddings/text-embeddings.cc` (`gamecult.vili.text_embedding.v0`, through
the cultcache-py in `vendor/CultLib`). A vector's key is
`sha256(encoderIdentity + newline + text)`. The identity names the three Hugging
Face snapshot revisions, the encoder assembly and prompt form, the device, dtype, the torch,
transformers, peft and accelerate versions, the Kimodo commit and the sha256 of
`kimodo-text-encode.py`, so a changed input misses instead of serving a stale
vector. Thread count is provenance, recorded in the worker identity, not identity.
Measured on Raven on `2026-10-22`, for a 2 second clip at 20 diffusion steps:
the worker is healthy 29 seconds after `docker rm` (was 179) and rests at 2.3 GiB
resident (was 10.4 GiB, with 5.8 GB swapped). A cache miss takes 59 to 103
seconds, the encode being nearly all of it (the 103 was with the model files cold
in the page cache); the CPU encoder took 316 to 376. A hit takes under one
second. The child peaks at 14.1 GiB of VRAM, so a miss needs almost the whole
16 GB card.
The encoder is the Llama 3 base with the MNTP adapter merged, then the supervised
adapter, in bfloat16 on CPU, and `kimodo-text-encode.py` wraps every prompt in
Llama 3 Instruct's chat header itself (`text_encoding.chat_header`, pooled after
LLM2Vec's query marker), as Kimodo's LLM2Vec does when loaded from the repo ids. The identity names that prompt form.
A fresh child vector matches Kimodo's own encoder from the repo ids
(`test_text_encoding.ReferenceVector`, run on Yggdrasil). The measured timings are
in `docs/research/vili-on-yggdrasil-cpu-2026-10.md`. Kimodo normalizes the prompt before encoding (`A person shrugs sadly.`),
and the cache keys on that text. A request's misses go to one transient child that
loads the encoder, encodes, writes its answer and exits, so the OS reclaims its
RAM; hits never touch it. Each job carries the encoder record (hits, misses,
seconds, vector hashes) and the worker identity. The cache starts empty on a new
host: vectors from another assembly or device are never copied in.

View file

@ -159,3 +159,5 @@ layer by layer and needs about 2 GiB resident, so it would fit Raven on CPU.
- **Yggdrasil scratch** is `~/scratch/kimodo-cpp-probe` (39 GiB: weights,
checkouts, outputs). The probe's containers are all removed. Delete the scratch
directory once nobody needs to rerun the probe.
**Correction, 2026-10-23.** The "MNTP missing" diagnosis in this document is wrong. The MNTP adapter is applied in both loads (the cut measured a q_proj difference of 0.0117 for repo ids and snapshot paths alike). What separates A from B is the Llama-3 chat header: Kimodo's `LLM2Vec.prepare_for_tokenization` wraps the text in it only when the model's `_name_or_path` is the repo id `meta-llama/Meta-Llama-3-8B-Instruct`, which a load from a snapshot path silently drops (`motion-pipeline:finding:cut-vili-cpu-body.s1.ab-premise-inverted`). B's weights with the bare header give cosine 0.9977 to A, and without it 1.0 to B; the header plus LLM2Vec's instruction marker, which keeps the header out of the pooling as Kimodo's encode path does, gives 1.0 to A (checked 2026-10-23). Vili now applies the header explicitly (`motion-pipeline:ruling:encoder-chat-header-on`). The measurements above stand as measured; read "MNTP" in the A/B account as "the chat header".

View file

@ -180,3 +180,5 @@ finger joints only coarsely.
- **Scratch:** `~/scratch/vili-ygg-cpu` holds the scripts, logs (`run*.log`, `out/*.log`),
vectors, joints and BVH files, plus 18 GiB of weights. Delete it once nobody reruns this.
All its containers are removed.
**Correction, 2026-10-23.** The "MNTP missing" diagnosis in this document is wrong. The MNTP adapter is applied in both loads (the cut measured a q_proj difference of 0.0117 for repo ids and snapshot paths alike). What separates A from B is the Llama-3 chat header: Kimodo's `LLM2Vec.prepare_for_tokenization` wraps the text in it only when the model's `_name_or_path` is the repo id `meta-llama/Meta-Llama-3-8B-Instruct`, which a load from a snapshot path silently drops (`motion-pipeline:finding:cut-vili-cpu-body.s1.ab-premise-inverted`). B's weights with the bare header give cosine 0.9977 to A, and without it 1.0 to B; the header plus LLM2Vec's instruction marker, which keeps the header out of the pooling as Kimodo's encode path does, gives 1.0 to A (checked 2026-10-23). Vili now applies the header explicitly (`motion-pipeline:ruling:encoder-chat-header-on`). The measurements above stand as measured; read "MNTP" in the A/B account as "the chat header".

View file

@ -2,43 +2,29 @@
## Objective
Vili turns Persona performance intent into animation work. The near-term body is
Kimodo on Raven; the durable contract is a service surface that can accept a
gesture/action prompt, run or queue motion synthesis, and publish inspectable
state for Odin and Idunn.
Vili is an animation service, not a per-request CLI wrapper. Its Kimodo backend
is a resident worker container that owns model lifetime for generation requests.
Vili turns Persona performance intent into animation work. It is an animation
service, not a per-request CLI wrapper: one daemon accepts a gesture or action
prompt, owns a resident Kimodo worker on CPU, and publishes inspectable state
for Odin and Idunn. The request fields and the runtime flags are in `README.md`
(Motion Generation, Local Runtime); this page holds the contract around them.
## CultMesh Verses
- `vili.service`: service identity, provider advertisement, health, and runtime.
- `vili.service`: service identity, health, and runtime.
- `vili.animation`: animation jobs, prompts, backend choices, and artifact refs.
- `vili.operator`: daemon status, backend readiness, and Eve/CultUI surface.
- `vili.operator`: daemon status and worker readiness, as typed state. Operator
surfaces wait for Thing 2.
## Witnesses
- `.vili/vili.service.cc`: Vili-owned CultCache store for provider
advertisement, operator state, Eve surface, command boundary, and transport
profile records.
- `.vili/provider-advertisement.json`: compatibility export derived from the
provider advertisement record.
- `.vili/operator-state.json`: compatibility export derived from daemon health
and backend readiness.
- `.vili/eve-operator-surface.json`: compatibility export of the current
operator surface.
- `.vili/command-boundary.json`: compatibility export of Vili command
authority.
- `.vili/transport-profile.json`: compatibility export of Vili transport
state.
- `.vili/vili.service.cc`: the daemon-owned CultCache store for operator state,
the command boundary and the transport profile.
- `.vili/jobs/{jobId}.json`: accepted animation job records.
- `.vili/artifacts/{jobId}/`: generated motion outputs when generation succeeds.
- `vili-kimodo-worker`: resident Docker container exposing the loaded Kimodo
model to Vili inside Raven's `Ubuntu-24.04` WSL runtime.
JSON is a compatibility witness. Vili's service/operator/command/transport
truth is the daemon-owned `.cc` store; animation job schemas may still be
promoted from JSON once that request/receipt shape stabilizes.
- `.vili/embeddings/text-embeddings.cc`: the worker's text-embedding cache.
- `.vili/worker.pid`: the process group of the running worker, so a restarted
daemon can find and kill a worker its crashed predecessor left behind.
- `.vili/logs/worker.log`: the worker's stdout and stderr.
## Command Boundary
@ -47,112 +33,33 @@ supported Kimodo generation controls to the resident worker. Vili owns job
identity, request persistence, worker lifecycle, timeout handling, and the
artifact stem. Kimodo owns synthesis and model-specific export.
```json
{
"prompt": "a person gives a small friendly wave",
"duration": 1,
"diffusionSteps": 20,
"seed": 123,
"bvh": true
}
```
Kimodo metadata can be sent directly when callers need multi-segment or
multi-sample generation:
```json
{
"meta": {
"texts": [
"a person gives a small friendly wave",
"a person takes one careful step"
],
"durations": [0.5, 0.5],
"num_samples": 1,
"diffusion_steps": 20
},
"model": "Kimodo-SOMA-RP-v1.1",
"numTransitionFrames": 3,
"cfgType": "regular",
"cfgWeight": [1.5],
"seed": 6101,
"bvh": true,
"bvhStandardTpose": true,
"saveExampleDir": true
}
```
Request fields accepted by Vili:
- `prompt` / `utterance` plus `duration` / `durationSeconds`
- `texts` plus `durations`, or a full `meta` object
- `inputFolder` / `input_folder` for a container-visible Kimodo input folder
containing `meta.json` and optional `constraints.json`
- `model`; the daemon keeps one resident model loaded, so a model change
restarts `vili-kimodo-worker`
- `diffusionSteps` / `diffusion_steps`
- `numSamples` / `num_samples`
- `numTransitionFrames` / `num_transition_frames`
- `constraints` as an object/list, or `constraints`, `constraintsPath`, or
`constraints_path` as a container-visible path
- `seed`
- `cfg`, `cfgType` / `cfg_type`, `cfgWeight` / `cfg_weight`
- `bvh`, `bvhStandardTpose` / `bvh_standard_tpose`
- `noPostprocess` / `no_postprocess`
- `saveExampleDir` / `save_example_dir`
- `timeoutMs`, `workerTimeoutMs`
Vili does not accept arbitrary `output` paths from callers. Every generation
writes beneath `.vili/artifacts/{jobId}/`, with the worker receiving
`/outputs/{jobId}/motion`. The job record is the authority for artifact refs.
Model-specific exports are preserved: SOMA writes NPZ and optional BVH, SMPLX
also writes AMASS NPZ, and G1 also writes CSV.
writes beneath `.vili/artifacts/{jobId}/`, and the job record is the authority
for artifact refs. SOMA writes NPZ and optional BVH, SMPLX also writes AMASS NPZ,
and G1 also writes CSV.
The daemon may reject or defer requests when:
A POST that carries an `Origin` header or is not `application/json` is refused
`403 cross-site` before its body is read.
- Kimodo image is unavailable.
- Docker/GPU backend is unhealthy.
- Hugging Face token is absent and the selected backend requires gated model
access.
- The resident Kimodo worker cannot start or does not become healthy before the
request timeout.
- The request omits concrete prompt, metadata, batch text, or input-folder
intent.
- Batch `texts` are provided without matching `durations`.
The daemon refuses or fails a request when:
## Credential State
- the models root, Kimodo commit or Python environment id is not configured
(`503 worker-unconfigured`, before anything spawns);
- the worker cannot start or does not become healthy in time (a `502` job);
- the request omits concrete prompt, metadata, batch text, or input-folder
intent;
- batch `texts` are provided without matching `durations`.
Raven has a Hugging Face token installed in the `Ubuntu-24.04` WSL cache, and
Vili mounts it into Kimodo containers read-only. Verified on `2026-06-12`: a
Kimodo container launched by Raven could read gated metadata for
`meta-llama/Meta-Llama-3-8B-Instruct`.
## Worker Boundary
## Kimodo Worker Boundary
The daemon spawns the resident worker as a child on the first job, in its own
process group, at nice 10, with a closed environment and the thread budget
`--threads`. The worker binds its own port and reports it in a ready line. The
daemon owns the group: it kills the whole group when the worker exits for any
reason (so no encoder child survives), on SIGTERM or SIGINT, and at start-up for
a group left by a crashed daemon. Under a service manager, the unit should also
own the tree from outside (systemd `KillMode=control-group`).
The resident worker is started by Vili as Docker container `vili-kimodo-worker`.
Because Raven's Docker Engine lives inside WSL, Vili also starts a minimal WSL
keepalive process recorded at `/tmp/vili-wsl-keepalive.pid`; it owns no service
truth and exists only to keep the Docker substrate alive while the resident
worker holds the model.
It mounts:
- `/root/.cache/huggingface` for model/token access.
- `E:\Projects\Vili\scripts\kimodo-resident-worker.py` as the worker program.
- `E:\Projects\Vili\.vili\artifacts` as `/outputs` for generated artifacts.
Inside the worker, Kimodo owns synthesis. Vili owns worker lifecycle,
request/response persistence, timeout handling, and artifact references. Direct
user requests, programmatic requests, and smoke-started warmup all use the same
resident worker health and generation path.
Observed on `2026-06-12`: first worker model load took about 100 seconds. Once
resident, five 1 second / 20 step service-path generations averaged about
54.4 seconds per clip, or about `0.0184x` realtime.
## First Runtime Invariant
Health checks must be cheap. They may verify process, port, backend metadata,
and cached readiness, but they must not launch a Kimodo generation job or start
model load. Smoke checks may start the resident worker because their purpose is
backend readiness, not cheap liveness.
Inside the worker, Kimodo owns synthesis and a one-shot encoder child owns
text encoding on a cache miss. Health checks must be cheap: `/health` reads the
daemon's own worker state and never starts a model load or a generation.

451
package-lock.json generated
View file

@ -8,11 +8,433 @@
"name": "@gamecult/vili",
"version": "0.1.0",
"dependencies": {
"cultcache-ts": "file:../CultLib/packages/cultcache-ts",
"ws": "^8.18.0"
"cultcache-ts": "file:vendor/CultLib/packages/cultcache-ts"
}
},
"../CultLib/packages/cultcache-ts": {
"name": "@gamecult/cultcache-ts",
"version": "0.14.2",
"extraneous": true,
"license": "MIT",
"dependencies": {
"@msgpack/msgpack": "^3.1.3",
"koffi": "3.3.0",
"zod": "^3.25.76"
},
"devDependencies": {
"@types/node": "^24.0.0",
"typescript": "^5.9.0"
},
"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",
"integrity": "sha512-lHXE2mq+uG6CVeX1iepn2ATcAWMZPagr+J0ICYge7wTMDYc5pPRQMq0Rr/16775i6QlUqQqpk8c5Gu66+u3gGA==",
"cpu": [
"arm64"
],
"license": "MIT",
"optional": true,
"os": [
"android"
],
"funding": {
"url": "https://liberapay.com/Koromix"
}
},
"node_modules/@koromix/koffi-android-x64": {
"version": "3.3.0",
"resolved": "https://registry.npmjs.org/@koromix/koffi-android-x64/-/koffi-android-x64-3.3.0.tgz",
"integrity": "sha512-7EeVK/KS9XNbOsUfy0B3uuSjPQZenjJ35y9QSi6R6eK4Fv0oGvREQaBzIJRJwonxp7sJfGObnwnQTnKTVK2DWA==",
"cpu": [
"x64"
],
"license": "MIT",
"optional": true,
"os": [
"android"
],
"funding": {
"url": "https://liberapay.com/Koromix"
}
},
"node_modules/@koromix/koffi-darwin-arm64": {
"version": "3.3.0",
"resolved": "https://registry.npmjs.org/@koromix/koffi-darwin-arm64/-/koffi-darwin-arm64-3.3.0.tgz",
"integrity": "sha512-2R9bihYmYFyRP98wRG2m5WQVKMONf7DoPJI8RkGCOUB2afOXJEVXFepiSKxBZGPml8i1USwAxMUwkzbJ5uMG1A==",
"cpu": [
"arm64"
],
"license": "MIT",
"optional": true,
"os": [
"darwin"
],
"funding": {
"url": "https://liberapay.com/Koromix"
}
},
"node_modules/@koromix/koffi-darwin-x64": {
"version": "3.3.0",
"resolved": "https://registry.npmjs.org/@koromix/koffi-darwin-x64/-/koffi-darwin-x64-3.3.0.tgz",
"integrity": "sha512-OecYCQfzhUx+XyAtatIHlk0jIz87wKNjtYvwlwgMwgaGQwO/TxFeReNo+s3167Rh1a9c8rUIOjq1jfzAPBxbKA==",
"cpu": [
"x64"
],
"license": "MIT",
"optional": true,
"os": [
"darwin"
],
"funding": {
"url": "https://liberapay.com/Koromix"
}
},
"node_modules/@koromix/koffi-freebsd-arm64": {
"version": "3.3.0",
"resolved": "https://registry.npmjs.org/@koromix/koffi-freebsd-arm64/-/koffi-freebsd-arm64-3.3.0.tgz",
"integrity": "sha512-6ytHzV1DuyxTP5LtFmQ3imBk71j/F/Nb0Hcwhjcn35esnknm5rQbgRl6tNRldrkGnM4GIAJpsagooUBh1xAzYg==",
"cpu": [
"arm64"
],
"license": "MIT",
"optional": true,
"os": [
"freebsd"
],
"funding": {
"url": "https://liberapay.com/Koromix"
}
},
"node_modules/@koromix/koffi-freebsd-ia32": {
"version": "3.3.0",
"resolved": "https://registry.npmjs.org/@koromix/koffi-freebsd-ia32/-/koffi-freebsd-ia32-3.3.0.tgz",
"integrity": "sha512-hINo5z1Iz2u3Rx3oHfzcmi5A03gYAoYf4nCJJYaY0bb5IuGdIC7noQTyyI9YxfyTJgoqw3wKFWWYtvsAmEo9Hw==",
"cpu": [
"ia32"
],
"license": "MIT",
"optional": true,
"os": [
"freebsd"
],
"funding": {
"url": "https://liberapay.com/Koromix"
}
},
"node_modules/@koromix/koffi-freebsd-x64": {
"version": "3.3.0",
"resolved": "https://registry.npmjs.org/@koromix/koffi-freebsd-x64/-/koffi-freebsd-x64-3.3.0.tgz",
"integrity": "sha512-7XQYAZh828nW6+y3IdyUNdo2VbVHL5ZC8Qb+yqHrlDjTjVXSyGYZJHAyJZ5c7eSTTKBhtIMTX238rIsdexxrog==",
"cpu": [
"x64"
],
"license": "MIT",
"optional": true,
"os": [
"freebsd"
],
"funding": {
"url": "https://liberapay.com/Koromix"
}
},
"node_modules/@koromix/koffi-linux-arm": {
"version": "3.3.0",
"resolved": "https://registry.npmjs.org/@koromix/koffi-linux-arm/-/koffi-linux-arm-3.3.0.tgz",
"integrity": "sha512-NbahRKKKFkHoFN/ZpLY+w4F6eYSqV4KMN40PmjZ1xQjG/XjYb6PIYgE41c7/acAPxOYqujrcaI3B4BAgWYm5/A==",
"cpu": [
"arm"
],
"license": "MIT",
"optional": true,
"os": [
"linux"
],
"funding": {
"url": "https://liberapay.com/Koromix"
}
},
"node_modules/@koromix/koffi-linux-arm64": {
"version": "3.3.0",
"resolved": "https://registry.npmjs.org/@koromix/koffi-linux-arm64/-/koffi-linux-arm64-3.3.0.tgz",
"integrity": "sha512-Eo/+wGeawGICzJkUlkzi/KerAJ6U1L/Hf7cnJKH7aGgCOY5RRnjUqXs/eL+UDpRcFHzGeUn+Fq+8q2VxYxvAfw==",
"cpu": [
"arm64"
],
"license": "MIT",
"optional": true,
"os": [
"linux"
],
"funding": {
"url": "https://liberapay.com/Koromix"
}
},
"node_modules/@koromix/koffi-linux-ia32": {
"version": "3.3.0",
"resolved": "https://registry.npmjs.org/@koromix/koffi-linux-ia32/-/koffi-linux-ia32-3.3.0.tgz",
"integrity": "sha512-u9tKuj42RFWJU8ZgZqcBhkFCkIZE/SE6oR1ljpukttJUhOVuaf/lmsWxyOxiVu3U8T1gtD658FlLgrbMVIn6aQ==",
"cpu": [
"ia32"
],
"license": "MIT",
"optional": true,
"os": [
"linux"
],
"funding": {
"url": "https://liberapay.com/Koromix"
}
},
"node_modules/@koromix/koffi-linux-loong64": {
"version": "3.3.0",
"resolved": "https://registry.npmjs.org/@koromix/koffi-linux-loong64/-/koffi-linux-loong64-3.3.0.tgz",
"integrity": "sha512-2hKlNZ6OXkTUp6fcueZcP0sQ6yhpdfANg9/mWxDKzT1o8qIj4qJD7UcZLCcRIC8JXiP8JQwZBtlPuyXIqofDQg==",
"cpu": [
"loong64"
],
"license": "MIT",
"optional": true,
"os": [
"linux"
],
"funding": {
"url": "https://liberapay.com/Koromix"
}
},
"node_modules/@koromix/koffi-linux-ppc64": {
"version": "3.3.0",
"resolved": "https://registry.npmjs.org/@koromix/koffi-linux-ppc64/-/koffi-linux-ppc64-3.3.0.tgz",
"integrity": "sha512-bIVAoHgzyRN7xs529BK7rF3BbZUTJgIr0t/wvBvAgaIpreKsdZ/QWxhedKRbTN7ljlDHkPZFmu5XcCjb/vqUag==",
"cpu": [
"ppc64"
],
"license": "MIT",
"optional": true,
"os": [
"linux"
],
"funding": {
"url": "https://liberapay.com/Koromix"
}
},
"node_modules/@koromix/koffi-linux-riscv64": {
"version": "3.3.0",
"resolved": "https://registry.npmjs.org/@koromix/koffi-linux-riscv64/-/koffi-linux-riscv64-3.3.0.tgz",
"integrity": "sha512-2niYAe+TBXbvf7yog/aiHiDUrKQn+RkMSH5x3ZYo65OQyII1CY3Jk2kP3Pe735gq0LzIYJsDuGMW+02ighSadQ==",
"cpu": [
"riscv64"
],
"license": "MIT",
"optional": true,
"os": [
"linux"
],
"funding": {
"url": "https://liberapay.com/Koromix"
}
},
"node_modules/@koromix/koffi-linux-x64": {
"version": "3.3.0",
"resolved": "https://registry.npmjs.org/@koromix/koffi-linux-x64/-/koffi-linux-x64-3.3.0.tgz",
"integrity": "sha512-apU8ZMOrOh1eqMP/Pp0sDZ/WY6A4r6JWph26h63VM9nZ+5KCTYuPgojfoLPNvrfI/bn1PY3+bpXKRhZHI3gxbQ==",
"cpu": [
"x64"
],
"license": "MIT",
"optional": true,
"os": [
"linux"
],
"funding": {
"url": "https://liberapay.com/Koromix"
}
},
"node_modules/@koromix/koffi-openbsd-arm64": {
"version": "3.3.0",
"resolved": "https://registry.npmjs.org/@koromix/koffi-openbsd-arm64/-/koffi-openbsd-arm64-3.3.0.tgz",
"integrity": "sha512-iTyHMCSSs+X579sWJv7XVFMp9gOfROI8IUEwp09YxJpGvmiSoipI0J7kytFuFq6dyuwO0Ymy4JdlmKnCM0qkxg==",
"cpu": [
"arm64"
],
"license": "MIT",
"optional": true,
"os": [
"openbsd"
],
"funding": {
"url": "https://liberapay.com/Koromix"
}
},
"node_modules/@koromix/koffi-openbsd-ia32": {
"version": "3.3.0",
"resolved": "https://registry.npmjs.org/@koromix/koffi-openbsd-ia32/-/koffi-openbsd-ia32-3.3.0.tgz",
"integrity": "sha512-xCvMV+4VNHjTxgVPEsBzU0k8dlA7aoFO5QjRF9lVU/NzGhUedlYvFgaF6IWsGOoWEaNt9hgP4KK/wxMbm9v8EQ==",
"cpu": [
"ia32"
],
"license": "MIT",
"optional": true,
"os": [
"openbsd"
],
"funding": {
"url": "https://liberapay.com/Koromix"
}
},
"node_modules/@koromix/koffi-openbsd-x64": {
"version": "3.3.0",
"resolved": "https://registry.npmjs.org/@koromix/koffi-openbsd-x64/-/koffi-openbsd-x64-3.3.0.tgz",
"integrity": "sha512-xItIwx1elM8k1pTudEfuV/1U2gQ06+rFfxCVXrZkyWYYGTzmA5To00f8garJFR+E9gbjQJbn8QDFvWMI+i2sMA==",
"cpu": [
"x64"
],
"license": "MIT",
"optional": true,
"os": [
"openbsd"
],
"funding": {
"url": "https://liberapay.com/Koromix"
}
},
"node_modules/@koromix/koffi-win32-arm64": {
"version": "3.3.0",
"resolved": "https://registry.npmjs.org/@koromix/koffi-win32-arm64/-/koffi-win32-arm64-3.3.0.tgz",
"integrity": "sha512-fD41LgKBx5i1YlJDuds4tRHAaG5vV55tvagzJwKFDR/5BDeaFvH9QFMUMDoNl2LkmVyeug0dIElHr/oxejh3IA==",
"cpu": [
"arm64"
],
"license": "MIT",
"optional": true,
"os": [
"win32"
],
"funding": {
"url": "https://liberapay.com/Koromix"
}
},
"node_modules/@koromix/koffi-win32-ia32": {
"version": "3.3.0",
"resolved": "https://registry.npmjs.org/@koromix/koffi-win32-ia32/-/koffi-win32-ia32-3.3.0.tgz",
"integrity": "sha512-BfDbDVEuQwPZOvWXawc3JT/7i9cN7QHYotgo014jjXb9pNv46Ykhb0RRglkcHg+SdytW/7pxYn18YwcQP6LLvQ==",
"cpu": [
"ia32"
],
"license": "MIT",
"optional": true,
"os": [
"win32"
],
"funding": {
"url": "https://liberapay.com/Koromix"
}
},
"node_modules/@koromix/koffi-win32-x64": {
"version": "3.3.0",
"resolved": "https://registry.npmjs.org/@koromix/koffi-win32-x64/-/koffi-win32-x64-3.3.0.tgz",
"integrity": "sha512-D+ORp73rTmvGVfq01+4uBmgVZqrPqU9N9xCZCfexaCDlVk0WQsaqid1dGsi6eScJuHt4Cp8s1i5/4CsecxFTKQ==",
"cpu": [
"x64"
],
"license": "MIT",
"optional": true,
"os": [
"win32"
],
"funding": {
"url": "https://liberapay.com/Koromix"
}
},
"node_modules/@msgpack/msgpack": {
"version": "3.1.3",
"resolved": "https://registry.npmjs.org/@msgpack/msgpack/-/msgpack-3.1.3.tgz",
"integrity": "sha512-47XIizs9XZXvuJgoaJUIE2lFoID8ugvc0jzSHP+Ptfk8nTbnR8g788wv48N03Kx0UkAv559HWRQ3yzOgzlRNUA==",
"license": "ISC",
"engines": {
"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,
"license": "MIT",
"dependencies": {
"undici-types": ">=7.24.0 <7.24.7"
}
},
"node_modules/cultcache-ts": {
"resolved": "vendor/CultLib/packages/cultcache-ts",
"link": true
},
"node_modules/koffi": {
"version": "3.3.0",
"resolved": "https://registry.npmjs.org/koffi/-/koffi-3.3.0.tgz",
"integrity": "sha512-UeyQtptuPeCP0BqKD+Q8ftTQhIUBHO7bXyK9DVIaW9hJSg0BrICm4K5N8Qtn7zKk20sb5vpnES9KBqVksM2kpQ==",
"hasInstallScript": true,
"license": "MIT",
"funding": {
"url": "https://liberapay.com/Koromix"
},
"optionalDependencies": {
"@koromix/koffi-android-arm64": "3.3.0",
"@koromix/koffi-android-x64": "3.3.0",
"@koromix/koffi-darwin-arm64": "3.3.0",
"@koromix/koffi-darwin-x64": "3.3.0",
"@koromix/koffi-freebsd-arm64": "3.3.0",
"@koromix/koffi-freebsd-ia32": "3.3.0",
"@koromix/koffi-freebsd-x64": "3.3.0",
"@koromix/koffi-linux-arm": "3.3.0",
"@koromix/koffi-linux-arm64": "3.3.0",
"@koromix/koffi-linux-ia32": "3.3.0",
"@koromix/koffi-linux-loong64": "3.3.0",
"@koromix/koffi-linux-ppc64": "3.3.0",
"@koromix/koffi-linux-riscv64": "3.3.0",
"@koromix/koffi-linux-x64": "3.3.0",
"@koromix/koffi-openbsd-arm64": "3.3.0",
"@koromix/koffi-openbsd-ia32": "3.3.0",
"@koromix/koffi-openbsd-x64": "3.3.0",
"@koromix/koffi-win32-arm64": "3.3.0",
"@koromix/koffi-win32-ia32": "3.3.0",
"@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"
},
"engines": {
"node": ">=14.17"
}
},
"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",
"integrity": "sha512-gzUt/qt81nXsFGKIFcC3YnfEAx5NkunCfnDlvuBSSFS02bcXu4Lmea0AFIUwbLWxWPx3d9p8S5QoaujKcNQxcQ==",
"license": "MIT",
"funding": {
"url": "https://github.com/sponsors/colinhacks"
}
},
"vendor/CultLib/packages/cultcache-ts": {
"name": "@gamecult/cultcache-ts",
"version": "0.14.2",
"license": "MIT",
@ -28,31 +450,6 @@
"engines": {
"node": ">=20"
}
},
"node_modules/cultcache-ts": {
"resolved": "../CultLib/packages/cultcache-ts",
"link": true
},
"node_modules/ws": {
"version": "8.21.0",
"resolved": "https://registry.npmjs.org/ws/-/ws-8.21.0.tgz",
"integrity": "sha512-Vsp28b7DRcimFQvrqu2Wek3z1iYxDCWqHYB8Qsnk/S4RfaCQzPGPyBNuVjJV3cd6UiKtUtp6sNM77gWvzcCH+g==",
"license": "MIT",
"engines": {
"node": ">=10.0.0"
},
"peerDependencies": {
"bufferutil": "^4.0.1",
"utf-8-validate": ">=5.0.2"
},
"peerDependenciesMeta": {
"bufferutil": {
"optional": true
},
"utf-8-validate": {
"optional": true
}
}
}
}
}

View file

@ -7,11 +7,9 @@
"scripts": {
"start": "node scripts/vili-daemon.mjs",
"health": "node scripts/vili-daemon.mjs --health",
"smoke": "node scripts/smoke-vili.mjs",
"provider": "node scripts/vili-daemon.mjs --print-provider-advertisement"
"test": "node --test --test-timeout=60000 scripts/test-vili-daemon.mjs"
},
"dependencies": {
"cultcache-ts": "file:../CultLib/packages/cultcache-ts",
"ws": "^8.18.0"
"cultcache-ts": "file:vendor/CultLib/packages/cultcache-ts"
}
}

View file

@ -0,0 +1,113 @@
// Stands in for kimodo-resident-worker.py in the daemon's tests. It binds port 0, prints the ready line the
// daemon waits for, and answers /health and /generate with what the daemon gave it. The --model value picks a
// misbehaviour: crash-before-ready, hang-before-ready, sick, slow-health, stubborn, fail-generate,
// refuse-generate, drop-generate, slow-generate, encoder (a grandchild, then no answer), graceful (exits 0.5 s after SIGTERM), die-after-ready. Any other value is just the model the worker serves.
import http from "node:http";
import os from "node:os";
import path from "node:path";
import { spawn } from "node:child_process";
import { appendFileSync } from "node:fs";
const argv = process.argv.slice(2);
const flag = (name) => {
const index = argv.indexOf(name);
return index < 0 ? undefined : argv[index + 1];
};
const model = flag("--model");
const spawnLog = path.join(process.env.VILI_EMBEDDINGS_DIR, "stub-spawns.log");
let requests = 0;
let healthCalls = 0;
appendFileSync(spawnLog, `${process.pid}\n`);
console.log("stub-stdout-line");
console.error("stub-stderr-line");
// decoys: neither is the ready line
console.log(JSON.stringify({ event: "progress", port: 1 }));
console.log(JSON.stringify({ event: "kimodo-worker-ready", port: "not-a-port" }));
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`);
}
if (model === "graceful") {
process.on("SIGTERM", () => {
setTimeout(() => {
appendFileSync(spawnLog, "graceful-exit\n");
process.exit(0);
}, 500);
});
}
if (model === "hang-before-ready") {
setInterval(() => {}, 1000);
} else {
const server = http.createServer((request, response) => {
const chunks = [];
request.on("data", (chunk) => chunks.push(chunk));
request.on("end", () => {
const send = (status, body) => {
response.writeHead(status, { "content-type": "application/json" });
response.end(JSON.stringify(body));
};
if (request.method === "GET" && request.url === "/health") {
healthCalls += 1;
if (model === "sick" || model === "die-after-ready" || (model === "slow-health" && healthCalls <= 2)) return send(200, { ok: false });
send(200, { ok: true, model: model || "stub-model", displayModel: model || "Stub Model", pid: process.pid, requests });
} else if (request.method === "POST" && request.url === "/generate") {
requests += 1;
const payload = JSON.parse(Buffer.concat(chunks).toString("utf8"));
if (model === "fail-generate") return send(500, { ok: false });
if (model === "refuse-generate") return send(200, { ok: false });
if (model === "drop-generate") return request.socket.destroy();
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`);
return;
}
send(200, {
ok: true,
model: model || "stub-model",
pid: process.pid,
requests,
contentType: request.headers["content-type"],
output: payload.output,
prompt: payload.prompt,
constraints: payload.constraints,
priority: os.getPriority(),
env: {
OMP_NUM_THREADS: process.env.OMP_NUM_THREADS,
MKL_NUM_THREADS: process.env.MKL_NUM_THREADS,
VILI_THREADS: process.env.VILI_THREADS,
HF_HUB_OFFLINE: process.env.HF_HUB_OFFLINE,
TRANSFORMERS_OFFLINE: process.env.TRANSFORMERS_OFFLINE,
HF_HOME: process.env.HF_HOME,
PYTHONPATH: process.env.PYTHONPATH,
PYTHONDONTWRITEBYTECODE: process.env.PYTHONDONTWRITEBYTECODE,
PYTHONUNBUFFERED: process.env.PYTHONUNBUFFERED,
VILI_KIMODO_COMMIT: process.env.VILI_KIMODO_COMMIT,
VILI_PYTHON_ENV: process.env.VILI_PYTHON_ENV,
PATH: process.env.PATH,
HOME: process.env.HOME,
LANG: process.env.LANG,
LEAK_CHECK: process.env.LEAK_CHECK,
},
argv,
});
} else {
response.writeHead(404);
response.end();
}
});
});
server.listen(Number(flag("--port")), flag("--host"), () => {
console.log(JSON.stringify({ event: "kimodo-worker-ready", port: server.address().port }));
if (model === "die-after-ready") setTimeout(() => process.exit(4), 300);
});
}

View file

@ -1,2 +0,0 @@
@echo off
node "%~dp0vili-daemon.mjs" --health --host 127.0.0.1 --port 8824 --state-root "%~dp0..\.vili"

View file

@ -1,19 +0,0 @@
param(
[string]$TaskName = "GameCult\Vili",
[int]$Port = 8824,
[string]$HostName = "0.0.0.0",
[string]$StateRoot = "E:\Projects\Vili\.vili"
)
$ErrorActionPreference = "Stop"
$ProgressPreference = "SilentlyContinue"
$HiddenLauncher = Join-Path $PSScriptRoot "run-hidden-powershell.vbs"
$RunScript = Join-Path $PSScriptRoot "start-vili-daemon.ps1"
$Action = New-ScheduledTaskAction -Execute "wscript.exe" -Argument "//B //nologo `"$HiddenLauncher`" -NoProfile -NonInteractive -WindowStyle Hidden -ExecutionPolicy Bypass -File `"$RunScript`" -Port $Port -HostName `"$HostName`" -StateRoot `"$StateRoot`""
$Trigger = New-ScheduledTaskTrigger -AtStartup
$UserId = [System.Security.Principal.WindowsIdentity]::GetCurrent().Name
$Principal = New-ScheduledTaskPrincipal -UserId $UserId -LogonType Interactive -RunLevel Limited
$Settings = New-ScheduledTaskSettingsSet -AllowStartIfOnBatteries -DontStopIfGoingOnBatteries -ExecutionTimeLimit ([TimeSpan]::Zero) -MultipleInstances IgnoreNew -Hidden
Register-ScheduledTask -TaskName $TaskName -Action $Action -Trigger $Trigger -Principal $Principal -Settings $Settings -Force | Out-Null
Write-Host "Installed scheduled task $TaskName"

View file

@ -9,9 +9,7 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from threading import Lock
from types import SimpleNamespace
import peft
import torch
import transformers
from huggingface_hub.constants import HF_HOME
from kimodo import DEFAULT_MODEL, load_model
@ -34,32 +32,25 @@ import text_encoding
def parse_args():
parser = argparse.ArgumentParser(description="Resident Vili worker for Kimodo motion generation")
parser.add_argument("--host", default="0.0.0.0")
parser.add_argument("--port", type=int, default=8825)
parser.add_argument("--host", default="127.0.0.1")
parser.add_argument("--port", type=int, default=0)
parser.add_argument("--model", default=DEFAULT_MODEL)
return parser.parse_args()
args = parse_args()
device = "cuda:0" if torch.cuda.is_available() else "cpu"
device = "cpu"
threads = text_encoding.apply_thread_cap(os.environ)
torch.set_num_interop_threads(1)
started_at = time.time()
model_load_started_at = time.time()
preset = TEXT_ENCODER_PRESETS["llm2vec"]["kwargs"]
kimodo_commit = os.environ["VILI_KIMODO_COMMIT"]
kimodo_image = os.environ["VILI_KIMODO_IMAGE"]
encoder_identity = text_encoding.encoder_identity(
revisions={
repo: text_encoding.read_snapshot_revision(HF_HOME, repo)
for repo in (preset["base_model_name_or_path"], preset["peft_model_name_or_path"], text_encoding.LLAMA_BASE)
},
dtype=preset["dtype"],
transformers_version=transformers.__version__,
peft_version=peft.__version__,
kimodo_commit=kimodo_commit,
)
python_env = os.environ["VILI_PYTHON_ENV"]
encoder_identity = text_encoding.live_identity(HF_HOME, preset, kimodo_commit)
# The 8B encoder never lives here: misses go to a one-shot child (kimodo-text-encode.py), hits come from the store.
text_encoder = text_encoding.CachedTextEncoder(
text_encoding.open_store(os.path.join(os.environ.get("VILI_EMBEDDINGS_DIR", "/vili-embeddings"), "text-embeddings.cc")),
text_encoding.open_store(os.path.join(os.environ["VILI_EMBEDDINGS_DIR"], "text-embeddings.cc")),
encoder_identity,
text_encoding.encode_misses_in_child,
device=device,
@ -72,7 +63,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, "imageId": kimodo_image, "modelRevision": model_revision}
worker_identity = {"kimodoCommit": kimodo_commit, "pythonEnv": python_env, "modelRevision": model_revision, "threads": threads}
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
@ -272,10 +263,12 @@ def generate(payload):
requested_model = payload.get("model")
if requested_model and requested_model != resolved_model and requested_model != display_model:
raise ValueError(f"Worker model is {resolved_model}; request asked for {requested_model}.")
raise ValueError("The request names a model this worker does not serve.")
meta = payload.get("meta") if isinstance(payload.get("meta"), dict) else {}
output_base = str(payload.get("output") or "/outputs/motion")
if not payload.get("output"):
raise ValueError("The request carries no output path.")
output_base = str(payload["output"])
diffusion_steps = max(1, int(payload.get("diffusionSteps") or payload.get("diffusion_steps") or meta.get("diffusion_steps") or 20))
num_samples = max(1, int(payload.get("numSamples") or payload.get("num_samples") or meta.get("num_samples") or 1))
num_transition_frames = int(payload.get("numTransitionFrames") or payload.get("num_transition_frames") or 5)
@ -394,22 +387,24 @@ class Handler(BaseHTTPRequestHandler):
payload = json.loads(self.rfile.read(length).decode("utf-8") or "{}")
write_json(self, 200, generate(payload))
except Exception as error:
write_json(self, 500, {
"ok": False,
"error": str(error),
"traceback": traceback.format_exc(),
})
# An error names its class and where it was raised, never a message: messages can carry the request.
frame = traceback.extract_tb(error.__traceback__)[-1]
where = f"{os.path.basename(frame.filename)}:{frame.lineno}"
print(f"generate failed: {type(error).__name__} at {where}", flush=True)
write_json(self, 500, {"ok": False, "error": f"{type(error).__name__} at {where}"})
def log_message(self, fmt, *values):
print(f"{self.address_string()} - {fmt % values}", flush=True)
server = ThreadingHTTPServer((args.host, args.port), Handler)
print(json.dumps({
"event": "kimodo-worker-ready",
"device": device,
"model": resolved_model,
"displayModel": display_model,
"modelLoadSeconds": model_load_seconds,
"port": args.port,
"port": server.server_address[1],
"threads": threads,
}), flush=True)
ThreadingHTTPServer((args.host, args.port), Handler).serve_forever()
server.serve_forever()

View file

@ -1,12 +1,10 @@
#!/usr/bin/env python3
"""One-shot Kimodo text encoder: texts on stdin, vectors in --out, then exit so the OS reclaims RAM and VRAM.
"""One-shot Kimodo text encoder: texts on stdin, vectors in --out, then exit so the OS reclaims its RAM.
Reads {"texts": [...]} as JSON on stdin. Loads Kimodo's LLM2Vec preset in bfloat16 with the layers that do not
fit in free VRAM held in RAM, encodes each text alone as LLM2VecEncoder does, and writes a MessagePack answer.
There is no CPU fallback; --device cpu exists only for the reference probe.
Reads {"texts": [...]} as JSON on stdin. Loads the LLM2Vec encoder on CPU in bfloat16, wraps each text in the Llama-3 chat header, encodes it alone, and writes a MessagePack answer. The torch thread budget is VILI_THREADS.
"""
import argparse
import json
import os
import resource
import sys
import time
@ -16,100 +14,63 @@ import msgpack
import torch
sys.path.insert(0, str(Path(__file__).resolve().parent))
from text_encoding import CHILD_EXIT_GPU_BUSY, EMBEDDING_DIM, GpuBusy, plan_placement # noqa: E402
from text_encoding import ASSEMBLY, EMBEDDING_DIM, LLAMA_BASE, apply_thread_cap, chat_header, snapshot_path # noqa: E402
from peft import PeftModel # noqa: E402
from transformers import AutoConfig, AutoTokenizer # noqa: E402
from kimodo.model.llm2vec.llm2vec import LLM2Vec, batch_to_device # noqa: E402
from kimodo.model.llm2vec.llm2vec import LLM2Vec # noqa: E402
from kimodo.model.load_model import TEXT_ENCODER_PRESETS # noqa: E402
def load_encoder(base_id, peft_id, dtype, peft_kwargs, **model_kwargs):
"""Assemble the encoder in the order and with the ids of LLM2Vec.from_pretrained.
def load_encoder(hf_home, mntp_id, supervised_id, dtype):
"""The Llama base with the MNTP adapter merged in, then the supervised adapter on top (motion-pipeline:ruling:encoder-own-loader).
MUST TRACK Kimodo's loader: kimodo/model/llm2vec/llm2vec.py, LLM2Vec.from_pretrained, Kimodo commit 6bb5848
(nv-tlabs/kimodo). It is mirrored, not called, because that method hands PeftModel.from_pretrained no kwargs, and
with layers offloaded to RAM peft then takes a path that raises TypeError in accelerate's get_balanced_memory
(peft 0.18.1, accelerate 1.13.0, transformers 5.1.0). Here every PeftModel.from_pretrained gets peft_kwargs
(device_map='sequential'). Operator ruling motion-pipeline:ruling:encoder-own-loader. The ids' llm2vec_config.json
is not a local path, so Kimodo's loader applies no encoder config; neither does this.
"""
tokenizer = AutoTokenizer.from_pretrained(base_id)
Everything loads from the local snapshot directories, so the model's _name_or_path is a path and LLM2Vec never
adds the chat header itself; main() wraps every text with chat_header, so the prompt form is ours and no
side effect of the load decides it. The identity carries the Kimodo commit and this file's sha256."""
llama, mntp, supervised = (snapshot_path(hf_home, repo) for repo in (LLAMA_BASE, mntp_id, supervised_id))
tokenizer = AutoTokenizer.from_pretrained(llama)
tokenizer.pad_token = tokenizer.eos_token
tokenizer.padding_side = "left"
model_class = LLM2Vec._get_model_class(AutoConfig.from_pretrained(base_id).__class__.__name__, enable_bidirectional=True)
model = model_class.from_pretrained(base_id, torch_dtype=dtype, **model_kwargs)
if hasattr(model, "peft_config"):
model = PeftModel.from_pretrained(model, base_id, **peft_kwargs).merge_and_unload()
model = PeftModel.from_pretrained(model, peft_id, **peft_kwargs)
model_class = LLM2Vec._get_model_class(AutoConfig.from_pretrained(llama).__class__.__name__, enable_bidirectional=True)
model = model_class.from_pretrained(llama, torch_dtype=dtype)
model = PeftModel.from_pretrained(model, mntp).merge_and_unload()
model = PeftModel.from_pretrained(model, supervised)
return LLM2Vec(model=model, tokenizer=tokenizer)
def mem_available() -> int:
for line in Path("/proc/meminfo").read_text().splitlines():
if line.startswith("MemAvailable:"):
return int(line.split()[1]) * 1024
raise RuntimeError("MemAvailable is not in /proc/meminfo")
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--out", required=True)
parser.add_argument("--device", choices=["gpu", "cpu"], default="gpu")
args = parser.parse_args()
apply_thread_cap(os.environ)
texts = json.load(sys.stdin)["texts"]
out = Path(sys.argv[sys.argv.index("--out") + 1])
preset = TEXT_ENCODER_PRESETS["llm2vec"]["kwargs"]
load_kwargs, peft_kwargs = {}, {}
if args.device == "gpu":
free_vram, _ = torch.cuda.mem_get_info(0)
try:
max_memory = plan_placement(free_vram, mem_available())
load_kwargs = {"device_map": "auto", "max_memory": max_memory}
peft_kwargs = {"device_map": "sequential", "max_memory": max_memory}
except GpuBusy as busy:
Path(args.out).write_bytes(msgpack.packb({"error": busy.code, "freeVramBytes": busy.free_vram}, use_bin_type=True))
return CHILD_EXIT_GPU_BUSY
started = time.perf_counter()
model = load_encoder(
preset["base_model_name_or_path"],
preset["peft_model_name_or_path"],
getattr(torch, preset["dtype"]),
peft_kwargs,
**load_kwargs,
)
model = load_encoder(os.environ["HF_HOME"], preset["base_model_name_or_path"], preset["peft_model_name_or_path"], getattr(torch, preset["dtype"]))
model.eval()
for p in model.parameters():
p.requires_grad = False
load_seconds = time.perf_counter() - started
placement = {"gpuBytes": 0, "cpuBytes": 0}
for p in model.parameters():
placement["gpuBytes" if p.device.type == "cuda" else "cpuBytes"] += p.numel() * p.element_size()
input_device = "cuda:0" if args.device == "gpu" else "cpu"
vectors, encode_seconds = [], []
for text in texts:
began = time.perf_counter()
# The path LLM2Vec.encode takes for one text, without its .to(device) that would undo the device map.
features = model.tokenize([model.prepare_for_tokenization(model._convert_to_str("", text))])
features = batch_to_device(features, input_device)
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)
Path(args.out).write_bytes(msgpack.packb({
out.write_bytes(msgpack.packb({
"vectors": vectors,
"encodeSeconds": encode_seconds,
"loadSeconds": load_seconds,
"placement": placement,
"peakVramBytes": torch.cuda.max_memory_allocated(0) if torch.cuda.is_available() else 0,
"placement": {"cpuBytes": sum(p.numel() * p.element_size() for p in model.parameters())},
"peakRssBytes": resource.getrusage(resource.RUSAGE_SELF).ru_maxrss * 1024,
"device": args.device,
"device": "cpu",
"assembly": ASSEMBLY,
}, use_bin_type=True))
return 0

View file

@ -1,2 +0,0 @@
@echo off
powershell.exe -NoProfile -NonInteractive -ExecutionPolicy Bypass -File "%~dp0restart-vili-daemon.ps1" %*

View file

@ -1,38 +0,0 @@
param(
[int]$Port = 8824,
[string]$HostName = "0.0.0.0",
[string]$StateRoot = "E:\Projects\Vili\.vili"
)
$ErrorActionPreference = "Stop"
$ProgressPreference = "SilentlyContinue"
$PidPath = Join-Path $StateRoot "vili-daemon.pid"
$TaskName = "GameCult\Vili"
if (Test-Path $PidPath) {
$existingPid = (Get-Content $PidPath -ErrorAction SilentlyContinue | Select-Object -First 1)
if ($existingPid) {
$process = Get-Process -Id ([int]$existingPid) -ErrorAction SilentlyContinue
if ($process) {
Stop-Process -Id $process.Id -Force
Start-Sleep -Seconds 1
}
}
}
if (Get-ScheduledTask -TaskName "Vili" -TaskPath "\GameCult\" -ErrorAction SilentlyContinue) {
Stop-ScheduledTask -TaskName "Vili" -TaskPath "\GameCult\" -ErrorAction SilentlyContinue
Start-Sleep -Seconds 1
Start-ScheduledTask -TaskName "Vili" -TaskPath "\GameCult\"
Start-Sleep -Seconds 3
node (Join-Path (Split-Path -Parent $PSScriptRoot) "scripts\vili-daemon.mjs") --health --host 127.0.0.1 --port $Port --state-root $StateRoot
if ($LASTEXITCODE -ne 0) {
throw "Vili scheduled task started but health check failed."
}
Write-Host "Vili restarted through scheduled task $TaskName"
} else {
& (Join-Path $PSScriptRoot "start-vili-daemon.ps1") `
-Port $Port `
-HostName $HostName `
-StateRoot $StateRoot
}

View file

@ -1,20 +0,0 @@
Option Explicit
Dim shell
Dim command
Dim i
Dim exitCode
Set shell = CreateObject("WScript.Shell")
command = QuoteArg(shell.ExpandEnvironmentStrings("%SystemRoot%\System32\WindowsPowerShell\v1.0\powershell.exe"))
For i = 0 To WScript.Arguments.Count - 1
command = command & " " & QuoteArg(CStr(WScript.Arguments(i)))
Next
exitCode = shell.Run(command, 0, True)
WScript.Quit exitCode
Function QuoteArg(value)
QuoteArg = """" & Replace(value, """", """""") & """"
End Function

View file

@ -1,17 +0,0 @@
param(
[int]$Port = 8824,
[string]$HostName = "0.0.0.0",
[string]$StateRoot = "E:\Projects\Vili\.vili"
)
$ErrorActionPreference = "Stop"
$Root = Split-Path -Parent $PSScriptRoot
$Node = (Get-Command node -ErrorAction Stop).Source
$LogDir = Join-Path $StateRoot "logs"
$OutLog = Join-Path $LogDir "vili-daemon.out.log"
$ErrLog = Join-Path $LogDir "vili-daemon.err.log"
New-Item -ItemType Directory -Force -Path $LogDir | Out-Null
Set-Location $Root
& $Node (Join-Path $Root "scripts\vili-daemon.mjs") --host $HostName --port $Port --state-root $StateRoot 1>> $OutLog 2>> $ErrLog

View file

@ -1,36 +0,0 @@
#!/usr/bin/env node
const baseUrl = process.env.VILI_URL || process.argv[2] || "http://10.77.0.4:8824";
async function get(path) {
const response = await fetch(`${baseUrl}${path}`);
const text = await response.text();
let body;
try {
body = JSON.parse(text);
} catch {
body = text;
}
if (!response.ok) {
throw new Error(`${path} returned ${response.status}: ${text.slice(0, 500)}`);
}
return body;
}
const health = await get("/health");
const providers = await get("/eve/deck/providers");
const smoke = await get("/smoke");
console.log(JSON.stringify({
ok: true,
baseUrl,
health: {
service: health.service,
providerId: health.providerId,
backend: health.state?.backend,
},
providers,
smoke: {
ok: smoke.ok,
backend: smoke.backendStatus,
},
}, null, 2));

View file

@ -1,42 +0,0 @@
param(
[int]$Port = 8824,
[string]$HostName = "0.0.0.0",
[string]$StateRoot = "E:\Projects\Vili\.vili"
)
$ErrorActionPreference = "Stop"
$Root = Split-Path -Parent $PSScriptRoot
$Node = (Get-Command node -ErrorAction Stop).Source
$LogDir = Join-Path $StateRoot "logs"
$PidPath = Join-Path $StateRoot "vili-daemon.pid"
$OutLog = Join-Path $LogDir "vili-daemon.out.log"
$ErrLog = Join-Path $LogDir "vili-daemon.err.log"
New-Item -ItemType Directory -Force -Path $LogDir | Out-Null
if (Test-Path $PidPath) {
$existingPid = (Get-Content $PidPath -ErrorAction SilentlyContinue | Select-Object -First 1)
if ($existingPid -and (Get-Process -Id ([int]$existingPid) -ErrorAction SilentlyContinue)) {
& $Node (Join-Path $Root "scripts\vili-daemon.mjs") --health --host 127.0.0.1 --port $Port --state-root $StateRoot
if ($LASTEXITCODE -eq 0) {
Write-Host "Vili already healthy on port $Port with PID $existingPid"
exit 0
}
}
}
$args = @(
(Join-Path $Root "scripts\vili-daemon.mjs"),
"--host", $HostName,
"--port", "$Port",
"--state-root", $StateRoot
)
$process = Start-Process -FilePath $Node -ArgumentList $args -WorkingDirectory $Root -WindowStyle Hidden -RedirectStandardOutput $OutLog -RedirectStandardError $ErrLog -PassThru
Set-Content -Path $PidPath -Value $process.Id
Start-Sleep -Seconds 2
& $Node (Join-Path $Root "scripts\vili-daemon.mjs") --health --host 127.0.0.1 --port $Port --state-root $StateRoot
if ($LASTEXITCODE -ne 0) {
throw "Vili did not become healthy. See $ErrLog"
}
Write-Host "Vili started on $HostName`:$Port with PID $($process.Id)"

View file

@ -0,0 +1,685 @@
// node --test scripts/test-vili-daemon.mjs
// 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 http from "node:http";
import os from "node:os";
import path from "node:path";
import test from "node:test";
import { fileURLToPath } from "node:url";
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 THREADS = 5;
async function until(condition, what, seconds = 15) {
const deadline = Date.now() + seconds * 1000;
for (;;) {
const value = await condition();
if (value) return value;
assert.ok(Date.now() < deadline, `timed out waiting for ${what}`);
await new Promise((resolve) => setTimeout(resolve, 25));
}
}
// A zombie is dead for our purposes: its parent (or init) has simply not reaped it yet.
function isAlive(pid) {
try {
process.kill(pid, 0);
} catch {
return false;
}
try {
return !/^\d+ \(.*\) Z/.test(readFileSync(`/proc/${pid}/stat`, "utf8"));
} catch {
return false;
}
}
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,
} = {}) {
const root = givenRoot ?? mkdtempSync(path.join(os.tmpdir(), "vili-test-"));
const stateRoot = nested ? path.join(root, "a", "b", "state") : root;
const modelsRoot = path.join(root, "models");
const settings = {
host,
port: "0",
"state-root": stateRoot,
python,
"worker-entry": stubPath,
"models-root": relativeModels ? "models-rel" : modelsRoot,
"kimodo-commit": "abc1234",
"python-env": "test-env",
};
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" };
for (const name of Object.keys(env)) if (name.startsWith("VILI_")) delete env[name];
const argv = [daemonPath, ...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);
}
const proc = spawn(process.execPath, argv, { env, cwd: root, stdio: ["ignore", "pipe", "pipe"] });
let output = "";
let errors = "";
proc.stdout.on("data", (chunk) => { output += chunk; });
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];
const spawnLog = () => {
const log = path.join(stateRoot, "embeddings", "stub-spawns.log");
return existsSync(log) ? readFileSync(log, "utf8").split("\n").filter(Boolean) : [];
};
return {
stateRoot,
modelsRoot,
proc,
boundHost: listening[1],
port: Number(port),
stderr: () => errors,
spawns: () => spawnLog().filter((line) => /^\d+$/.test(line)).map(Number),
grandchildren: () => spawnLog().filter((line) => line.startsWith("grandchild ")).map((line) => Number(line.split(" ")[1])),
async stop() {
if (proc.exitCode === null && proc.signalCode === null) {
proc.kill("SIGTERM");
await until(() => proc.exitCode !== null || proc.signalCode !== null, "the daemon to exit", 30);
}
rmSync(root, { recursive: true, force: true });
},
};
}
function request(port, method, urlPath, { body, headers = {} } = {}) {
return new Promise((resolve, reject) => {
const req = http.request({ host: "127.0.0.1", port, method, path: urlPath, headers }, (res) => {
const chunks = [];
res.on("data", (chunk) => chunks.push(chunk));
res.on("end", () => {
const raw = Buffer.concat(chunks).toString("utf8");
let parsed = null;
try {
parsed = JSON.parse(raw);
} catch {
// not JSON
}
resolve({ status: res.statusCode, headers: res.headers, raw, body: parsed });
});
});
req.on("error", reject);
req.end(body);
});
}
const generate = (daemon, payload = { prompt: "a person waves", duration: 2 }, headers = {}) =>
request(daemon.port, "POST", "/motion/generate", {
body: typeof payload === "string" ? payload : JSON.stringify(payload),
headers: { "content-type": "application/json", ...headers },
});
const operatorState = async (daemon) => (await request(daemon.port, "GET", "/operator-state")).body;
const backendOf = async (daemon) => (await operatorState(daemon)).backend;
test("the worker's port comes from its ready line", async () => {
const daemon = await startDaemon();
try {
const answer = await generate(daemon);
assert.equal(answer.status, 200);
const backend = await backendOf(daemon);
assert.equal(backend.state, "ready");
assert.notEqual(daemon.port, 8824, "--port 0 binds a free port");
assert.ok(Number.isInteger(backend.port) && backend.port > 1 && backend.port !== 8825);
assert.equal(answer.body.worker.pid, backend.pid);
assert.deepEqual(answer.body.worker.argv.slice(-4), ["--host", "127.0.0.1", "--port", "0"]);
// the request reached the stub on the port the stub printed, not on a decoy line's port
assert.equal((await request(backend.port, "GET", "/health")).body.requests, 1);
} finally {
await daemon.stop();
}
});
test("one child, respawned after exit", async () => {
const daemon = await startDaemon();
try {
const [first, second] = await Promise.all([generate(daemon), generate(daemon)]);
assert.equal(first.status, 200);
assert.equal(second.status, 200);
assert.equal(daemon.spawns().length, 1);
const [pid] = daemon.spawns();
process.kill(pid, "SIGKILL");
await until(async () => (await backendOf(daemon)).state === "stopped", "the daemon to see the exit");
assert.match((await backendOf(daemon)).lastError, /^the worker exited \(SIGKILL\)$/);
const third = await generate(daemon);
assert.equal(third.status, 200);
const spawns = daemon.spawns();
assert.equal(spawns.length, 2);
assert.notEqual(spawns[1], pid);
assert.equal(third.body.worker.pid, spawns[1]);
assert.equal((await backendOf(daemon)).lastError, null);
} finally {
await daemon.stop();
}
});
test("environment and nice", async () => {
const daemon = await startDaemon();
try {
const { env, priority } = (await generate(daemon)).body.worker;
assert.equal(env.OMP_NUM_THREADS, String(THREADS));
assert.equal(env.MKL_NUM_THREADS, String(THREADS));
assert.equal(env.VILI_THREADS, String(THREADS));
assert.equal(env.HF_HUB_OFFLINE, "1");
assert.equal(env.TRANSFORMERS_OFFLINE, "1");
assert.equal(env.HF_HOME, daemon.modelsRoot);
assert.equal(env.VILI_KIMODO_COMMIT, "abc1234");
assert.equal(env.VILI_PYTHON_ENV, "test-env");
assert.equal(env.PYTHONDONTWRITEBYTECODE, "1");
assert.equal(env.PYTHONUNBUFFERED, "1");
assert.equal(env.PATH, process.env.PATH);
assert.equal(env.HOME, process.env.HOME);
assert.equal(env.LANG, "xx_TEST.UTF-8");
assert.ok(env.PYTHONPATH.endsWith(path.join("vendor", "CultLib", "packages", "cultcache-py", "src")));
assert.equal(env.LEAK_CHECK, undefined, "the daemon's own environment does not reach the worker");
assert.equal(priority, 10);
} finally {
await daemon.stop();
}
});
test("every flag has an environment twin, and the thread cap defaults to 8", async () => {
const viaEnv = await startDaemon({ viaEnv: true, threads: "16" });
try {
const answer = await generate(viaEnv);
assert.equal(answer.status, 200);
assert.equal(answer.body.worker.env.OMP_NUM_THREADS, "16");
assert.deepEqual(await backendOf(viaEnv), {
state: "ready", pid: answer.body.worker.pid, port: (await backendOf(viaEnv)).port, threads: 16, pythonEnv: "test-env",
kimodoCommit: "abc1234", modelsRoot: viaEnv.modelsRoot, lastError: null,
});
} finally {
await viaEnv.stop();
}
const byDefault = await startDaemon({ threads: null });
try {
const answer = await generate(byDefault);
assert.equal(answer.body.worker.env.OMP_NUM_THREADS, "8");
assert.equal((await backendOf(byDefault)).threads, 8);
} finally {
await byDefault.stop();
}
const one = await startDaemon({ threads: "1", nested: true });
try {
assert.equal((await generate(one)).body.worker.env.OMP_NUM_THREADS, "1");
} finally {
await one.stop();
}
});
test("the worker's output lands in its log, and the ready line does not", async () => {
const daemon = await startDaemon();
try {
assert.equal((await generate(daemon)).status, 200);
const log = path.join(daemon.stateRoot, "logs", "worker.log");
const text = await until(() => {
const content = existsSync(log) ? readFileSync(log, "utf8") : "";
return content.includes("stub-stdout-line") && content.includes("stub-stderr-line") ? content : null;
}, "the worker log");
assert.ok(!text.includes(`"port":${(await backendOf(daemon)).port}`), "the real ready line is not logged");
} finally {
await daemon.stop();
}
});
test("a job is a record on disk", async () => {
const daemon = await startDaemon();
try {
const answer = await generate(daemon, { prompt: "a person waves", duration: 2, constraints: [{ type: "root2d" }] });
assert.equal(answer.status, 200);
const job = answer.body;
assert.equal(job.schema, "gamecult.vili.motion_job.v0");
assert.equal(job.request.schema, "gamecult.vili.motion_request.v0");
assert.equal(job.prompt, "a person waves");
assert.equal(job.model, "stub-model");
assert.equal(job.ok, true);
assert.equal(job.error, null);
assert.equal(job.artifactDir, path.join(daemon.stateRoot, "artifacts", job.jobId));
assert.equal(job.worker.output, path.join(job.artifactDir, "motion"));
assert.equal(job.worker.contentType, "application/json");
assert.equal(job.worker.prompt, "a person waves");
assert.equal(job.worker.constraints, path.join(job.artifactDir, "constraints.json"));
assert.deepEqual(JSON.parse(readFileSync(job.worker.constraints, "utf8")), [{ type: "root2d" }]);
assert.deepEqual(JSON.parse(readFileSync(path.join(daemon.stateRoot, "jobs", `${job.jobId}.json`), "utf8")), job);
const batch = await generate(daemon, { texts: ["a", "b"], durations: [1, 1] });
assert.equal(batch.status, 200);
assert.equal(batch.body.prompt, null);
assert.equal(batch.body.worker.prompt, undefined);
} finally {
await daemon.stop();
}
});
test("a requested model restarts the worker only when it is not the one serving", async () => {
const daemon = await startDaemon();
try {
assert.equal((await generate(daemon)).status, 200);
assert.equal((await generate(daemon, { prompt: "x", model: "stub-model" })).status, 200);
assert.equal((await generate(daemon, { prompt: "x", model: "Stub Model" })).status, 200);
assert.equal(daemon.spawns().length, 1);
const [first] = daemon.spawns();
const other = await generate(daemon, { prompt: "x", model: "other-model" });
assert.equal(other.status, 200);
assert.equal(other.body.model, "other-model");
assert.deepEqual(other.body.worker.argv.slice(-2), ["--model", "other-model"]);
assert.equal(daemon.spawns().length, 2);
await until(() => !isAlive(first), "the first worker to be gone");
assert.equal((await generate(daemon)).status, 200);
assert.equal(daemon.spawns().length, 2, "a request that names no model keeps the one serving");
} finally {
await daemon.stop();
}
});
for (const signal of ["SIGTERM", "SIGINT"]) {
test(`${signal} leaves no child`, async () => {
const daemon = await startDaemon();
try {
assert.equal((await generate(daemon)).status, 200);
const [pid] = daemon.spawns();
assert.ok(isAlive(pid));
daemon.proc.kill(signal);
await until(() => daemon.proc.exitCode !== null, "the daemon to exit", 8);
assert.equal(isAlive(pid), false);
} finally {
await daemon.stop();
}
});
}
test("a worker that is stopping gets its grace, not a SIGKILL", async () => {
const daemon = await startDaemon();
try {
assert.equal((await generate(daemon, { prompt: "x", model: "graceful" })).status, 200);
daemon.proc.kill("SIGTERM");
await until(() => daemon.proc.exitCode !== null, "the daemon to exit", 8);
assert.ok(readFileSync(path.join(daemon.stateRoot, "embeddings", "stub-spawns.log"), "utf8").includes("graceful-exit"));
} finally {
await daemon.stop();
}
});
test("a worker that dies while the daemon waits for its health fails the job at once", async () => {
const daemon = await startDaemon();
try {
const began = Date.now();
const failed = await generate(daemon, { prompt: "x", model: "die-after-ready" });
assert.equal(failed.status, 502);
assert.ok(Date.now() - began < 20000, "the default deadline is minutes; a dead worker does not wait for it");
const backend = await backendOf(daemon);
assert.equal(backend.state, "stopped");
assert.equal(backend.lastError, "the worker exited (code 4)");
} finally {
await daemon.stop();
}
});
test("the daemon binds loopback unless told otherwise, and ignores flags it does not know", async () => {
const byDefault = await startDaemon({ omit: ["host"], extraArgs: ["--bogus"] });
try {
assert.equal(byDefault.boundHost, "127.0.0.1");
assert.equal((await request(byDefault.port, "GET", "/health")).status, 200);
} finally {
await byDefault.stop();
}
const named = await startDaemon({ host: "127.0.0.2" });
try {
assert.equal(named.boundHost, "127.0.0.2");
assert.equal((await fetch(`http://127.0.0.2:${named.port}/health`)).status, 200);
} finally {
await named.stop();
}
});
test("a relative models root reaches the worker as an absolute path", async () => {
const daemon = await startDaemon({ relativeModels: true });
try {
const { env } = (await generate(daemon)).body.worker;
assert.equal(env.HF_HOME, path.join(path.dirname(daemon.modelsRoot), "models-rel"));
} finally {
await daemon.stop();
}
});
test("a worker that ignores SIGTERM is killed with everything it started", async () => {
const daemon = await startDaemon();
try {
assert.equal((await generate(daemon, { prompt: "x", model: "stubborn" })).status, 200);
const [pid] = daemon.spawns();
const [grandchild] = daemon.grandchildren();
assert.ok(isAlive(pid) && isAlive(grandchild));
daemon.proc.kill("SIGTERM");
await until(() => daemon.proc.exitCode !== null, "the daemon to exit", 30);
assert.equal(isAlive(pid), false);
assert.equal(isAlive(grandchild), false);
} finally {
await daemon.stop();
}
});
// The daemon owns the worker's whole tree (motion-pipeline:finding:cut-vili-cpu-body.s1.orphaned-descendants).
const aliveWorkers = (daemon) => daemon.spawns().filter(isAlive);
test("a worker killed mid-request leaves no encoder child, and the next job has exactly one worker", async () => {
const daemon = await startDaemon();
try {
const pending = generate(daemon, { prompt: "x", model: "encoder" });
const [grandchild] = await until(() => { const found = daemon.grandchildren(); return found.length ? found : null; }, "the encoder child");
const [worker] = daemon.spawns();
assert.ok(isAlive(worker) && isAlive(grandchild));
process.kill(worker, "SIGKILL");
assert.equal(readFileSync(path.join(daemon.stateRoot, "worker.pid"), "utf8"), `${worker}\n`);
assert.equal((await pending).status, 502);
await until(() => !isAlive(grandchild), "the encoder child to die with its worker");
await until(() => !existsSync(path.join(daemon.stateRoot, "worker.pid")), "the dead worker's pid file to go");
assert.equal((await generate(daemon, { prompt: "y" })).status, 200);
assert.equal(aliveWorkers(daemon).length, 1);
assert.equal(readFileSync(path.join(daemon.stateRoot, "worker.pid"), "utf8"), `${aliveWorkers(daemon)[0]}\n`);
assert.equal(isAlive(grandchild), false);
} finally {
await daemon.stop();
}
});
// "encoder" leaves a SIGTERM-proof grandchild beside the worker; "stubborn" is a worker that ignores SIGTERM; the plain stub is just a worker.
for (const model of ["encoder", "stubborn", "stub-model"]) {
test(`a restarted daemon kills the ${model} worker its crashed predecessor left, and runs exactly one`, async () => {
const root = mkdtempSync(path.join(os.tmpdir(), "vili-test-"));
const first = await startDaemon({ root });
const stray = generate(first, { prompt: "x", model }).catch(() => null);
await until(() => first.spawns().length, "the worker");
if (model === "encoder") await until(() => first.grandchildren().length, "the encoder child");
await until(() => request(first.port, "GET", "/health").then((r) => r.body?.state?.backend?.state === "ready"), "the worker to be ready");
const [worker] = first.spawns();
const strays = [worker, ...first.grandchildren()];
first.proc.kill("SIGKILL");
await until(() => first.proc.exitCode !== null || first.proc.signalCode !== null, "the first daemon to die");
await stray;
assert.ok(strays.every(isAlive), "the probe must reproduce the orphan before the fix can be measured");
const second = await startDaemon({ root });
try {
await until(() => !strays.some(isAlive), "the orphans to be killed");
assert.equal(existsSync(path.join(root, "worker.pid")), false);
assert.equal((await generate(second, { prompt: "y" })).status, 200);
assert.equal(aliveWorkers(second).length, 1);
second.proc.kill("SIGTERM");
await until(() => second.proc.exitCode !== null, "the second daemon to exit", 30);
assert.equal(aliveWorkers(second).length, 0);
} finally {
await second.stop();
}
});
}
test("a live worker outside the pid file's group is left alone", async () => {
const root = mkdtempSync(path.join(os.tmpdir(), "vili-test-"));
const first = await startDaemon({ root });
const bystander = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { detached: true, stdio: "ignore" });
try {
assert.equal((await generate(first)).status, 200);
const [worker] = first.spawns();
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);
} finally {
bystander.kill("SIGKILL");
await first.stop();
}
});
test("a pid file naming a process group that is not the worker's kills nothing", async () => {
const root = mkdtempSync(path.join(os.tmpdir(), "vili-test-"));
const bystander = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { detached: true, stdio: "ignore" });
writeFileSync(path.join(root, "worker.pid"), `${bystander.pid}\n`);
const daemon = await startDaemon({ root });
try {
assert.equal(isAlive(bystander.pid), true);
assert.equal(existsSync(path.join(root, "worker.pid")), false);
} finally {
bystander.kill("SIGKILL");
await daemon.stop();
}
});
test("cross-site is refused before the worker", async () => {
const daemon = await startDaemon();
try {
assert.equal((await generate(daemon)).status, 200);
const withOrigin = await generate(daemon, undefined, { origin: "https://example.test" });
assert.equal(withOrigin.status, 403);
assert.deepEqual(withOrigin.body, { ok: false, code: "cross-site" });
const plain = await generate(daemon, undefined, { "content-type": "text/plain" });
assert.equal(plain.status, 403);
assert.deepEqual(plain.body, { ok: false, code: "cross-site" });
const untyped = await request(daemon.port, "POST", "/motion/generate", { body: JSON.stringify({ prompt: "x" }) });
assert.equal(untyped.status, 403);
const options = await request(daemon.port, "OPTIONS", "/motion/generate", { headers: { origin: "https://example.test" } });
assert.equal(options.status, 404);
const health = await request(daemon.port, "GET", "/health");
for (const reply of [withOrigin, plain, untyped, options, health]) {
assert.equal(Object.keys(reply.headers).filter((name) => name.startsWith("access-control-")).length, 0);
}
const backend = await backendOf(daemon);
assert.equal((await request(backend.port, "GET", "/health")).body.requests, 1, "only the same-site request reached the worker");
// the media type is matched as a type, not as a string
assert.equal((await generate(daemon, undefined, { "content-type": "Application/JSON; charset=utf-8" })).status, 200);
assert.equal((await generate(daemon, undefined, { "content-type": "application/json ; charset=utf-8" })).status, 200);
// and only the generate route answers a POST
const stray = await request(daemon.port, "POST", "/motion/other", {
body: JSON.stringify({ prompt: "x" }),
headers: { "content-type": "application/json" },
});
assert.equal(stray.status, 404);
} finally {
await daemon.stop();
}
});
for (const omitted of ["models-root", "kimodo-commit", "python-env"]) {
test(`unconfigured (no ${omitted}) refuses before spawning`, async () => {
const daemon = await startDaemon({ omit: [omitted] });
try {
const answer = await generate(daemon);
assert.equal(answer.status, 503);
assert.deepEqual(answer.body, { ok: false, code: "worker-unconfigured" });
assert.deepEqual(daemon.spawns(), []);
assert.equal((await backendOf(daemon)).state, "stopped");
} finally {
await daemon.stop();
}
});
}
test("an error never echoes the request", async () => {
const daemon = await startDaemon();
try {
const malformed = await generate(daemon, "{\"prompt\": \"the launch code is swordfish\", ");
assert.equal(malformed.status, 400);
assert.deepEqual(malformed.body, { ok: false, error: "Invalid JSON." });
const missing = await generate(daemon, { texts: ["the launch code is swordfish"] });
assert.equal(missing.status, 400);
assert.ok(!missing.raw.includes("swordfish"));
// a body the handler cannot use at all: the answer names nothing, and neither does the log
const unusable = await generate(daemon, "null");
assert.equal(unusable.status, 500);
assert.deepEqual(unusable.body, { ok: false, code: "internal" });
await until(() => daemon.stderr().includes("request failed: TypeError"), "the daemon to log the class");
assert.ok(!daemon.stderr().includes("swordfish"));
} finally {
await daemon.stop();
}
});
test("a worker that dies before it is ready fails the job, and the next job tries again", async () => {
const daemon = await startDaemon();
try {
const failed = await generate(daemon, { prompt: "x", model: "crash-before-ready" });
assert.equal(failed.status, 502);
assert.equal(failed.body.ok, false);
assert.equal(failed.body.error, "the worker request failed (Error)");
const backend = await backendOf(daemon);
assert.equal(backend.state, "failed");
assert.equal(backend.lastError, "the worker exited (code 3)");
assert.equal(backend.pid, null);
assert.equal(backend.port, null);
assert.equal((await generate(daemon)).status, 200);
assert.equal((await backendOf(daemon)).state, "ready");
assert.equal(daemon.spawns().length, 2);
} finally {
await daemon.stop();
}
});
test("a worker that never says it is ready is killed at the deadline", async () => {
const daemon = await startDaemon();
try {
const pending = generate(daemon, { prompt: "x", model: "hang-before-ready", workerTimeoutMs: 1500 });
const starting = await until(async () => {
const backend = await backendOf(daemon);
return backend.state === "starting" ? backend : null;
}, "the starting state");
assert.ok(Number.isInteger(starting.pid) && starting.port === null);
const failed = await pending;
assert.equal(failed.status, 502);
const backend = await backendOf(daemon);
assert.equal(backend.state, "failed");
assert.equal(backend.lastError, "the worker was not ready in time");
assert.equal(isAlive(daemon.spawns()[0]), false);
} finally {
await daemon.stop();
}
});
test("a worker that is ready but not healthy fails at the deadline; one that gets healthy in time serves", async () => {
const daemon = await startDaemon();
try {
const failed = await generate(daemon, { prompt: "x", model: "sick", workerTimeoutMs: 1500 });
assert.equal(failed.status, 502);
const backend = await backendOf(daemon);
assert.equal(backend.state, "failed");
assert.equal(backend.lastError, "the worker did not become healthy in time");
assert.equal(isAlive(daemon.spawns()[0]), false);
const slow = await generate(daemon, { prompt: "x", model: "slow-health" });
assert.equal(slow.status, 200);
assert.equal((await backendOf(daemon)).state, "ready");
} finally {
await daemon.stop();
}
});
test("a missing interpreter fails the job at once", async () => {
const daemon = await startDaemon({ python: "/no/such/python" });
try {
const began = Date.now();
const failed = await generate(daemon);
assert.equal(failed.status, 502);
assert.ok(Date.now() - began < 5000, "nothing waits on a process that never existed");
const backend = await backendOf(daemon);
assert.equal(backend.state, "failed");
assert.equal(backend.lastError, "the worker could not be spawned");
assert.equal(existsSync(path.join(daemon.stateRoot, "worker.pid")), false);
} finally {
await daemon.stop();
}
});
test("a worker that refuses, errors or drops the request fails the job without echoing it", async () => {
const daemon = await startDaemon();
try {
const refused = await generate(daemon, { prompt: "swordfish", model: "refuse-generate" });
assert.equal(refused.status, 502);
assert.equal(refused.body.ok, false);
assert.equal(refused.body.error, null);
const errored = await generate(daemon, { prompt: "swordfish", model: "fail-generate" });
assert.equal(errored.status, 502);
assert.equal(errored.body.error, "the worker answered 500");
const dropped = await generate(daemon, { prompt: "swordfish", model: "drop-generate" });
assert.equal(dropped.status, 502);
assert.equal(dropped.body.error, "the worker request failed (TypeError)");
const stalled = await generate(daemon, { prompt: "swordfish", model: "slow-generate", timeoutMs: 700 });
assert.equal(stalled.status, 502);
assert.equal(stalled.body.error, "the worker request failed (TimeoutError)");
for (const answer of [refused, errored, dropped, stalled]) assert.ok(!answer.body.error?.includes("swordfish"));
} finally {
await daemon.stop();
}
});
test("health, operator state and the unknown", async () => {
const daemon = await startDaemon();
try {
for (const route of ["/", "/health"]) {
const answer = await request(daemon.port, "GET", route);
assert.equal(answer.status, 200);
assert.equal(answer.body.ok, true);
assert.equal(answer.body.service, "vili");
assert.equal(answer.body.state.backend.threads, THREADS);
}
assert.equal((await request(daemon.port, "GET", "/motion/generate")).status, 404);
// the retired operator route, spelled in parts so the retirement grep over scripts/ stays empty
assert.equal((await request(daemon.port, "GET", ["", "eve", "operator"].join("/"))).status, 404);
const unknown = await request(daemon.port, "GET", "/provider-advertisement");
assert.equal(unknown.status, 404);
assert.equal(unknown.raw, "not found\n");
const state = await operatorState(daemon);
assert.equal(state.host, os.hostname());
assert.deepEqual(Object.keys(state.routes), ["health", "operatorState", "generate"]);
} finally {
await daemon.stop();
}
});
test("the typed service state is refreshed while the daemon runs", async () => {
const daemon = await startDaemon({ nested: true });
try {
const store = path.join(daemon.stateRoot, "vili.service.cc");
assert.ok(existsSync(store));
assert.ok(!readFileSync(store).includes("ready"));
assert.equal((await generate(daemon)).status, 200);
await until(() => readFileSync(store).includes("ready"), "the store to show the worker ready", 20);
} finally {
await daemon.stop();
}
});
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"] });
let output = "";
proc.stdout.on("data", (chunk) => { output += chunk; });
proc.stderr.on("data", (chunk) => { output += chunk; });
const code = await new Promise((resolve) => proc.on("exit", resolve));
assert.equal(code, 2);
assert.match(output, /threads must be an integer from 1 to 16/);
assert.ok(!output.replace("from 1 to 16", "").includes(threads), `the refusal repeated ${threads}`);
}
});

View file

@ -1,8 +1,13 @@
"""Run inside the Kimodo image: python -m unittest discover -s /opt/vili -p test_text_encoding.py"""
import hashlib
import os
import subprocess
import sys
import tempfile
import textwrap
import types
import unittest
from unittest import mock
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
@ -27,7 +32,7 @@ class FakeChild:
return {
"vectors": [vector(len(text)) for text in texts],
"encodeSeconds": [0.5] * len(texts),
"placement": {"gpuBytes": 11 * GIB, "cpuBytes": 5 * GIB},
"placement": {"cpuBytes": 5 * GIB},
}
@ -36,28 +41,6 @@ def open_encoder(directory, child, identity="id-a"):
return te.CachedTextEncoder(store, identity, child)
class PlanPlacement(unittest.TestCase):
def test_refuses_below_two_gib_with_a_typed_error_and_no_cpu_placement(self):
for free in (0, 1 * GIB, int(1.5 * GIB), 2 * GIB - 1):
with self.assertRaises(te.GpuBusy) as raised:
te.plan_placement(free, 64 * GIB)
self.assertEqual(raised.exception.code, "gpu-busy")
self.assertEqual(raised.exception.free_vram, free)
def test_splits_free_vram_and_ram_each_less_the_reserve(self):
self.assertEqual(te.plan_placement(12 * GIB, 7 * GIB), {0: 12 * GIB - te.GPU_RESERVE, "cpu": 7 * GIB - te.CPU_RESERVE})
self.assertEqual(te.plan_placement(14 * GIB, 10 * GIB), {0: 14 * GIB - te.GPU_RESERVE, "cpu": 10 * GIB - te.CPU_RESERVE})
def test_the_gpu_reserve_exceeds_the_old_one_that_crashed_at_the_card_ceiling(self):
self.assertGreater(te.GPU_RESERVE, int(1.5 * GIB))
def test_exactly_two_gib_free_is_admitted(self):
self.assertEqual(te.plan_placement(2 * GIB, 10 * GIB)[0], 2 * GIB - te.GPU_RESERVE)
def test_ram_below_the_reserve_never_goes_negative(self):
self.assertEqual(te.plan_placement(12 * GIB, GIB)["cpu"], 0)
class Identity(unittest.TestCase):
def test_key_changes_with_identity_and_text_and_is_stable(self):
self.assertEqual(te.embedding_key("a", "x"), te.embedding_key("a", "x"))
@ -65,19 +48,95 @@ class Identity(unittest.TestCase):
self.assertNotEqual(te.embedding_key("a", "x"), te.embedding_key("a", "y"))
self.assertNotEqual(te.embedding_key("ab", "x"), te.embedding_key("a", "bx"))
def test_identity_names_every_input(self):
base = dict(revisions={"r/one": "111", "r/two": "222"}, dtype="bfloat16", transformers_version="5.1.0", peft_version="0.18.1", kimodo_commit="6bb5848")
identity = te.encoder_identity(**base)
for field, other in (("dtype", "float32"), ("transformers_version", "5.2.0"), ("peft_version", "0.19.0"), ("kimodo_commit", "abcdef0"),
("revisions", {"r/one": "111", "r/two": "333"})):
self.assertNotEqual(identity, te.encoder_identity(**{**base, field: other}), field)
def test_key_is_a_full_sha256_over_the_whole_text(self):
key = te.embedding_key("id", "text a")
self.assertEqual(len(key), 64)
int(key, 16)
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_snapshot_revision_is_read_from_the_mounted_cache(self):
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|>")
self.assertEqual(te.chat_header(" hello"), header + "!@#$%^&*() hello<|eot_id|>")
form = hashlib.sha256(("|".join((header, "!@#$%^&*()", "<|eot_id|>"))).encode("utf-8")).hexdigest()
self.assertEqual(te.PROMPT_FORM, "llama3-chat-header:" + form)
def test_every_identity_field_changes_the_key(self):
# Built the way the worker builds it, from a mounted cache with 40-hex revisions, so the first 268 characters
# are the revisions and every other field sits past any prefix a truncated identity could keep.
preset = {"base_model_name_or_path": "McGill-NLP/LLM2Vec-Meta-Llama-3-8B-Instruct-mntp", "peft_model_name_or_path":
"McGill-NLP/LLM2Vec-Meta-Llama-3-8B-Instruct-mntp-supervised", "dtype": "bfloat16"}
versions = {"torch": "2.5.1+cpu", "transformers": "5.1.0", "peft": "0.18.1", "accelerate": "1.13.0"}
with tempfile.TemporaryDirectory() as home, tempfile.TemporaryDirectory() as bytes_dir:
loader = Path(bytes_dir) / "child.py"
loader.write_bytes(b"one")
repos = [preset["base_model_name_or_path"], preset["peft_model_name_or_path"], te.LLAMA_BASE]
revision = {repo: f"{index}" * 40 for index, repo in enumerate(repos, start=1)}
def write_revision(repo, value):
refs = Path(home) / "hub" / ("models--" + repo.replace("/", "--")) / "refs"
refs.mkdir(parents=True, exist_ok=True)
(refs / "main").write_text(value + chr(10))
for repo in repos:
write_revision(repo, revision[repo])
real_child, real_versions = te.CHILD_PATH, te.installed_versions
te.CHILD_PATH = loader
try:
def identity(preset=preset, commit="6bb5848", versions=versions):
te.installed_versions = lambda: versions
return te.live_identity(home, preset, commit)
base = identity()
self.assertGreater(base.index("assembly="), 200)
changed = {
"dtype": identity(preset={**preset, "dtype": "float32"}),
"kimodo": identity(commit="abcdef0"),
**{f"version:{name}": identity(versions={**versions, name: "9.9.9"}) for name in versions},
}
for repo in repos:
write_revision(repo, "f" * 40)
changed[f"revision:{repo}"] = identity()
write_revision(repo, revision[repo])
loader.write_bytes(b"two")
changed["loader"] = identity()
loader.write_bytes(b"one")
self.assertEqual(identity(), base)
finally:
te.CHILD_PATH, te.installed_versions = real_child, real_versions
self.assertEqual(len(changed), 10)
for field, other in changed.items():
self.assertNotEqual(te.embedding_key(base, "x"), te.embedding_key(other, "x"), field)
for part in (f"assembly={te.ASSEMBLY}", f"prompt={te.PROMPT_FORM}", "device=cpu", "dtype=bfloat16", "torch=2.5.1+cpu"):
self.assertIn(part, base)
def test_assembly_prompt_and_device_change_the_key(self):
fields = dict(
assembly=te.ASSEMBLY, prompt_form=te.PROMPT_FORM, device="cpu", revisions={"r/one": "1" * 40}, dtype="bfloat16",
torch_version="2.5.1+cpu", transformers_version="5.1.0", peft_version="0.18.1", accelerate_version="1.13.0",
kimodo_commit="6bb5848", loader_sha256="a" * 64,
)
key = te.embedding_key(te.encoder_identity(**fields), "x")
for name, other in (("assembly", "llm2vec-A"), ("prompt_form", "none"), ("device", "other-device")):
self.assertNotEqual(key, te.embedding_key(te.encoder_identity(**{**fields, name: other}), "x"), name)
def test_installed_versions_are_read_from_the_libraries_in_use(self):
fakes = {name: types.SimpleNamespace(__version__=f"{name}-v") for name in ("transformers", "peft", "accelerate")}
with mock.patch.dict(sys.modules, fakes), mock.patch.object(te.torch, "__version__", "torch-v"):
self.assertEqual(te.installed_versions(), {
"torch": "torch-v", "transformers": "transformers-v", "peft": "peft-v", "accelerate": "accelerate-v",
})
class SnapshotPath(unittest.TestCase):
def test_resolves_the_revision_the_mounted_cache_names(self):
with tempfile.TemporaryDirectory() as home:
refs = Path(home) / "hub" / "models--org--name" / "refs"
refs.mkdir(parents=True)
(refs / "main").write_text("deadbeef\n")
self.assertEqual(te.read_snapshot_revision(Path(home), "org/name"), "deadbeef")
repo = Path(home) / "hub" / "models--org--name"
(repo / "refs").mkdir(parents=True)
(repo / "refs" / "main").write_text("deadbeef" + chr(10))
(repo / "snapshots" / "deadbeef").mkdir(parents=True)
self.assertEqual(te.snapshot_path(Path(home), "org/name"), repo / "snapshots" / "deadbeef")
class CachedEncoder(unittest.TestCase):
@ -144,17 +203,6 @@ class CachedEncoder(unittest.TestCase):
self.assertEqual(tuple(features.shape), (1, 1, 4096))
self.assertEqual(encoder.store.get_all(te.TEXT_EMBEDDING)[0].dim, 4096)
def test_gpu_busy_propagates_without_a_cpu_path_and_stores_nothing(self):
def busy(texts):
raise te.GpuBusy(GIB)
with tempfile.TemporaryDirectory() as directory:
encoder = open_encoder(directory, busy)
with self.assertRaises(te.GpuBusy) as raised:
encoder([SECRET])
self.assertNotIn(SECRET, str(raised.exception))
self.assertEqual(encoder.count(), 0)
def test_the_contract_methods_return_self_or_the_device(self):
with tempfile.TemporaryDirectory() as directory:
encoder = open_encoder(directory, FakeChild())
@ -163,6 +211,40 @@ class CachedEncoder(unittest.TestCase):
self.assertEqual(encoder.get_device(), "cpu")
class ThreadCap(unittest.TestCase):
def test_accepts_one_to_sixteen(self):
for raw in ("1", "8", "16"):
self.assertEqual(te.thread_cap({"VILI_THREADS": raw}), int(raw))
def test_refuses_the_rest_without_echoing_the_value(self):
for raw in ("", "0", "17", "-1", "8.5", "eight", " 8", "\u0668"):
with self.assertRaises(ValueError) as raised:
te.thread_cap({"VILI_THREADS": raw})
self.assertNotIn(raw or "NUL", str(raised.exception))
with self.assertRaises(ValueError):
te.thread_cap({})
def test_apply_sets_torch_to_the_cap(self):
before = te.torch.get_num_threads()
try:
self.assertEqual(te.apply_thread_cap({"VILI_THREADS": "3"}), 3)
self.assertEqual(te.torch.get_num_threads(), 3)
finally:
te.torch.set_num_threads(before)
def test_the_child_applies_the_cap_before_it_loads_anything(self):
source = (Path(te.__file__).with_name("kimodo-text-encode.py")).read_text()
self.assertLess(source.index("apply_thread_cap(os.environ)"), source.index("load_encoder(os.environ"))
def test_the_worker_caps_interop_threads_before_it_loads_the_model(self):
source = (Path(te.__file__).with_name("kimodo-resident-worker.py")).read_text()
self.assertLess(source.index("torch.set_num_interop_threads(1)"), source.index("model, resolved_model = load_model("))
def test_the_worker_applies_the_cap_before_it_loads_the_model(self):
source = (Path(te.__file__).with_name("kimodo-resident-worker.py")).read_text()
self.assertLess(source.index("text_encoding.apply_thread_cap(os.environ)"), source.index("model, resolved_model = load_model("))
class ChildBoundary(unittest.TestCase):
def run_stub(self, body, texts=("aa",)):
with tempfile.TemporaryDirectory() as directory:
@ -170,16 +252,6 @@ class ChildBoundary(unittest.TestCase):
stub.write_text(textwrap.dedent(body))
return te.run_child(list(texts), Path(directory) / "out.mc", child_path=stub, timeout=60)
def test_exit_three_is_gpu_busy_with_the_measured_free_vram(self):
with self.assertRaises(te.GpuBusy) as raised:
self.run_stub("""
import sys, msgpack
out = sys.argv[sys.argv.index("--out") + 1]
open(out, "wb").write(msgpack.packb({"error": "gpu-busy", "freeVramBytes": 1234}, use_bin_type=True))
sys.exit(3)
""")
self.assertEqual(raised.exception.free_vram, 1234)
def test_other_failures_name_the_exit_code_and_never_the_text(self):
with self.assertRaises(RuntimeError) as raised:
self.run_stub("import sys; sys.exit(1)", texts=(SECRET,))
@ -197,5 +269,36 @@ class ChildBoundary(unittest.TestCase):
self.assertEqual(len(answer["vectors"]), 2)
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 = """
import sys, numpy as np
from kimodo.model.load_model import TEXT_ENCODER_PRESETS
from kimodo.model.llm2vec.llm2vec_wrapper import LLM2VecEncoder
kwargs = TEXT_ENCODER_PRESETS["llm2vec"]["kwargs"]
encoder = LLM2VecEncoder(**{**kwargs, "device": "cpu"})
vector, _ = encoder(sys.argv[2])
np.save(sys.argv[1], vector.float().numpy().reshape(-1))
"""
@unittest.skipUnless(REFERENCE_MODELS, "set VILI_REFERENCE_MODELS to a Hugging Face cache with the three repos (Yggdrasil)")
class ReferenceVector(unittest.TestCase):
"""The encoder child, in a fresh process, against Kimodo's own encoder loaded from the repo ids (which applies the chat header)."""
def test_the_child_vector_matches_kimodo_with_the_chat_header(self):
import numpy as np
env = {**os.environ, "HF_HOME": REFERENCE_MODELS, "HF_HUB_OFFLINE": "1", "VILI_THREADS": os.environ.get("VILI_THREADS", "8")}
with tempfile.TemporaryDirectory() as directory:
reference = Path(directory) / "reference.npy"
subprocess.run([sys.executable, "-c", REFERENCE_CODE, str(reference), ROCK], env=env, check=True)
os.environ.update(env)
ours = np.frombuffer(te.run_child([ROCK], Path(directory) / "answer.mp")["vectors"][0], dtype=np.float32)
expected = np.load(reference)
cosine = float(ours @ expected / (np.linalg.norm(ours) * np.linalg.norm(expected)))
print("cosine to Kimodo from repo ids:", cosine)
self.assertGreaterEqual(cosine, 0.9999)
if __name__ == "__main__":
unittest.main()

View file

@ -22,31 +22,34 @@ import torch
from cultcache_py import CultCache, SingleFileMessagePackBackingStore, define_database_entry_type, database_entry_field, leaves as t
GIB = 1024 ** 3
MIN_FREE_VRAM = 2 * GIB
GPU_RESERVE = int(3 * GIB) # headroom for activations and offloaded layers streamed through the card
CPU_RESERVE = int(1.5 * GIB)
EMBEDDING_DIM = 4096
EMBEDDING_BYTES = EMBEDDING_DIM * 4
CHILD_EXIT_GPU_BUSY = 3
CHILD_PATH = Path(__file__).with_name("kimodo-text-encode.py")
LLAMA_BASE = "meta-llama/Meta-Llama-3-8B-Instruct" # the base model the mntp adapter repo points at
LLAMA_BASE = "meta-llama/Meta-Llama-3-8B-Instruct" # the base model both LLM2Vec adapters are applied to
ASSEMBLY = "llm2vec-mntp-merged-then-supervised"
# Llama-3 Instruct's chat header, applied by Vili's own code (motion-pipeline:ruling:encoder-chat-header-on): the
# text goes in as a user turn, as LLM2Vec does for this model when it is loaded from the repo id. The separator is
# LLM2Vec's instruction/query marker: the model attends over the whole turn, and pooling covers what follows the
# marker (the text and the end-of-turn token), not the header, exactly as Kimodo's encoder pools.
CHAT_HEADER_PREFIX = "<|start_header_id|>user<|end_header_id|>\n\n"
CHAT_HEADER_SUFFIX = "<|eot_id|>"
QUERY_SEPARATOR = "!@#$%^&*()"
PROMPT_FORM = "llama3-chat-header:" + hashlib.sha256("|".join((CHAT_HEADER_PREFIX, QUERY_SEPARATOR, CHAT_HEADER_SUFFIX)).encode("utf-8")).hexdigest()
MAX_THREADS = 16
class GpuBusy(RuntimeError):
"""The card has too little free VRAM to host the encoder; there is no CPU fallback."""
code = "gpu-busy"
def __init__(self, free_vram: int) -> None:
self.free_vram = free_vram
super().__init__(f"gpu-busy: {free_vram / GIB:.2f} GiB of VRAM free, {MIN_FREE_VRAM / GIB:.0f} GiB needed")
def thread_cap(environ) -> int:
"""VILI_THREADS, the one torch thread budget of the worker and its encoder child: an integer from 1 to MAX_THREADS."""
raw = environ.get("VILI_THREADS", "")
if not raw.isascii() or not raw.isdigit() or not 1 <= int(raw) <= MAX_THREADS:
raise ValueError(f"VILI_THREADS must be an integer from 1 to {MAX_THREADS}")
return int(raw)
def plan_placement(free_vram: int, mem_available: int) -> dict:
"""The accelerate max_memory for one child call: free VRAM and available RAM, each less its reserve."""
if free_vram < MIN_FREE_VRAM:
raise GpuBusy(free_vram)
return {0: free_vram - GPU_RESERVE, "cpu": max(mem_available - CPU_RESERVE, 0)}
def apply_thread_cap(environ) -> int:
threads = thread_cap(environ)
torch.set_num_threads(threads)
return threads
def read_snapshot_revision(hf_home: Path, repo_id: str) -> str:
@ -55,13 +58,52 @@ def read_snapshot_revision(hf_home: Path, repo_id: str) -> str:
return ref.read_text().strip()
def encoder_identity(*, revisions: dict, dtype: str, transformers_version: str, peft_version: str, kimodo_commit: str) -> str:
"""The canonical string a vector is keyed under: every input that can change it."""
def snapshot_path(hf_home: Path, repo_id: str) -> Path:
"""The directory of the snapshot read_snapshot_revision names: assembly B loads from these, never from repo ids."""
return Path(hf_home) / "hub" / ("models--" + repo_id.replace("/", "--")) / "snapshots" / read_snapshot_revision(hf_home, repo_id)
def chat_header(text: str) -> str:
"""The text as the encoder reads it: a user turn in Llama-3 Instruct's chat format, pooled after the separator."""
return CHAT_HEADER_PREFIX + (QUERY_SEPARATOR + text).strip() + CHAT_HEADER_SUFFIX
def encoder_identity(*, assembly: str, prompt_form: str, device: str, revisions: dict, dtype: str, torch_version: str,
transformers_version: str, peft_version: str, accelerate_version: str, kimodo_commit: str,
loader_sha256: str) -> str:
"""The canonical string a vector is keyed under: every input that can change it, in a fixed field order.
Thread count is provenance, not identity; the encoder child's own bytes are, through loader_sha256."""
parts = [f"{repo}@{revision}" for repo, revision in sorted(revisions.items())]
parts += [f"dtype={dtype}", f"transformers={transformers_version}", f"peft={peft_version}", f"kimodo={kimodo_commit}"]
parts += [
f"assembly={assembly}", f"prompt={prompt_form}", f"device={device}", f"dtype={dtype}", f"torch={torch_version}", f"transformers={transformers_version}",
f"peft={peft_version}", f"accelerate={accelerate_version}", f"kimodo={kimodo_commit}", f"loader={loader_sha256}",
]
return ";".join(parts)
def installed_versions() -> dict:
"""The versions of the libraries that load and run the encoder, read from the installed packages."""
import accelerate
import peft
import transformers
return {"torch": torch.__version__, "transformers": transformers.__version__, "peft": peft.__version__, "accelerate": accelerate.__version__}
def live_identity(hf_home, preset: dict, kimodo_commit: str) -> str:
"""The identity the worker keys vectors under: the three snapshot revisions the mounted cache names, the preset's
dtype, the library versions, the Kimodo commit and the sha256 of the encoder child's own bytes."""
versions = installed_versions()
repos = (preset["base_model_name_or_path"], preset["peft_model_name_or_path"], LLAMA_BASE)
return encoder_identity(
assembly=ASSEMBLY, prompt_form=PROMPT_FORM, device="cpu",
revisions={repo: read_snapshot_revision(hf_home, repo) for repo in repos},
dtype=preset["dtype"], torch_version=versions["torch"], transformers_version=versions["transformers"],
peft_version=versions["peft"], accelerate_version=versions["accelerate"], kimodo_commit=kimodo_commit,
loader_sha256=hashlib.sha256(CHILD_PATH.read_bytes()).hexdigest(),
)
def embedding_key(identity: str, text: str) -> str:
return hashlib.sha256((identity + "\n" + text).encode("utf-8")).hexdigest()
@ -110,7 +152,7 @@ def open_store(path) -> CultCache:
def run_child(texts: list, out_path, child_path=CHILD_PATH, timeout: float = 1800.0) -> dict:
"""Run the one-shot encoder child for `texts` and return its answer file. Raises GpuBusy or RuntimeError."""
"""Run the one-shot encoder child for `texts` and return its answer file. Raises RuntimeError on any non-zero exit."""
process = subprocess.run(
[sys.executable, str(child_path), "--out", str(out_path)],
input=json.dumps({"texts": texts}).encode("utf-8"),
@ -118,8 +160,6 @@ def run_child(texts: list, out_path, child_path=CHILD_PATH, timeout: float = 180
timeout=timeout,
)
answer = msgpack.unpackb(Path(out_path).read_bytes(), raw=False) if Path(out_path).exists() else {}
if process.returncode == CHILD_EXIT_GPU_BUSY:
raise GpuBusy(int(answer.get("freeVramBytes", 0)))
if process.returncode != 0:
raise RuntimeError(f"text encoder child failed with exit code {process.returncode}")
return answer
@ -196,7 +236,7 @@ class CachedTextEncoder:
dim=EMBEDDING_DIM,
vector=vector,
vectorSha256=hashlib.sha256(vector).hexdigest(),
gpuBytes=int(placement.get("gpuBytes", 0)),
gpuBytes=0,
cpuBytes=int(placement.get("cpuBytes", 0)),
encodeSeconds=float(took),
createdAt=datetime.now(timezone.utc).isoformat(),

View file

@ -1,11 +1,13 @@
#!/usr/bin/env node
import http from "node:http";
import os from "node:os";
import { closeSync, openSync, readdirSync, readFileSync, unlinkSync, writeFileSync, writeSync } from "node:fs";
import { mkdir, writeFile } from "node:fs/promises";
import path from "node:path";
import readline from "node:readline";
import { fileURLToPath } from "node:url";
import { createRequire } from "node:module";
import { execFile } from "node:child_process";
import { WebSocketServer } from "ws";
import { spawn } from "node:child_process";
const __dirname = path.dirname(fileURLToPath(import.meta.url));
const require = createRequire(import.meta.url);
@ -16,37 +18,9 @@ const {
defineDocumentType,
} = require("cultcache-ts");
const projectRoot = path.resolve(__dirname, "..");
const ravenBaseUrl = process.env.VILI_PUBLIC_BASE_URL || "http://10.77.0.4:8824";
const ravenDeckUrl = process.env.VILI_PUBLIC_DECK_URL || "ws://10.77.0.4:8824/eve/deck";
const providerId = "vili.animation";
const viliStoreFileName = "vili.service.cc";
const kimodoWorkerName = process.env.VILI_KIMODO_WORKER_NAME || "vili-kimodo-worker";
const kimodoWorkerPort = Number(process.env.VILI_KIMODO_WORKER_PORT || 8825);
const kimodoWorkerUrl = `http://127.0.0.1:${kimodoWorkerPort}`;
const dockerHuggingFaceEnv = [
"export HF_TOKEN=\"$(cat /root/.cache/huggingface/token 2>/dev/null || true)\"",
"export HUGGING_FACE_HUB_TOKEN=\"$HF_TOKEN\"",
].join("; ");
const dockerHuggingFaceArgs = [
"-e HF_TOKEN",
"-e HUGGING_FACE_HUB_TOKEN",
"-v /root/.cache/huggingface:/root/.cache/huggingface",
].join(" ");
const objectSchema = { parse: (value) => value };
const providerAdvertisementDefinition = defineDocumentType({
type: "gamecult.eve.provider_advertisement",
schemaName: "gamecult.eve.provider_advertisement",
schemaId: "gamecult.eve.provider_advertisement.v1",
schemaVersion: "gamecult.eve.provider_advertisement.v1",
name: (value) => value?.providerId || "provider",
schema: objectSchema,
members: [
{ slot: 0, memberName: "providerId", typeName: "string", isName: true },
{ slot: 1, memberName: "title", typeName: "string" },
{ slot: 2, memberName: "verseId", typeName: "string" },
{ slot: 3, memberName: "updatedAt", typeName: "string" },
],
});
const operatorStateDefinition = defineDocumentType({
type: "gamecult.vili.operator_state",
schemaName: "gamecult.vili.operator_state",
@ -61,20 +35,6 @@ const operatorStateDefinition = defineDocumentType({
{ slot: 3, memberName: "backend", typeName: "object" },
],
});
const eveSurfaceDefinition = defineDocumentType({
type: "gamecult.eve.surface_state",
schemaName: "gamecult.eve.surface_state",
schemaId: "gamecult.eve.surface_state.v1",
schemaVersion: "gamecult.eve.surface_state.v1",
name: (value) => value?.providerId || "surface",
schema: objectSchema,
members: [
{ slot: 0, memberName: "providerId", typeName: "string", isName: true },
{ slot: 1, memberName: "title", typeName: "string" },
{ slot: 2, memberName: "updatedAt", typeName: "string" },
{ slot: 3, memberName: "surface", typeName: "object" },
],
});
const commandBoundaryDefinition = defineDocumentType({
type: "gamecult.vili.command_boundary",
schemaName: "gamecult.vili.command_boundary",
@ -106,20 +66,26 @@ const transportProfileDefinition = defineDocumentType({
],
});
const viliDocumentRegistry = defineDocumentRegistry(
providerAdvertisementDefinition,
operatorStateDefinition,
eveSurfaceDefinition,
commandBoundaryDefinition,
transportProfileDefinition,
);
const MAX_THREADS = 16;
function parseArgs(argv) {
const env = process.env;
const args = {
host: process.env.VILI_DAEMON_HOST || "127.0.0.1",
port: Number(process.env.VILI_DAEMON_PORT || 8824),
stateRoot: process.env.VILI_STATE_ROOT || path.join(projectRoot, ".vili"),
host: env.VILI_DAEMON_HOST || "127.0.0.1",
port: Number(env.VILI_DAEMON_PORT || 8824),
stateRoot: path.resolve(env.VILI_STATE_ROOT || path.join(projectRoot, ".vili")),
python: env.VILI_PYTHON || "python3",
workerEntry: path.resolve(env.VILI_WORKER_ENTRY || path.join(__dirname, "kimodo-resident-worker.py")),
modelsRoot: env.VILI_MODELS_ROOT || "",
threads: env.VILI_THREADS || "8",
kimodoCommit: env.VILI_KIMODO_COMMIT || "",
pythonEnv: env.VILI_PYTHON_ENV || "",
health: false,
printProviderAdvertisement: false,
};
for (let index = 0; index < argv.length; index += 1) {
@ -127,53 +93,29 @@ function parseArgs(argv) {
if (arg === "--host") args.host = argv[++index];
else if (arg === "--port") args.port = Number(argv[++index]);
else if (arg === "--state-root") args.stateRoot = path.resolve(argv[++index]);
else if (arg === "--python") args.python = argv[++index];
else if (arg === "--worker-entry") args.workerEntry = path.resolve(argv[++index]);
else if (arg === "--models-root") args.modelsRoot = argv[++index];
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;
else if (arg === "--print-provider-advertisement") args.printProviderAdvertisement = 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}`);
process.exit(2);
}
args.threads = Number(args.threads);
if (args.modelsRoot) args.modelsRoot = path.resolve(args.modelsRoot);
return args;
}
const args = parseArgs(process.argv.slice(2));
const startedAt = new Date().toISOString();
let backendStatus = {
checkedAt: null,
dockerImage: "unknown",
nvidiaRuntime: "unknown",
huggingFaceToken: "unknown",
kimodoHelp: "unknown",
kimodoWorker: "unknown",
kimodoWorkerDetail: null,
lastError: null,
};
function run(command, commandArgs, options = {}) {
return new Promise((resolve) => {
execFile(command, commandArgs, {
cwd: projectRoot,
timeout: options.timeout ?? 20000,
env: { ...process.env, ...(options.env || {}) },
windowsHide: true,
maxBuffer: 1024 * 1024 * 4,
}, (error, stdout, stderr) => {
resolve({
ok: !error,
code: error?.code ?? 0,
signal: error?.signal ?? null,
stdout: String(stdout || "").trim(),
stderr: String(stderr || "").trim(),
error: error ? String(error.message || error) : null,
});
});
});
}
function wsl(script, options = {}) {
return run("wsl.exe", ["-d", "Ubuntu-24.04", "-u", "root", "--", "bash", "-lc", script], options);
}
function shQuote(value) {
return `'${String(value).replaceAll("'", "'\\''")}'`;
}
// The worker child. Display state only: nothing reads `worker` to decide anything but ensureWorker.
const worker = { state: "stopped", proc: null, pid: null, port: null, servedModels: [], lastError: null };
let starting = null;
function firstDefined(...values) {
return values.find((value) => value !== undefined && value !== null);
@ -183,22 +125,11 @@ function boolValue(value) {
return value === true || String(value).toLowerCase() === "true" || String(value) === "1";
}
async function wslJson(script, options = {}) {
const result = await wsl(script, options);
if (!result.ok) return { ...result, body: null };
try {
return { ...result, body: JSON.parse(result.stdout) };
} catch (error) {
return { ...result, ok: false, body: null, error: `Invalid JSON: ${error.message}`, stderr: result.stderr };
}
}
function json(response, statusCode, body) {
const payload = JSON.stringify(body, null, 2);
response.writeHead(statusCode, {
"content-type": "application/json; charset=utf-8",
"cache-control": "no-store",
"access-control-allow-origin": "*",
});
response.end(payload);
}
@ -207,38 +138,31 @@ function text(response, statusCode, body) {
response.writeHead(statusCode, {
"content-type": "text/plain; charset=utf-8",
"cache-control": "no-store",
"access-control-allow-origin": "*",
});
response.end(body);
}
function providerAdvertisement() {
// A browser page can POST to localhost; a browser cannot send application/json cross-site without a preflight,
// and no route answers a preflight. So a POST that carries an Origin, or is not JSON, never reaches the body.
function refuseCrossSite(request, response) {
const contentType = String(request.headers["content-type"] || "").split(";")[0].trim().toLowerCase();
if (request.headers.origin !== undefined || contentType !== "application/json") {
json(response, 403, { ok: false, code: "cross-site" });
return true;
}
return false;
}
function backendStatus() {
return {
schema: "gamecult.eve.provider_advertisement.v1",
providerId,
title: "Vili Animation",
description: "Persona animation and Kimodo motion generation daemon on Raven.",
canonicalService: "asgard.vili",
locatedService: "asgard.raven.vili",
verseId: "raven.local",
cultMeshAddress: "asgard.raven.vili/eve/operator",
updatedAt: new Date().toISOString(),
endpoints: {
http: ravenBaseUrl,
health: `${ravenBaseUrl}/health`,
smoke: `${ravenBaseUrl}/smoke`,
providerAdvertisement: `${ravenBaseUrl}/provider-advertisement`,
operatorState: `${ravenBaseUrl}/operator-state`,
operatorSurface: `${ravenBaseUrl}/eve/operator`,
deckBroker: ravenDeckUrl,
providerDeck: `${ravenDeckUrl}/${providerId}`,
},
capabilities: [
"kimodo.motion.generate",
"persona.performance.intent",
"eve.operator.surface",
"cultmesh.provider.discovery",
],
state: worker.state,
pid: worker.pid,
port: worker.port,
threads: args.threads,
pythonEnv: args.pythonEnv || null,
kimodoCommit: args.kimodoCommit || null,
modelsRoot: args.modelsRoot || null,
lastError: worker.lastError,
};
}
@ -247,18 +171,15 @@ function operatorState() {
schema: "gamecult.vili.operator_state.v0",
service: "vili",
providerId,
verseId: "raven.local",
host: "raven",
host: os.hostname(),
startedAt,
updatedAt: new Date().toISOString(),
backend: backendStatus,
backend: backendStatus(),
authority: {
owns: [
"animation job intake",
"Kimodo runtime checks",
"Kimodo resident worker lifecycle",
"worker process lifecycle",
"motion artifact references",
"Vili provider and operator surfaces",
],
doesNotOwn: [
"Persona speech generation",
@ -269,11 +190,7 @@ function operatorState() {
},
routes: {
health: "/health",
providerAdvertisement: "/provider-advertisement",
operatorState: "/operator-state",
operatorSurface: "/eve/operator",
deckProviders: "/eve/deck/providers",
smoke: "/smoke",
generate: "/motion/generate",
},
};
@ -293,16 +210,11 @@ function commandBoundary() {
owns: [
"animation job identity",
"request persistence",
"Kimodo resident worker lifecycle",
"worker process lifecycle",
"timeout handling",
"motion artifact references",
],
},
smoke: {
method: "GET",
path: "/smoke",
owns: ["backend readiness proof", "resident worker warmup"],
},
liveness: {
method: "GET",
path: "/health",
@ -311,7 +223,7 @@ function commandBoundary() {
},
},
forbiddenAuthority:
"HTTP health, WebSocket decks, and JSON sidecars are compatibility/operator lowerings; they do not own daemon health or provider truth when CultCache/CultNet records exist.",
"HTTP health and the operator-state projection are compatibility lowerings; they do not own daemon health or provider truth when CultCache/CultNet records exist.",
};
}
@ -321,178 +233,226 @@ function transportProfile() {
profileId: "vili",
daemonId: "vili",
targetTransport: "cultnet.transport.rudp.v0",
currentTransport: "CultCache local state + compatibility HTTP/WS lowerings",
currentTransport: "CultCache local state + compatibility HTTP lowerings",
state: "local-cultcache-only",
providerAdvertisementSchema: "gamecult.eve.provider_advertisement.v1",
operatorStateSchema: "gamecult.vili.operator_state.v1",
commandBoundarySchema: "gamecult.vili.command_boundary.v1",
compatibilityMechanisms: [
`${ravenBaseUrl}/health`,
`${ravenBaseUrl}/provider-advertisement`,
`${ravenDeckUrl}/${providerId}`,
],
compatibilityMechanisms: ["/health", "/operator-state"],
cutLine:
"Vili owns typed provider, operator, command, and transport state. HTTP/WS/JSON projections are lowerings until Odin consumes Vili provider records over CultNet/RUDP.",
"Vili owns typed operator, command, and transport state. HTTP projections are lowerings until Odin consumes Vili records over CultNet/RUDP.",
updatedAt: new Date().toISOString(),
};
}
function eveSurface() {
const state = operatorState();
return {
schema: "gamecult.eve.surface.v1",
providerId,
title: "Vili Animation",
updatedAt: state.updatedAt,
surface: {
root: {
type: "panel",
title: "Vili Animation",
children: [
{ type: "stat", label: "Host", value: "Raven" },
{ type: "stat", label: "Verse", value: "raven.local" },
{ type: "stat", label: "Docker image", value: state.backend.dockerImage },
{ type: "stat", label: "NVIDIA runtime", value: state.backend.nvidiaRuntime },
{ type: "stat", label: "Kimodo", value: state.backend.kimodoHelp },
{ type: "stat", label: "Kimodo worker", value: state.backend.kimodoWorker },
{ type: "stat", label: "HF token", value: state.backend.huggingFaceToken },
{ type: "command", label: "Smoke", method: "GET", href: `${ravenBaseUrl}/smoke` },
{ type: "command", label: "Generate motion", method: "POST", href: `${ravenBaseUrl}/motion/generate` },
],
},
},
state,
};
}
async function persistSurfaces() {
await mkdir(args.stateRoot, { recursive: true });
await persistCultCacheSurfaces();
await writeFile(path.join(args.stateRoot, "vili-daemon.pid"), `${process.pid}\n`);
await writeFile(path.join(args.stateRoot, "provider-advertisement.json"), JSON.stringify(providerAdvertisement(), null, 2));
await writeFile(path.join(args.stateRoot, "operator-state.json"), JSON.stringify(operatorState(), null, 2));
await writeFile(path.join(args.stateRoot, "eve-operator-surface.json"), JSON.stringify(eveSurface(), null, 2));
await writeFile(path.join(args.stateRoot, "command-boundary.json"), JSON.stringify(commandBoundary(), null, 2));
await writeFile(path.join(args.stateRoot, "transport-profile.json"), JSON.stringify(transportProfile(), null, 2));
}
async function persistCultCacheSurfaces() {
const cache = CultCache.builder()
.withRegistry(viliDocumentRegistry)
.withGenericStore(new SingleFileMessagePackBackingStore(path.join(args.stateRoot, viliStoreFileName)))
.build();
await cache.put(providerAdvertisementDefinition, providerId, providerAdvertisement());
await cache.put(operatorStateDefinition, "vili", operatorState());
await cache.put(eveSurfaceDefinition, providerId, eveSurface());
await cache.put(commandBoundaryDefinition, "vili", commandBoundary());
await cache.put(transportProfileDefinition, "vili", transportProfile());
}
async function refreshBackendStatus({ includeKimodoHelp = false } = {}) {
const docker = await wsl([
"set -e",
"service docker start >/dev/null 2>&1 || true",
"docker images gamecult/kimodo:latest --format '{{.Repository}}:{{.Tag}}' | head -n 1",
].join("; "), { timeout: 20000 });
let nvidia = await wsl([
"set -e",
"service docker start >/dev/null 2>&1 || true",
"docker info --format '{{json .Runtimes}}' | grep -q nvidia && echo configured || echo missing",
].join("; "), { timeout: 20000 });
const token = await wsl("test -f /root/.cache/huggingface/token && echo present || echo missing", { timeout: 10000 });
const worker = await kimodoWorkerHealth(20000);
let help = { ok: true, stdout: backendStatus.kimodoHelp === "ok" ? "ok" : "not checked", stderr: "" };
if (includeKimodoHelp) {
nvidia = await wsl([
"set -e",
"service docker start >/dev/null 2>&1 || true",
"docker run --rm --runtime=nvidia --gpus all nvidia/cuda:12.4.1-base-ubuntu22.04 nvidia-smi --query-gpu=name --format=csv,noheader | head -n 1",
].join("; "), { timeout: 45000 });
help = await wsl([
"set -e",
"service docker start >/dev/null 2>&1 || true",
dockerHuggingFaceEnv,
`docker run --rm --runtime=nvidia --gpus all ${dockerHuggingFaceArgs} gamecult/kimodo:latest kimodo_gen --help | sed -n '1,6p'`,
].join("; "), { timeout: 60000 });
}
backendStatus = {
checkedAt: new Date().toISOString(),
dockerImage: docker.ok && docker.stdout ? docker.stdout : "missing",
nvidiaRuntime: nvidia.ok && nvidia.stdout ? nvidia.stdout : "unavailable",
huggingFaceToken: token.ok ? token.stdout : "unknown",
kimodoHelp: includeKimodoHelp ? (help.ok ? "ok" : "failed") : backendStatus.kimodoHelp,
kimodoWorker: worker.ok && worker.body?.ok ? "resident" : "not running",
kimodoWorkerDetail: worker.body || null,
lastError: [docker, nvidia, token, help].find((result) => !result.ok)?.stderr || null,
// The whole child environment, in one place. Nothing else from process.env reaches the worker.
function childEnv() {
const threads = String(args.threads);
const env = {
HF_HOME: args.modelsRoot,
HF_HUB_OFFLINE: "1",
TRANSFORMERS_OFFLINE: "1",
VILI_EMBEDDINGS_DIR: path.join(args.stateRoot, "embeddings"),
VILI_THREADS: threads,
OMP_NUM_THREADS: threads,
MKL_NUM_THREADS: threads,
PYTHONPATH: path.join(projectRoot, "vendor", "CultLib", "packages", "cultcache-py", "src"),
PYTHONDONTWRITEBYTECODE: "1",
PYTHONUNBUFFERED: "1",
VILI_KIMODO_COMMIT: args.kimodoCommit,
VILI_PYTHON_ENV: args.pythonEnv,
};
await persistSurfaces();
return { docker, nvidia, token, help, backendStatus };
for (const name of ["PATH", "HOME", "LANG"]) {
if (process.env[name] !== undefined) env[name] = process.env[name];
}
return env;
}
async function kimodoWorkerHealth(timeout = 10000) {
return wslJson(`curl -fsS --max-time ${Math.ceil(timeout / 1000)} ${shQuote(`${kimodoWorkerUrl}/health`)}`, { timeout });
function workerUnconfigured() {
return !args.modelsRoot || !args.kimodoCommit || !args.pythonEnv;
}
async function startKimodoWorker({ model = null } = {}) {
const script = [
"set -e",
"service docker start >/dev/null 2>&1 || true",
dockerHuggingFaceEnv,
"mkdir -p /mnt/e/Projects/Vili/.vili/artifacts /mnt/e/Projects/Vili/.vili/logs /mnt/e/Projects/Vili/.vili/embeddings",
"export VILI_KIMODO_COMMIT=\"$(git -C /opt/gamecult/kimodo rev-parse HEAD)\"",
"export VILI_KIMODO_IMAGE=\"$(docker image inspect --format '{{.Id}}' gamecult/kimodo:latest)\"",
"if ! kill -0 $(cat /tmp/vili-wsl-keepalive.pid 2>/dev/null) >/dev/null 2>&1; then nohup bash -c 'while true; do sleep 3600; done' >/tmp/vili-wsl-keepalive.log 2>&1 & echo $! >/tmp/vili-wsl-keepalive.pid; fi",
`if docker ps --format '{{.Names}}' | grep -qx ${shQuote(kimodoWorkerName)}; then echo running; exit 0; fi`,
`docker rm -f ${shQuote(kimodoWorkerName)} >/dev/null 2>&1 || true`,
[
"docker run -d",
`--name ${shQuote(kimodoWorkerName)}`,
"--runtime=nvidia --gpus all",
`-p 127.0.0.1:${kimodoWorkerPort}:${kimodoWorkerPort}`,
dockerHuggingFaceArgs,
"-e VILI_KIMODO_COMMIT -e VILI_KIMODO_IMAGE",
"-e PYTHONPATH=/opt/cultlib/cultcache-py",
"-v /mnt/e/Projects/Vili/scripts:/opt/vili:ro",
"-v /mnt/e/Projects/CultLib/packages/cultcache-py/src:/opt/cultlib/cultcache-py:ro",
"-v /mnt/e/Projects/Vili/.vili/embeddings:/vili-embeddings",
"-v /mnt/e/Projects/Vili/.vili/artifacts:/outputs",
"gamecult/kimodo:latest",
"python",
"/opt/vili/kimodo-resident-worker.py",
`--port ${kimodoWorkerPort}`,
model ? `--model ${shQuote(model)}` : "",
].join(" "),
].join("; ");
return wsl(script, { timeout: 30000 });
function waitForExit(proc, timeoutMs) {
return new Promise((resolve) => {
if (proc.exitCode !== null || proc.signalCode !== null) return resolve(true);
const timer = setTimeout(() => resolve(false), timeoutMs);
proc.once("exit", () => {
clearTimeout(timer);
resolve(true);
});
});
}
async function ensureKimodoWorker({ timeoutMs = 1000 * 60 * 8, model = null } = {}) {
const health = await kimodoWorkerHealth(10000);
if (health.ok && health.body?.ok) {
if (!model || health.body.model === model || health.body.displayModel === model) return health;
await wsl(`docker rm -f ${shQuote(kimodoWorkerName)} >/dev/null 2>&1 || true`, { timeout: 30000 });
// 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");
function killGroup(pgid, signal) {
try {
process.kill(-pgid, signal);
} catch {
// no such group
}
}
function forgetWorkerPid(pid) {
try {
if (readFileSync(workerPidFile(), "utf8").trim() === String(pid)) unlinkSync(workerPidFile());
} catch {
// no pid file, or another worker's
}
}
// 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
}
}
} catch {
// no /proc: nothing can be verified, so nothing is killed
}
if (ours) killGroup(pgid, "SIGKILL");
try {
unlinkSync(workerPidFile());
} catch {
// already gone
}
}
async function stopWorker() {
const proc = worker.proc;
if (!proc || proc.pid === undefined) return;
worker.state = "stopped";
for (const signal of ["SIGTERM", "SIGKILL"]) {
killGroup(proc.pid, signal);
if (await waitForExit(proc, signal === "SIGTERM" ? 10000 : 5000)) break;
}
}
async function startWorker(model, timeoutMs) {
await mkdir(path.join(args.stateRoot, "logs"), { recursive: true });
await mkdir(path.join(args.stateRoot, "embeddings"), { recursive: true });
const logFd = openSync(path.join(args.stateRoot, "logs", "worker.log"), "a");
const argv = [args.workerEntry, "--host", "127.0.0.1", "--port", "0"];
if (model) argv.push("--model", model);
const proc = spawn(args.python, argv, {
cwd: __dirname,
env: childEnv(),
stdio: ["ignore", "pipe", logFd],
shell: false,
detached: true,
});
closeSync(logFd);
if (proc.pid !== undefined) {
writeFileSync(workerPidFile(), `${proc.pid}\n`);
proc.once("exit", () => {
killGroup(proc.pid, "SIGKILL"); // whatever the worker left running, an encoder child included
forgetWorkerPid(proc.pid);
});
}
Object.assign(worker, { state: "starting", proc, pid: proc.pid ?? null, port: null, servedModels: [], lastError: null });
const exited = new Promise((resolve) => {
proc.once("error", () => resolve("the worker could not be spawned"));
proc.once("exit", (code, signal) => resolve(`the worker exited (${signal ?? `code ${code}`})`));
});
exited.then((reason) => {
if (worker.proc !== proc) return;
Object.assign(worker, { state: "stopped", proc: null, pid: null, port: null, lastError: worker.state === "stopped" ? worker.lastError : reason });
});
const fail = async (reason) => {
worker.lastError = reason;
await stopWorker();
Object.assign(worker, { state: "failed", proc: null, pid: null, port: null, lastError: reason });
throw new Error(reason);
};
if (proc.pid === undefined) await fail(await exited);
try {
os.setPriority(proc.pid, 10);
} catch {
await fail("the worker priority could not be lowered");
}
const started = await startKimodoWorker({ model });
if (!started.ok) return started;
// Stdout lines are the worker's log, until the one that says it is ready.
const deadline = Date.now() + timeoutMs;
let last = started;
while (Date.now() < deadline) {
await new Promise((resolve) => setTimeout(resolve, 5000));
last = await kimodoWorkerHealth(10000);
if (last.ok && last.body?.ok) return last;
let readyTimer;
const ready = await Promise.race([
new Promise((resolve) => {
const lines = readline.createInterface({ input: proc.stdout });
const sink = openSync(path.join(args.stateRoot, "logs", "worker.log"), "a");
lines.on("line", (line) => {
try {
const event = JSON.parse(line);
if (event?.event === "kimodo-worker-ready" && Number.isInteger(event.port)) return resolve(event);
} catch {
// not JSON: an ordinary log line
}
writeSync(sink, `${line}\n`);
});
lines.on("close", () => closeSync(sink));
}),
exited.then((reason) => ({ failed: reason })),
new Promise((resolve) => { readyTimer = setTimeout(() => resolve({ failed: "the worker was not ready in time" }), timeoutMs); }),
]);
clearTimeout(readyTimer);
if (ready.failed) await fail(ready.failed);
worker.port = ready.port;
for (;;) {
try {
const health = await fetch(`http://127.0.0.1:${worker.port}/health`, { signal: AbortSignal.timeout(10000) });
const body = await health.json();
if (health.ok && body?.ok) {
worker.servedModels = [body.model, body.displayModel].filter(Boolean);
worker.state = "ready";
return;
}
} catch {
// not answering yet
}
if (worker.proc !== proc) throw new Error(worker.lastError || "the worker exited");
if (Date.now() > deadline) await fail("the worker did not become healthy in time");
await new Promise((resolve) => setTimeout(resolve, 500));
}
return {
ok: false,
stdout: last.stdout || "",
stderr: last.stderr || "",
error: `Kimodo worker did not become healthy within ${timeoutMs}ms.`,
body: last.body || null,
};
}
async function ensureWorker({ model = null, timeoutMs = 1000 * 60 * 8 } = {}) {
if (starting) await starting;
if (worker.state === "ready" && model && !worker.servedModels.includes(model)) await stopWorker();
if (worker.state === "ready") return;
if (!starting) starting = startWorker(model, timeoutMs).finally(() => { starting = null; });
await starting;
}
async function readRequestBody(request) {
@ -503,11 +463,15 @@ async function readRequestBody(request) {
}
async function generateMotion(request, response) {
if (workerUnconfigured()) {
json(response, 503, { ok: false, code: "worker-unconfigured" });
return;
}
let body;
try {
body = await readRequestBody(request);
} catch (error) {
json(response, 400, { ok: false, error: `Invalid JSON: ${error.message}` });
} catch {
json(response, 400, { ok: false, error: "Invalid JSON." });
return;
}
@ -560,7 +524,7 @@ async function generateMotion(request, response) {
if (typeof body.constraints === "string") return body.constraints;
if (body.constraints_path) return body.constraints_path;
if (body.constraintsPath) return body.constraintsPath;
if (body.constraints !== undefined) return `/outputs/${jobId}/constraints.json`;
if (body.constraints !== undefined) return path.join(artifactDir, "constraints.json");
return undefined;
})();
if (body.constraints !== undefined && typeof body.constraints !== "string") {
@ -588,12 +552,13 @@ async function generateMotion(request, response) {
inputFolder: firstDefined(body.inputFolder, body.input_folder),
};
await writeFile(path.join(artifactDir, "meta.json"), JSON.stringify(meta, null, 2));
const worker = await ensureKimodoWorker({
timeoutMs: Number(firstDefined(body.workerTimeoutMs, body.worker_timeout_ms, 1000 * 60 * 8)),
model: body.model || null,
});
let result = worker;
if (worker.ok) {
let result;
try {
await ensureWorker({
timeoutMs: Number(firstDefined(body.workerTimeoutMs, body.worker_timeout_ms, 1000 * 60 * 8)),
model: body.model || null,
});
const workerRequest = {
model: body.model,
prompt: prompt || undefined,
@ -613,20 +578,19 @@ async function generateMotion(request, response) {
bvhStandardTpose: boolValue(firstDefined(body.bvhStandardTpose, body.bvh_standard_tpose)),
noPostprocess: boolValue(firstDefined(body.noPostprocess, body.no_postprocess)),
saveExampleDir: boolValue(firstDefined(body.saveExampleDir, body.save_example_dir)),
output: `/outputs/${jobId}/motion`,
output: path.join(artifactDir, "motion"),
};
await writeFile(path.join(artifactDir, "request.json"), JSON.stringify(workerRequest, null, 2));
const curlCommand = [
`curl -sS --max-time ${Math.ceil(Number(body.timeoutMs || 1000 * 60 * 30) / 1000)}`,
"-H 'content-type: application/json'",
`--data-binary @${shQuote(`/mnt/e/Projects/Vili/.vili/artifacts/${jobId}/request.json`)}`,
shQuote(`${kimodoWorkerUrl}/generate`),
].join(" ");
result = await wslJson([
"set -e",
curlCommand,
].join("; "), { timeout: Number(body.timeoutMs || 1000 * 60 * 30) });
if (result.body?.ok === false) result = { ...result, ok: false };
const answer = await fetch(`http://127.0.0.1:${worker.port}/generate`, {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify(workerRequest),
signal: AbortSignal.timeout(Number(body.timeoutMs || 1000 * 60 * 30)),
});
const workerBody = await answer.json();
result = { ok: answer.ok && workerBody?.ok !== false, error: answer.ok ? null : `the worker answered ${answer.status}`, body: workerBody };
} catch (error) {
result = { ok: false, error: `the worker request failed (${error?.name || "Error"})`, body: null };
}
const job = {
@ -641,54 +605,21 @@ async function generateMotion(request, response) {
seed,
artifactDir,
ok: result.ok,
stdout: result.stdout,
stderr: result.stderr,
error: result.error,
worker: result.body,
};
await writeFile(path.join(args.stateRoot, "jobs", `${jobId}.json`), JSON.stringify(job, null, 2)).catch(async () => {
await mkdir(path.join(args.stateRoot, "jobs"), { recursive: true });
await writeFile(path.join(args.stateRoot, "jobs", `${jobId}.json`), JSON.stringify(job, null, 2));
});
await writeFile(path.join(args.stateRoot, "jobs", `${jobId}.json`), JSON.stringify(job, null, 2));
json(response, result.ok ? 200 : 502, job);
}
async function handleRequest(request, response) {
const url = new URL(request.url, `http://${request.headers.host || `${args.host}:${args.port}`}`);
if (request.method === "OPTIONS") {
response.writeHead(204, {
"access-control-allow-origin": "*",
"access-control-allow-methods": "GET,POST,OPTIONS",
"access-control-allow-headers": "content-type",
});
response.end();
return;
}
if (request.method === "POST" && refuseCrossSite(request, response)) return;
if (url.pathname === "/" || url.pathname === "/health") {
await refreshBackendStatus({ includeKimodoHelp: false });
json(response, 200, { ok: true, service: "vili", providerId, state: operatorState() });
} else if (url.pathname === "/provider-advertisement") {
json(response, 200, providerAdvertisement());
} else if (url.pathname === "/operator-state") {
json(response, 200, operatorState());
} else if (url.pathname === "/eve/operator") {
json(response, 200, eveSurface());
} else if (url.pathname === "/eve/deck/providers") {
json(response, 200, {
providers: [{
id: providerId,
title: "Vili Animation",
description: "Persona animation and Kimodo motion generation daemon.",
endpoint: `${ravenDeckUrl}/${providerId}`,
}],
});
} else if (url.pathname === "/smoke") {
const result = await refreshBackendStatus({ includeKimodoHelp: true });
const worker = await ensureKimodoWorker();
await refreshBackendStatus({ includeKimodoHelp: false });
const ok = result.backendStatus.kimodoHelp === "ok" && worker.ok && worker.body?.ok;
json(response, ok ? 200 : 502, { ok, ...result, worker: worker.body || worker });
} else if (url.pathname === "/motion/generate" && request.method === "POST") {
await generateMotion(request, response);
} else {
@ -704,11 +635,6 @@ async function healthCheck() {
if (!body.ok) throw new Error("health returned ok=false");
}
if (args.printProviderAdvertisement) {
console.log(JSON.stringify(providerAdvertisement(), null, 2));
process.exit(0);
}
if (args.health) {
healthCheck().then(() => process.exit(0), (error) => {
console.error(error.message);
@ -720,32 +646,26 @@ if (args.health) {
const server = http.createServer((request, response) => {
handleRequest(request, response).catch((error) => {
json(response, 500, { ok: false, error: error.message, stack: process.env.NODE_ENV === "development" ? error.stack : undefined });
// The error's message can carry the request; name only its class.
console.error(`request failed: ${error?.name || "Error"}`);
if (!response.headersSent) json(response, 500, { ok: false, code: "internal" });
else response.end();
});
});
const wss = new WebSocketServer({ noServer: true });
wss.on("connection", (socket) => {
const sendSurface = () => {
if (socket.readyState === socket.OPEN) socket.send(JSON.stringify(eveSurface()));
};
sendSurface();
const timer = setInterval(sendSurface, 5000);
socket.on("close", () => clearInterval(timer));
});
server.on("upgrade", (request, socket, head) => {
const url = new URL(request.url, `http://${request.headers.host || `${args.host}:${args.port}`}`);
if (url.pathname === `/eve/deck/${providerId}` || url.pathname === "/eve/deck") {
wss.handleUpgrade(request, socket, head, (ws) => wss.emit("connection", ws, request));
} else {
socket.destroy();
}
});
server.listen(args.port, args.host, () => {
console.log(`Vili listening on http://${args.host}:${args.port}`);
reclaimStrayWorker(); // synchronous: no request is served before a predecessor's worker is gone
console.log(`Vili listening on http://${args.host}:${server.address().port}`);
});
setInterval(() => persistSurfaces().catch((error) => console.error(error)), 5000);
const surfaceTimer = setInterval(() => persistSurfaces().catch((error) => console.error(`persist failed: ${error?.name || "Error"}`)), 5000);
const shutdown = async () => {
clearInterval(surfaceTimer);
await stopWorker();
server.close();
process.exit(0);
};
process.on("SIGTERM", shutdown);
process.on("SIGINT", shutdown);
}

1
vendor/CultLib vendored Submodule

@ -0,0 +1 @@
Subproject commit b549d73183aa36c6ebb8163ebc68cad60b09c85d