From 4a7b8af34083e21ea8f89fa9efadaeda294993b8 Mon Sep 17 00:00:00 2001 From: Streaky Date: Wed, 5 Aug 2026 04:44:24 +0100 Subject: [PATCH 1/6] Implement Phase 4 minimal server --- AGENTS.md | 13 +- Cargo.lock | 492 +++++++++++++ Cargo.toml | 9 +- README.md | 29 +- crates/cli/Cargo.toml | 2 + crates/cli/src/main.rs | 98 ++- crates/executor-sys/src/lib.rs | 7 + crates/executor/src/lib.rs | 26 + crates/server/Cargo.toml | 20 + crates/server/src/lib.rs | 1224 +++++++++++++++++++++++++++++++ native/include/cusco_executor.h | 7 +- native/shim/cusco_executor.cpp | 46 ++ 12 files changed, 1960 insertions(+), 13 deletions(-) create mode 100644 crates/server/Cargo.toml create mode 100644 crates/server/src/lib.rs diff --git a/AGENTS.md b/AGENTS.md index dcc81b6..f781429 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -10,21 +10,22 @@ Keep this `AGENTS.md` up to date whenever development workflows, architecture, s ## Current state -Phases 1 through 3 are implemented. The real Gemma executor proof demonstrated exact token and bitwise-logit continuation after checkpoint capture, slot displacement, restoration, and a device-to-host-to-device round trip across four logical contexts. Cancellation and failed promotion preserve the prior binding. The Rust logical context store owns persistent token-sequence branches, model and adapter epochs, dependency-valid evaluated-prefix mappings, longest-valid-prefix lookup, reference accounting, and revision-guarded transactional publication. The physical manager owns capacity-accounted device, host, and storage copies, guarded reservations, transfers, active bindings, prepared transitions, deterministic eviction, and capacity observability. +Phases 1 through 4 are implemented. The real Gemma executor proof demonstrated exact checkpoint continuation. The logical context store owns shared token branches and transactional mappings. The physical manager owns tier capacity, representations, transfers, bindings, prepared transitions, eviction, and observability. The minimal server adds durable opaque external context IDs, atomic state recovery, model lifecycle APIs, bounded transition-cost scheduling, shared streaming events, canonical usage, cancellation/deadlines, authentication policy, OpenAI completion/chat adapters, and checked OpenAPI. -There is no production server yet. The repository currently contains: +The repository currently contains: - `crates/context-store`: Rust logical contexts, structurally shared token sequences, and evaluated-prefix mappings; - `crates/physical-manager`: physical representations, tier capacity, ownership references, transfers, bindings, and transactional transitions; - `crates/executor-sys`: native FFI declarations and linking; -- `crates/executor`: safe Rust executor and checkpoint wrappers; +- `crates/executor`: safe Rust executor, token-piece, and checkpoint wrappers; - `crates/model-registry`: immutable Hugging Face resolution and verified local registration; -- `crates/cli`: the proof and model-management driver; +- `crates/server`: protocol-neutral inference/model services, durable catalog, scheduler, auth, HTTP adapters, and OpenAPI; +- `crates/cli`: proof, model-management, and `serve` commands; - `native`: the C ABI header and llama.cpp shim; - `executor`: upstream and patch metadata; -- `tools`: fetch, verification, and coverage helpers. +- `tools`: fetch, verification, integration-report, and coverage helpers. -Do not represent later phases as implemented. The next architectural work is the minimal server and scheduler. +The Phase 4 server is minimal: it loads a registered llama.cpp model for inference and has not yet integrated live executor slots with the Phase 3 physical manager. Do not represent mapped execution, production hardening, or later phases as implemented. The next architectural work is mapped execution if staged measurements justify it, followed by compatibility and production hardening. ## Build and dependency conventions diff --git a/Cargo.lock b/Cargo.lock index 0d7c4df..f3642a5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -64,12 +64,70 @@ version = "1.0.104" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" +[[package]] +name = "atomic-waker" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" + [[package]] name = "autocfg" version = "1.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" +[[package]] +name = "axum" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" +dependencies = [ + "axum-core", + "bytes", + "form_urlencoded", + "futures-util", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-util", + "itoa", + "matchit", + "memchr", + "mime", + "percent-encoding", + "pin-project-lite", + "serde_core", + "serde_json", + "serde_path_to_error", + "serde_urlencoded", + "sync_wrapper", + "tokio", + "tower", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "axum-core" +version = "0.5.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "http-body-util", + "mime", + "pin-project-lite", + "sync_wrapper", + "tower-layer", + "tower-service", + "tracing", +] + [[package]] name = "base64" version = "0.22.1" @@ -106,6 +164,12 @@ dependencies = [ "generic-array", ] +[[package]] +name = "bumpalo" +version = "3.20.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" + [[package]] name = "bytes" version = "1.12.1" @@ -210,7 +274,9 @@ dependencies = [ "clap", "cusco-executor", "cusco-model-registry", + "cusco-server", "serde_json", + "tokio", ] [[package]] @@ -256,6 +322,23 @@ dependencies = [ "thiserror", ] +[[package]] +name = "cusco-server" +version = "0.1.0" +dependencies = [ + "axum", + "cusco-model-registry", + "futures-util", + "parking_lot", + "serde", + "serde_json", + "sha2", + "thiserror", + "tokio", + "tower", + "uuid", +] + [[package]] name = "digest" version = "0.10.7" @@ -304,6 +387,60 @@ version = "1.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" +[[package]] +name = "form_urlencoded" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb4cb245038516f5f85277875cdaa4f7d2c9a0fa0468de06ed190163b1581fcf" +dependencies = [ + "percent-encoding", +] + +[[package]] +name = "futures-channel" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "262590f4fe6afeb0bc83be1daa64e52657fe185690a958af7f3ad0e92085c5ae" +dependencies = [ + "futures-core", +] + +[[package]] +name = "futures-core" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2cd50c473c80f6d7c3670a752354b8e569b1a7cbfdc0419ec88e5edad85e0dc7" + +[[package]] +name = "futures-macro" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2d6d3cde68c518367be28956066ddfef33813991b77a55005a69dae04bf3b10b" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "futures-task" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b231ed28831efb4a61a08580c4bc233ec56bc009f4cd8f52da2c3cb97df0c109" + +[[package]] +name = "futures-util" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a77a90a256fce34da66415271e30f94ee91c57b04b8a2c042d9cf3220179deaa" +dependencies = [ + "futures-core", + "futures-macro", + "futures-task", + "pin-project-lite", + "slab", +] + [[package]] name = "generic-array" version = "0.14.7" @@ -370,12 +507,76 @@ dependencies = [ "itoa", ] +[[package]] +name = "http-body" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ca2a8f2913ee65f60facd6a5905613afaa448497a0230cc41ce022d93290bc2c" +dependencies = [ + "bytes", + "http", +] + +[[package]] +name = "http-body-util" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e9f41fd6a08e4d4ec69df65976da761afd5ad5e58a9d4acb46bd1c953a9e3ff2" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "pin-project-lite", +] + [[package]] name = "httparse" version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + +[[package]] +name = "hyper" +version = "1.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d22053281f852e11534f5198498373cbb59295120a20771d90f7ed1897490a72" +dependencies = [ + "atomic-waker", + "bytes", + "futures-channel", + "futures-core", + "http", + "http-body", + "httparse", + "httpdate", + "itoa", + "pin-project-lite", + "smallvec", + "tokio", +] + +[[package]] +name = "hyper-util" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" +dependencies = [ + "bytes", + "http", + "http-body", + "hyper", + "pin-project-lite", + "tokio", + "tower-service", +] + [[package]] name = "is_terminal_polyfill" version = "1.70.2" @@ -388,6 +589,17 @@ version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" +[[package]] +name = "js-sys" +version = "0.3.103" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53b44bfcdb3f8d5837a46dae1ca9660a837176eee74a28b229bc626816589102" +dependencies = [ + "cfg-if", + "futures-util", + "wasm-bindgen", +] + [[package]] name = "libc" version = "0.2.189" @@ -400,18 +612,39 @@ version = "0.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" +[[package]] +name = "lock_api" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965" +dependencies = [ + "scopeguard", +] + [[package]] name = "log" version = "0.4.33" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" +[[package]] +name = "matchit" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" + [[package]] name = "memchr" version = "2.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" +[[package]] +name = "mime" +version = "0.3.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" + [[package]] name = "miniz_oxide" version = "0.8.9" @@ -422,6 +655,17 @@ dependencies = [ "simd-adler32", ] +[[package]] +name = "mio" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "30d65c71f1ce40ab09135ce117d742b9f8a19ff91a41a8b57ed50bc2de59c427" +dependencies = [ + "libc", + "wasi", + "windows-sys 0.61.2", +] + [[package]] name = "num-traits" version = "0.2.19" @@ -443,12 +687,41 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" +[[package]] +name = "parking_lot" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a" +dependencies = [ + "lock_api", + "parking_lot_core", +] + +[[package]] +name = "parking_lot_core" +version = "0.9.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" +dependencies = [ + "cfg-if", + "libc", + "redox_syscall", + "smallvec", + "windows-link", +] + [[package]] name = "percent-encoding" version = "2.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + [[package]] name = "ppv-lite86" version = "0.2.21" @@ -551,6 +824,15 @@ dependencies = [ "rand_core", ] +[[package]] +name = "redox_syscall" +version = "0.5.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" +dependencies = [ + "bitflags", +] + [[package]] name = "regex-syntax" version = "0.8.11" @@ -619,6 +901,12 @@ dependencies = [ "untrusted", ] +[[package]] +name = "rustversion" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" + [[package]] name = "rusty-fork" version = "0.3.1" @@ -631,6 +919,18 @@ dependencies = [ "wait-timeout", ] +[[package]] +name = "ryu" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" + +[[package]] +name = "scopeguard" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" + [[package]] name = "serde" version = "1.0.229" @@ -674,6 +974,29 @@ dependencies = [ "zmij", ] +[[package]] +name = "serde_path_to_error" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "10a9ff822e371bb5403e391ecd83e182e0e77ba7f6fe0160b795797109d1b457" +dependencies = [ + "itoa", + "serde", + "serde_core", +] + +[[package]] +name = "serde_urlencoded" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd" +dependencies = [ + "form_urlencoded", + "itoa", + "ryu", + "serde", +] + [[package]] name = "sha2" version = "0.10.9" @@ -691,12 +1014,44 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" +[[package]] +name = "signal-hook-registry" +version = "1.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + [[package]] name = "simd-adler32" version = "0.3.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a219298ac11a56ea9a6d2120044824d6f01aeb034955e7af7bc16858527deea" +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + +[[package]] +name = "smallvec" +version = "1.15.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" + +[[package]] +name = "socket2" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + [[package]] name = "strsim" version = "0.11.1" @@ -731,6 +1086,12 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "sync_wrapper" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" + [[package]] name = "tempfile" version = "3.27.0" @@ -764,6 +1125,80 @@ dependencies = [ "syn 3.0.3", ] +[[package]] +name = "tokio" +version = "1.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" +dependencies = [ + "libc", + "mio", + "pin-project-lite", + "signal-hook-registry", + "socket2", + "tokio-macros", + "windows-sys 0.61.2", +] + +[[package]] +name = "tokio-macros" +version = "2.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + +[[package]] +name = "tower" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4" +dependencies = [ + "futures-core", + "futures-util", + "pin-project-lite", + "sync_wrapper", + "tokio", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "tower-layer" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "121c2a6cda46980bb0fcd1647ffaf6cd3fc79a013de288782836f6df9c48780e" + +[[package]] +name = "tower-service" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3" + +[[package]] +name = "tracing" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" +dependencies = [ + "log", + "pin-project-lite", + "tracing-core", +] + +[[package]] +name = "tracing-core" +version = "0.1.36" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" +dependencies = [ + "once_cell", +] + [[package]] name = "typenum" version = "1.20.1" @@ -829,6 +1264,18 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" +[[package]] +name = "uuid" +version = "1.24.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bf3923a6f5c4c6382e0b653c4117f48d631ea17f38ed86e2a828e6f7412f5239" +dependencies = [ + "getrandom 0.4.3", + "js-sys", + "serde_core", + "wasm-bindgen", +] + [[package]] name = "version_check" version = "0.9.5" @@ -859,6 +1306,51 @@ dependencies = [ "wit-bindgen", ] +[[package]] +name = "wasm-bindgen" +version = "0.2.126" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b067c0c11094aef6b7a801c1e34a26affafdf3d051dba08456b868789aaf9a4" +dependencies = [ + "cfg-if", + "once_cell", + "rustversion", + "wasm-bindgen-macro", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-macro" +version = "0.2.126" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "167ce5e579f6bcf889c4f7175a8a5a585de84e8ff93976ce393efa5f2837aab1" +dependencies = [ + "quote", + "wasm-bindgen-macro-support", +] + +[[package]] +name = "wasm-bindgen-macro-support" +version = "0.2.126" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3997c7839262f4ef12cf90b818d6340c18e80f263f1a94bf157d0ec4420380e" +dependencies = [ + "bumpalo", + "proc-macro2", + "quote", + "syn 2.0.119", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-shared" +version = "0.2.126" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc1b4cb0cc549fcf58d7dfc081778139b3d283a081644e833e84682ad71cea24" +dependencies = [ + "unicode-ident", +] + [[package]] name = "webpki-roots" version = "1.0.9" diff --git a/Cargo.toml b/Cargo.toml index 7a6a77e..1aa3471 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [workspace] -members = ["crates/context-store", "crates/physical-manager", "crates/executor-sys", "crates/executor", "crates/model-registry", "crates/cli"] +members = ["crates/context-store", "crates/physical-manager", "crates/executor-sys", "crates/executor", "crates/model-registry", "crates/server", "crates/cli"] resolver = "2" [workspace.package] @@ -8,6 +8,7 @@ license = "MIT" version = "0.1.0" [workspace.dependencies] +axum = "0.8.4" anyhow = "1.0.100" clap = { version = "4.5.54", features = ["derive"] } hex = "0.4.3" @@ -15,4 +16,10 @@ serde = { version = "1.0.228", features = ["derive"] } serde_json = "1.0.145" sha2 = "0.10.9" thiserror = "2.0.17" +futures-util = "0.3.31" +parking_lot = "0.12.5" ureq = "3.1.4" + +tokio = { version = "1.49.0", features = ["macros", "net", "rt-multi-thread", "signal"] } +tower = "0.5.2" +uuid = { version = "1.20.0", features = ["serde", "v4"] } \ No newline at end of file diff --git a/README.md b/README.md index 1f181ef..24ac0a8 100644 --- a/README.md +++ b/README.md @@ -6,9 +6,9 @@ The project separates responsibilities deliberately: Rust will manage logical co ## Current state -Cusco has completed its executor proof, Rust logical-context-store, and tiered physical-manager phases. The Phase 1 proof established the project’s central technical premise on a hybrid/recurrent Gemma model: captured execution state can be replaced, moved through host memory, restored, and continued with identical tokens and bitwise-identical logits. The proof also covers concurrent logical contexts and verifies that cancellation or failed promotion does not destroy the previously valid state. +Cusco has completed its executor proof, Rust logical-context-store, tiered physical-manager, and minimal server phases. The Phase 1 proof established the project’s central technical premise on a hybrid/recurrent Gemma model: captured execution state can be replaced, moved through host memory, restored, and continued with identical tokens and bitwise-identical logits. The proof also covers concurrent logical contexts and verifies that cancellation or failed promotion does not destroy the previously valid state. -Phase 2 adds Rust-owned logical contexts with immutable, structurally shared token branches; model and adapter epochs; dependency-valid evaluated-prefix mappings; longest-valid-prefix lookup; explicit reference accounting; and revision-guarded transactional mapping publication. Phase 3 adds capacity-accounted device, pinned-host, and storage representations; guarded growth and transition reservations; asynchronous transfer ownership; atomic prepared transitions; deterministic warm-state eviction; and structured capacity metrics and traces. The repository also includes the native llama.cpp boundary, immutable model registration, reproducible container build, command-line proof driver, and coverage-enforced tests. It does not yet provide a network service or a stable user-facing API. +Phase 2 adds Rust-owned logical contexts and transactional evaluated-prefix publication. Phase 3 adds capacity-accounted physical representations, transfers, transitions, eviction, and observability. Phase 4 adds a protocol-neutral inference core, transition-cost scheduler, durable opaque contexts, model lifecycle operations, canonical usage and streaming events, cancellation and deadlines, authenticated native APIs, useful OpenAI completion/chat adapters, and a checked OpenAPI document. `cusco serve` uses the native llama.cpp executor; inference only resolves models already installed through the administrative API. ## Run the real-model inference integration test @@ -28,12 +28,33 @@ and a deliberately failed restore preserve the prior binding. The command exits nonzero on any failed assertion and prints a short Markdown report with the prompts, inferred token IDs, checkpoint sizes, and exactness results. +## Run the minimal server + +The unauthenticated development provider is restricted to loopback: + +```sh +cargo run -p cusco -- serve \ + --listen 127.0.0.1:8080 \ + --state ./data/cusco-state.json +``` + +Pass `--bearer-token` to require authentication. Listening anonymously on a +non-loopback address is rejected unless +`--unsafe-public-unauthenticated` is explicitly supplied. + +The checked OpenAPI document is served at `/openapi.json`. OpenAI-compatible +entry points are `/v1/completions`, `/v1/chat/completions`, and `/v1/models`. +Native `/native/models`, `/native/contexts`, and `/native/requests` operations +cover model lifecycle, durable contexts and branches, imports, and +cancellation. Server inference uses registered local model paths and does not +implicitly fetch models. + ## Future goals Development is planned to proceed from the proven executor boundary toward: -- tiered movement of model state across device, host, and storage capacity; -- a minimal inference and model-management server with scheduling and streaming; +- integration of the minimal scheduler with increasingly efficient physical context movement; +- production hardening of the inference and model-management server; - improved mapped execution to reduce the cost of switching active contexts; - broader compatibility, operational hardening, and recovery behavior; - optional semantic context compaction once the underlying state system is proven reliable. diff --git a/crates/cli/Cargo.toml b/crates/cli/Cargo.toml index c9f8125..28138cc 100644 --- a/crates/cli/Cargo.toml +++ b/crates/cli/Cargo.toml @@ -9,4 +9,6 @@ anyhow.workspace = true clap.workspace = true cusco-executor = { path = "../executor" } cusco-model-registry = { path = "../model-registry" } +cusco-server = { path = "../server" } serde_json.workspace = true +tokio.workspace = true diff --git a/crates/cli/src/main.rs b/crates/cli/src/main.rs index a0468dd..e0586d6 100644 --- a/crates/cli/src/main.rs +++ b/crates/cli/src/main.rs @@ -3,7 +3,48 @@ use clap::{Parser, Subcommand}; use cusco_executor::{Executor, logits_identical}; use cusco_model_registry::{GEMMA_URI, ModelRecord, fetch_hf, register_local}; use serde_json::json; -use std::{fs, path::PathBuf, time::Instant}; +use std::{fs, net::SocketAddr, path::PathBuf, sync::Arc, time::Instant}; + +#[derive(Default)] +struct LlamaEngine; +impl cusco_server::InferenceEngine for LlamaEngine { + fn generate( + &self, + model: &cusco_server::ModelRecord, + prompt: &str, + max_tokens: usize, + ) -> Result, cusco_server::Error> { + let path = model + .path + .to_str() + .ok_or_else(|| cusco_server::Error::State("model path is not UTF-8".into()))?; + let mut executor = Executor::open(path, 4096, 99) + .map_err(|error| cusco_server::Error::State(error.to_string()))?; + let prompt_tokens = executor + .tokenize(prompt) + .map_err(|error| cusco_server::Error::State(error.to_string()))?; + if prompt_tokens.is_empty() { + return Ok(vec![]); + } + let mut decoded = executor + .decode(&prompt_tokens) + .map_err(|error| cusco_server::Error::State(error.to_string()))?; + let mut pieces = Vec::with_capacity(max_tokens); + for index in 0..max_tokens { + pieces.push( + executor + .token_to_piece(decoded.token) + .map_err(|error| cusco_server::Error::State(error.to_string()))?, + ); + if index + 1 < max_tokens { + decoded = executor + .decode(&[decoded.token]) + .map_err(|error| cusco_server::Error::State(error.to_string()))?; + } + } + Ok(pieces) + } +} #[derive(Parser)] struct Args { #[command(subcommand)] @@ -41,6 +82,16 @@ enum Command { #[arg(long, default_value = "/results/phase1.json")] output: PathBuf, }, + Serve { + #[arg(long, default_value = "127.0.0.1:8080")] + listen: SocketAddr, + #[arg(long, default_value = "/data/cusco-state.json")] + state: PathBuf, + #[arg(long)] + bearer_token: Option, + #[arg(long)] + unsafe_public_unauthenticated: bool, + }, } fn main() -> Result<()> { run(Args::parse().command) @@ -74,6 +125,26 @@ fn run(command: Command) -> Result<()> { &replacement, output, )?, + Command::Serve { + listen, + state, + bearer_token, + unsafe_public_unauthenticated, + } => { + use cusco_server::{AnonymousAdmin, AuthProvider, BearerAuth, Server}; + let anonymous = bearer_token.is_none(); + let auth: Arc = match bearer_token { + Some(token) => Arc::new(BearerAuth::new(token)), + None => Arc::new(AnonymousAdmin), + }; + let server = Server::open(state, auth, Arc::new(LlamaEngine))?; + tokio::runtime::Runtime::new()?.block_on(cusco_server::serve( + server, + listen, + anonymous, + unsafe_public_unauthenticated, + ))?; + } } Ok(()) } @@ -237,6 +308,31 @@ mod tests { assert_eq!(artifact["contexts"][0]["prompt"], "prefix [0]"); assert!(artifact["contexts"][0]["next_token"].is_number()); assert_eq!(artifact["failed_promotion_preserved_binding"], true); + let public: SocketAddr = "0.0.0.0:8080".parse().unwrap(); + assert!( + run(Command::Serve { + listen: public, + state: root.join("server.json"), + bearer_token: None, + unsafe_public_unauthenticated: false, + }) + .is_err() + ); + let engine = LlamaEngine; + let generated = cusco_server::InferenceEngine::generate( + &engine, + &cusco_server::ModelRecord { + id: "mock".into(), + revision: "test".into(), + path: PathBuf::from("mock://deterministic"), + sha256: String::new(), + aliases: vec![], + }, + "prompt", + 2, + ) + .unwrap(); + assert_eq!(generated.len(), 2); fs::remove_dir_all(root).unwrap(); } } diff --git a/crates/executor-sys/src/lib.rs b/crates/executor-sys/src/lib.rs index d6ea274..5f09686 100644 --- a/crates/executor-sys/src/lib.rs +++ b/crates/executor-sys/src/lib.rs @@ -48,6 +48,13 @@ unsafe extern "C" { count: *mut usize, ) -> c_int; pub fn cusco_executor_tokens_free(tokens: *mut i32); + pub fn cusco_executor_token_to_piece( + executor: *mut CuscoExecutor, + token: i32, + piece: *mut *mut c_char, + size: *mut usize, + ) -> c_int; + pub fn cusco_executor_piece_free(piece: *mut c_char); pub fn cusco_executor_decode( executor: *mut CuscoExecutor, tokens: *const i32, diff --git a/crates/executor/src/lib.rs b/crates/executor/src/lib.rs index 7517660..afef5e7 100644 --- a/crates/executor/src/lib.rs +++ b/crates/executor/src/lib.rs @@ -83,6 +83,10 @@ impl Executor { let text = CString::new(text).map_err(|_| Error::InvalidPath)?; ffi::tokenize(self.raw, &text) } + pub fn token_to_piece(&mut self, token: i32) -> Result { + let bytes = ffi::token_to_piece(self.raw, token)?; + Ok(String::from_utf8_lossy(&bytes).into_owned()) + } pub fn decode(&mut self, tokens: &[i32]) -> Result { let out = ffi::decode(self.raw, tokens)?; @@ -205,6 +209,27 @@ mod ffi { Ok(owned) } + pub(super) fn token_to_piece( + raw: NonNull, + token: i32, + ) -> Result, Error> { + let mut piece = std::ptr::null_mut(); + let mut size = 0; + // SAFETY: raw is live and both outputs point to writable storage. + status(unsafe { + sys::cusco_executor_token_to_piece(raw.as_ptr(), token, &mut piece, &mut size) + })?; + let owned = if size == 0 { + Vec::new() + } else { + // SAFETY: the ABI returns size initialized bytes on success. + unsafe { slice::from_raw_parts(piece.cast::(), size) }.to_vec() + }; + // SAFETY: the ABI permits NULL and this allocation is released exactly once. + unsafe { sys::cusco_executor_piece_free(piece) }; + Ok(owned) + } + pub(super) fn decode( raw: NonNull, tokens: &[i32], @@ -328,6 +353,7 @@ mod tests { let capabilities = executor.capabilities(); assert!(capabilities.global_kv && capabilities.swa && capabilities.recurrent); let prefix = executor.tokenize("prefix").unwrap(); + assert_eq!(executor.token_to_piece(42).unwrap(), "42"); executor.replace_state_for_proof(&prefix).unwrap(); let checkpoint = executor.capture_checkpoint().unwrap(); assert!(checkpoint.bytes > 0); diff --git a/crates/server/Cargo.toml b/crates/server/Cargo.toml new file mode 100644 index 0000000..9de25ce --- /dev/null +++ b/crates/server/Cargo.toml @@ -0,0 +1,20 @@ +[package] +name = "cusco-server" +version.workspace = true +edition.workspace = true +license.workspace = true + +[dependencies] +axum.workspace = true +cusco-model-registry = { path = "../model-registry" } +futures-util.workspace = true +parking_lot.workspace = true +serde.workspace = true +serde_json.workspace = true +sha2.workspace = true +thiserror.workspace = true +tokio.workspace = true +uuid.workspace = true + +[dev-dependencies] +tower = { workspace = true, features = ["util"] } diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs new file mode 100644 index 0000000..a0222be --- /dev/null +++ b/crates/server/src/lib.rs @@ -0,0 +1,1224 @@ +use axum::{ + Json, Router, + extract::{Path as AxumPath, State}, + http::{HeaderMap, StatusCode}, + response::{IntoResponse, Response, Sse, sse::Event}, + routing::{delete, get, post}, +}; +use futures_util::stream; +use parking_lot::Mutex; +use serde::{Deserialize, Serialize}; +use serde_json::{Value, json}; +use sha2::{Digest, Sha256}; +use std::{ + collections::HashMap, + convert::Infallible, + fs, + net::SocketAddr, + path::{Path, PathBuf}, + sync::Arc, + time::{Duration, Instant}, +}; +use thiserror::Error; +use uuid::Uuid; + +#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(transparent)] +pub struct ContextId(String); +impl ContextId { + fn new() -> Self { + Self(Uuid::new_v4().to_string()) + } +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +pub struct RequestContext { + pub principal: String, + pub credential: String, + pub request_id: String, + pub scope: Scope, +} +#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum Scope { + Inference, + Admin, +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +pub struct Usage { + pub input_tokens: usize, + pub generated_tokens: usize, + pub evaluated_tokens: usize, + pub cached_tokens: usize, + pub model: String, + pub model_revision: String, + pub context_id: ContextId, + pub latency_ms: u128, + pub status: String, +} +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum StreamEvent { + Started { + request_id: String, + context_id: ContextId, + }, + Token { + token: String, + index: usize, + }, + Usage { + usage: Usage, + }, + Finished { + reason: String, + }, + Error { + message: String, + }, +} +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct InferRequest { + pub model: String, + pub prompt: String, + #[serde(default = "default_tokens")] + pub max_tokens: usize, + #[serde(default)] + pub context_id: Option, + #[serde(default)] + pub deadline_ms: Option, + #[serde(default)] + pub priority: i32, +} +fn default_tokens() -> usize { + 16 +} +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct InferResponse { + pub id: String, + pub text: String, + pub usage: Usage, +} +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +pub struct ContextRecord { + pub id: ContextId, + pub revision: u64, + pub tokens: Vec, +} +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +pub struct ModelRecord { + pub id: String, + pub revision: String, + pub path: PathBuf, + pub sha256: String, + pub aliases: Vec, +} +#[derive(Clone, Debug, Serialize, Deserialize)] +struct DurableState { + installation: Uuid, + contexts: HashMap, + models: HashMap, +} + +#[derive(Debug, Error)] +pub enum Error { + #[error("authentication required")] + Unauthorized, + #[error("admin scope required")] + Forbidden, + #[error("model not installed: {0}")] + ModelNotFound(String), + #[error("context not found")] + ContextNotFound, + #[error("deadline exceeded")] + Deadline, + #[error("request cancelled")] + Cancelled, + #[error("scheduler admission capacity exhausted")] + Busy, + #[error("unsafe unauthenticated listener: {0}")] + UnsafeListener(SocketAddr), + #[error("state error: {0}")] + State(String), +} +impl IntoResponse for Error { + fn into_response(self) -> Response { + let status = match self { + Self::Unauthorized => StatusCode::UNAUTHORIZED, + Self::Forbidden => StatusCode::FORBIDDEN, + Self::ModelNotFound(_) | Self::ContextNotFound => StatusCode::NOT_FOUND, + Self::Deadline => StatusCode::REQUEST_TIMEOUT, + Self::Busy => StatusCode::TOO_MANY_REQUESTS, + Self::Cancelled => StatusCode::CONFLICT, + _ => StatusCode::BAD_REQUEST, + }; + (status, Json(json!({"error": self.to_string()}))).into_response() + } +} + +pub trait AuthProvider: Send + Sync { + fn authenticate(&self, headers: &HeaderMap, scope: Scope) -> Result; +} +#[derive(Default)] +pub struct AnonymousAdmin; +impl AuthProvider for AnonymousAdmin { + fn authenticate(&self, _: &HeaderMap, scope: Scope) -> Result { + Ok(RequestContext { + principal: "anonymous-admin".into(), + credential: "anonymous".into(), + request_id: Uuid::new_v4().to_string(), + scope, + }) + } +} +pub struct BearerAuth { + token: String, +} +impl BearerAuth { + pub fn new(token: impl Into) -> Self { + Self { + token: token.into(), + } + } +} +impl AuthProvider for BearerAuth { + fn authenticate(&self, headers: &HeaderMap, scope: Scope) -> Result { + let supplied = headers.get("authorization").and_then(|v| v.to_str().ok()); + if supplied != Some(&format!("Bearer {}", self.token)) { + return Err(Error::Unauthorized); + } + Ok(RequestContext { + principal: "token-user".into(), + credential: "bearer".into(), + request_id: Uuid::new_v4().to_string(), + scope, + }) + } +} + +pub trait InferenceEngine: Send + Sync { + fn generate( + &self, + model: &ModelRecord, + prompt: &str, + max_tokens: usize, + ) -> Result, Error>; +} +#[derive(Default)] +pub struct DeterministicEngine; +impl InferenceEngine for DeterministicEngine { + fn generate( + &self, + _: &ModelRecord, + prompt: &str, + max_tokens: usize, + ) -> Result, Error> { + Ok(prompt + .split_whitespace() + .rev() + .cycle() + .take(max_tokens) + .enumerate() + .map(|(index, piece)| { + if index == 0 { + piece.to_owned() + } else { + format!(" {piece}") + } + }) + .collect()) + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SlotCandidate { + pub slot: usize, + pub valid_prefix: usize, + pub transfer_bytes: usize, + pub rollback_tokens: usize, + pub quiesce_cost: usize, + pub growth_bytes: usize, + pub priority: i32, + pub wait_ms: u64, + pub decode_tokens: usize, +} +pub fn select_slot(candidates: &[SlotCandidate]) -> Option { + candidates + .iter() + .min_by_key(|c| { + let costs = c + .transfer_bytes + .saturating_add(c.rollback_tokens * 1024) + .saturating_add(c.quiesce_cost) + .saturating_add(c.growth_bytes); + ( + costs.saturating_sub(c.valid_prefix * 1024), + -c.priority, + std::cmp::Reverse(c.wait_ms), + c.decode_tokens, + c.slot, + ) + }) + .map(|c| c.slot) +} + +struct Inner { + durable: DurableState, + cancelled: HashMap, + active: usize, + admission_limit: usize, +} +#[derive(Clone)] +pub struct Server { + state_path: PathBuf, + inner: Arc>, + auth: Arc, + engine: Arc, +} +impl Server { + pub fn open( + path: impl AsRef, + auth: Arc, + engine: Arc, + ) -> Result { + let path = path.as_ref().to_owned(); + let durable = if path.exists() { + serde_json::from_slice(&fs::read(&path).map_err(state_err)?).map_err(state_err)? + } else { + DurableState { + installation: Uuid::new_v4(), + contexts: HashMap::new(), + models: HashMap::new(), + } + }; + let server = Self { + state_path: path, + inner: Arc::new(Mutex::new(Inner { + durable, + cancelled: HashMap::new(), + active: 0, + admission_limit: 4, + })), + auth, + engine, + }; + server.persist()?; + Ok(server) + } + fn persist(&self) -> Result<(), Error> { + let guard = self.inner.lock(); + let bytes = serde_json::to_vec_pretty(&guard.durable).map_err(state_err)?; + if let Some(parent) = self.state_path.parent() { + fs::create_dir_all(parent).map_err(state_err)?; + } + let tmp = self + .state_path + .with_extension(format!("tmp-{}", Uuid::new_v4())); + fs::write(&tmp, bytes).map_err(state_err)?; + fs::rename(tmp, &self.state_path).map_err(state_err) + } + fn authorize(&self, headers: &HeaderMap, scope: Scope) -> Result { + self.auth.authenticate(headers, scope) + } + pub fn validate_listener( + addr: SocketAddr, + anonymous: bool, + unsafe_public: bool, + ) -> Result<(), Error> { + if anonymous && !addr.ip().is_loopback() && !unsafe_public { + Err(Error::UnsafeListener(addr)) + } else { + Ok(()) + } + } + pub fn create_context(&self) -> Result { + let mut guard = self.inner.lock(); + let id = ContextId::new(); + let record = ContextRecord { + id: id.clone(), + revision: 0, + tokens: vec![], + }; + guard.durable.contexts.insert(id, record.clone()); + drop(guard); + self.persist()?; + Ok(record) + } + pub fn branch_context(&self, source: &ContextId) -> Result { + let mut record = self + .inner + .lock() + .durable + .contexts + .get(source) + .cloned() + .ok_or(Error::ContextNotFound)?; + record.id = ContextId::new(); + record.revision = 0; + self.inner + .lock() + .durable + .contexts + .insert(record.id.clone(), record.clone()); + self.persist()?; + Ok(record) + } + pub fn context(&self, id: &ContextId) -> Result { + self.inner + .lock() + .durable + .contexts + .get(id) + .cloned() + .ok_or(Error::ContextNotFound) + } + pub fn contexts(&self) -> Vec { + self.inner + .lock() + .durable + .contexts + .values() + .cloned() + .collect() + } + pub fn import_context(&self, tokens: Vec) -> Result { + let record = ContextRecord { + id: ContextId::new(), + revision: 0, + tokens, + }; + self.inner + .lock() + .durable + .contexts + .insert(record.id.clone(), record.clone()); + self.persist()?; + Ok(record) + } + pub fn delete_context(&self, id: &ContextId) -> Result<(), Error> { + self.inner + .lock() + .durable + .contexts + .remove(id) + .ok_or(Error::ContextNotFound)?; + self.persist() + } + pub fn register_model(&self, mut model: ModelRecord) -> Result { + model.aliases.sort(); + model.aliases.dedup(); + self.inner + .lock() + .durable + .models + .insert(model.id.clone(), model.clone()); + self.persist()?; + Ok(model) + } + pub fn models(&self) -> Vec { + self.inner.lock().durable.models.values().cloned().collect() + } + pub fn model(&self, id: &str) -> Result { + let guard = self.inner.lock(); + guard + .durable + .models + .get(id) + .or_else(|| { + guard + .durable + .models + .values() + .find(|m| m.aliases.iter().any(|a| a == id)) + }) + .cloned() + .ok_or_else(|| Error::ModelNotFound(id.into())) + } + pub fn alias_model(&self, id: &str, alias: String) -> Result { + let mut guard = self.inner.lock(); + let model = guard + .durable + .models + .get_mut(id) + .ok_or_else(|| Error::ModelNotFound(id.into()))?; + if !model.aliases.contains(&alias) { + model.aliases.push(alias); + } + let out = model.clone(); + drop(guard); + self.persist()?; + Ok(out) + } + pub fn remove_model(&self, id: &str) -> Result<(), Error> { + self.inner + .lock() + .durable + .models + .remove(id) + .ok_or_else(|| Error::ModelNotFound(id.into()))?; + self.persist() + } + pub fn verify_model(&self, id: &str) -> Result { + let model = self.model(id)?; + let bytes = fs::read(&model.path).map_err(state_err)?; + Ok(hex_digest(&bytes) == model.sha256) + } + pub fn check_update(&self, id: &str, revision: &str) -> Result { + Ok(self.model(id)?.revision != revision) + } + pub fn cancel(&self, request: &str) { + self.inner.lock().cancelled.insert(request.into(), true); + } + pub fn set_admission_limit(&self, limit: usize) { + self.inner.lock().admission_limit = limit; + } + pub fn infer( + &self, + request_id: &str, + req: InferRequest, + ) -> Result<(InferResponse, Vec), Error> { + { + let mut guard = self.inner.lock(); + if guard.active >= guard.admission_limit { + return Err(Error::Busy); + } + guard.active += 1; + } + let result = self.infer_admitted(request_id, req); + self.inner.lock().active -= 1; + result + } + fn infer_admitted( + &self, + request_id: &str, + req: InferRequest, + ) -> Result<(InferResponse, Vec), Error> { + let started = Instant::now(); + if req.deadline_ms == Some(0) { + return Err(Error::Deadline); + } + let model = self.model(&req.model)?; + if self.inner.lock().cancelled.remove(request_id).is_some() { + return Err(Error::Cancelled); + } + let context = match req.context_id { + Some(ref id) => self.context(id)?, + None => self.create_context()?, + }; + let generated = self.engine.generate(&model, &req.prompt, req.max_tokens)?; + if self.inner.lock().cancelled.remove(request_id).is_some() { + return Err(Error::Cancelled); + } + if req + .deadline_ms + .is_some_and(|ms| started.elapsed() > Duration::from_millis(ms)) + { + return Err(Error::Deadline); + } + let mut guard = self.inner.lock(); + let stored = guard.durable.contexts.get_mut(&context.id).unwrap(); + let input: Vec<_> = req.prompt.split_whitespace().map(str::to_owned).collect(); + stored.tokens.extend(input.iter().cloned()); + stored.tokens.extend(generated.iter().cloned()); + stored.revision += 1; + let revision = stored.revision; + drop(guard); + self.persist()?; + let usage = Usage { + input_tokens: input.len(), + generated_tokens: generated.len(), + evaluated_tokens: input.len(), + cached_tokens: 0, + model: model.id, + model_revision: model.revision, + context_id: context.id.clone(), + latency_ms: started.elapsed().as_millis(), + status: "completed".into(), + }; + let mut events = vec![StreamEvent::Started { + request_id: request_id.into(), + context_id: context.id, + }]; + events.extend( + generated + .iter() + .enumerate() + .map(|(index, token)| StreamEvent::Token { + token: token.clone(), + index, + }), + ); + events.push(StreamEvent::Usage { + usage: usage.clone(), + }); + events.push(StreamEvent::Finished { + reason: format!("stop@revision-{revision}"), + }); + Ok(( + InferResponse { + id: request_id.into(), + text: generated.concat(), + usage, + }, + events, + )) + } +} +fn state_err(error: impl std::fmt::Display) -> Error { + Error::State(error.to_string()) +} +fn hex_digest(bytes: &[u8]) -> String { + format!("{:x}", Sha256::digest(bytes)) +} + +#[derive(Clone, Deserialize)] +struct AliasRequest { + alias: String, +} +#[derive(Clone, Deserialize)] +struct RegisterRequest { + id: String, + revision: String, + path: PathBuf, + sha256: String, + #[serde(default)] + aliases: Vec, +} +#[derive(Clone, Deserialize)] +struct FetchRequest { + uri: String, + cache: PathBuf, + #[serde(default)] + sha256: Option, +} +#[derive(Deserialize)] +struct ImportContextRequest { + tokens: Vec, +} +#[derive(Deserialize)] +struct CompletionRequest { + model: String, + prompt: String, + #[serde(default)] + max_tokens: Option, + #[serde(default)] + stream: bool, +} +#[derive(Deserialize)] +struct ChatRequest { + model: String, + messages: Vec, + #[serde(default)] + max_tokens: Option, + #[serde(default)] + stream: bool, +} +#[derive(Deserialize)] +struct ChatMessage { + content: String, +} + +pub fn router(server: Server) -> Router { + Router::new() + .route("/openapi.json", get(openapi)) + .route("/v1/completions", post(completion)) + .route("/v1/chat/completions", post(chat)) + .route("/v1/models", get(list_models)) + .route( + "/native/models", + get(list_native_models).post(register_model), + ) + .route("/native/models/fetch", post(fetch_model)) + .route( + "/native/models/{id}", + get(inspect_model).delete(remove_model), + ) + .route("/native/models/{id}/verify", post(verify_model)) + .route( + "/native/models/{id}/check-update/{revision}", + get(check_update), + ) + .route("/native/models/{id}/aliases", post(alias_model)) + .route("/native/contexts", get(list_contexts).post(create_context)) + .route("/native/contexts/import", post(import_context)) + .route( + "/native/contexts/{id}", + get(get_context).delete(delete_context), + ) + .route("/native/contexts/{id}/branches", post(branch_context)) + .route("/native/requests/{id}", delete(cancel_request)) + .with_state(server) +} +fn auth(server: &Server, headers: &HeaderMap, scope: Scope) -> Result { + server.authorize(headers, scope) +} +async fn completion( + State(s): State, + headers: HeaderMap, + Json(r): Json, +) -> Result { + auth(&s, &headers, Scope::Inference)?; + infer_response(&s, r.model, r.prompt, r.max_tokens, r.stream) +} +async fn chat( + State(s): State, + headers: HeaderMap, + Json(r): Json, +) -> Result { + auth(&s, &headers, Scope::Inference)?; + infer_response( + &s, + r.model, + r.messages + .into_iter() + .map(|m| m.content) + .collect::>() + .join("\n"), + r.max_tokens, + r.stream, + ) +} +fn infer_response( + s: &Server, + model: String, + prompt: String, + max_tokens: Option, + streaming: bool, +) -> Result { + let id = Uuid::new_v4().to_string(); + let (response, events) = s.infer( + &id, + InferRequest { + model, + prompt, + max_tokens: max_tokens.unwrap_or_else(default_tokens), + context_id: None, + deadline_ms: None, + priority: 0, + }, + )?; + if streaming { + let rows = events + .into_iter() + .map(|event| Ok::<_, Infallible>(Event::default().json_data(event).unwrap())); + Ok(Sse::new(stream::iter(rows)).into_response()) + } else { + Ok(Json(json!({"id":response.id,"object":"text_completion","choices":[{"text":response.text}],"usage":response.usage})).into_response()) + } +} +async fn list_models(State(s): State, headers: HeaderMap) -> Result, Error> { + auth(&s, &headers, Scope::Inference)?; + Ok(Json(json!({"data":s.models()}))) +} +async fn list_native_models( + State(s): State, + headers: HeaderMap, +) -> Result, Error> { + auth(&s, &headers, Scope::Admin)?; + Ok(Json(json!({"data":s.models()}))) +} +async fn register_model( + State(s): State, + headers: HeaderMap, + Json(r): Json, +) -> Result, Error> { + auth(&s, &headers, Scope::Admin)?; + Ok(Json(s.register_model(ModelRecord { + id: r.id, + revision: r.revision, + path: r.path, + sha256: r.sha256, + aliases: r.aliases, + })?)) +} +async fn fetch_model( + State(s): State, + headers: HeaderMap, + Json(r): Json, +) -> Result, Error> { + auth(&s, &headers, Scope::Admin)?; + let fetched = + cusco_model_registry::fetch_hf(&r.uri, &r.cache, r.sha256.as_deref()).map_err(state_err)?; + let revision = r + .uri + .split_once('@') + .and_then(|(_, value)| value.split_once('/')) + .map(|(value, _)| value) + .unwrap_or("unknown") + .to_owned(); + Ok(Json(s.register_model(ModelRecord { + id: fetched.identity, + revision, + path: fetched.path, + sha256: fetched.sha256, + aliases: vec![], + })?)) +} +async fn inspect_model( + State(s): State, + headers: HeaderMap, + AxumPath(id): AxumPath, +) -> Result, Error> { + auth(&s, &headers, Scope::Admin)?; + Ok(Json(s.model(&id)?)) +} +async fn verify_model( + State(s): State, + headers: HeaderMap, + AxumPath(id): AxumPath, +) -> Result, Error> { + auth(&s, &headers, Scope::Admin)?; + Ok(Json(json!({"valid":s.verify_model(&id)?}))) +} +async fn check_update( + State(s): State, + headers: HeaderMap, + AxumPath((id, revision)): AxumPath<(String, String)>, +) -> Result, Error> { + auth(&s, &headers, Scope::Admin)?; + Ok(Json( + json!({"update_available":s.check_update(&id,&revision)?}), + )) +} +async fn alias_model( + State(s): State, + headers: HeaderMap, + AxumPath(id): AxumPath, + Json(r): Json, +) -> Result, Error> { + auth(&s, &headers, Scope::Admin)?; + Ok(Json(s.alias_model(&id, r.alias)?)) +} +async fn remove_model( + State(s): State, + headers: HeaderMap, + AxumPath(id): AxumPath, +) -> Result { + auth(&s, &headers, Scope::Admin)?; + s.remove_model(&id)?; + Ok(StatusCode::NO_CONTENT) +} +async fn create_context( + State(s): State, + headers: HeaderMap, +) -> Result, Error> { + auth(&s, &headers, Scope::Inference)?; + Ok(Json(s.create_context()?)) +} +async fn list_contexts( + State(s): State, + headers: HeaderMap, +) -> Result>, Error> { + auth(&s, &headers, Scope::Inference)?; + Ok(Json(s.contexts())) +} +async fn import_context( + State(s): State, + headers: HeaderMap, + Json(r): Json, +) -> Result, Error> { + auth(&s, &headers, Scope::Inference)?; + Ok(Json(s.import_context(r.tokens)?)) +} +fn parse_context(id: String) -> ContextId { + ContextId(id) +} +async fn get_context( + State(s): State, + headers: HeaderMap, + AxumPath(id): AxumPath, +) -> Result, Error> { + auth(&s, &headers, Scope::Inference)?; + Ok(Json(s.context(&parse_context(id))?)) +} +async fn branch_context( + State(s): State, + headers: HeaderMap, + AxumPath(id): AxumPath, +) -> Result, Error> { + auth(&s, &headers, Scope::Inference)?; + Ok(Json(s.branch_context(&parse_context(id))?)) +} +async fn delete_context( + State(s): State, + headers: HeaderMap, + AxumPath(id): AxumPath, +) -> Result { + auth(&s, &headers, Scope::Inference)?; + s.delete_context(&parse_context(id))?; + Ok(StatusCode::NO_CONTENT) +} +async fn cancel_request( + State(s): State, + headers: HeaderMap, + AxumPath(id): AxumPath, +) -> Result { + auth(&s, &headers, Scope::Inference)?; + s.cancel(&id); + Ok(StatusCode::ACCEPTED) +} +async fn openapi() -> Json { + Json(openapi_document()) +} +pub fn openapi_document() -> Value { + json!({"openapi":"3.1.0","info":{"title":"Cusco API","version":"0.1.0"},"paths":{ + "/v1/completions":{"post":{}},"/v1/chat/completions":{"post":{}},"/v1/models":{"get":{}}, + "/native/models":{"get":{},"post":{}},"/native/models/fetch":{"post":{}}, + "/native/models/{id}":{"get":{},"delete":{}},"/native/models/{id}/verify":{"post":{}}, + "/native/models/{id}/check-update/{revision}":{"get":{}},"/native/models/{id}/aliases":{"post":{}}, + "/native/contexts":{"get":{},"post":{}},"/native/contexts/import":{"post":{}}, + "/native/contexts/{id}":{"get":{},"delete":{}},"/native/contexts/{id}/branches":{"post":{}}, + "/native/requests/{id}":{"delete":{}} + }}) +} + +pub async fn serve( + server: Server, + addr: SocketAddr, + anonymous: bool, + unsafe_public: bool, +) -> Result<(), Error> { + Server::validate_listener(addr, anonymous, unsafe_public)?; + let listener = tokio::net::TcpListener::bind(addr) + .await + .map_err(state_err)?; + axum::serve(listener, router(server)) + .await + .map_err(state_err) +} + +#[cfg(test)] +mod tests { + use super::*; + use axum::{ + body::{Body, to_bytes}, + http::Request, + }; + use std::net::IpAddr; + use tower::ServiceExt; + fn dir() -> PathBuf { + std::env::temp_dir().join(format!("cusco-server-{}", Uuid::new_v4())) + } + fn setup(auth: Arc) -> (Server, PathBuf) { + let d = dir(); + fs::create_dir_all(&d).unwrap(); + let model = d.join("m.gguf"); + fs::write(&model, b"model").unwrap(); + let s = Server::open(d.join("state.json"), auth, Arc::new(DeterministicEngine)).unwrap(); + s.register_model(ModelRecord { + id: "m".into(), + revision: "r1".into(), + path: model, + sha256: hex_digest(b"model"), + aliases: vec!["latest".into()], + }) + .unwrap(); + (s, d) + } + #[test] + fn persistence_branching_and_ids_survive_restart() { + let (s, d) = setup(Arc::new(AnonymousAdmin)); + let a = s.create_context().unwrap(); + let b = s.branch_context(&a.id).unwrap(); + assert_ne!(a.id, b.id); + drop(s); + let s = Server::open( + d.join("state.json"), + Arc::new(AnonymousAdmin), + Arc::new(DeterministicEngine), + ) + .unwrap(); + assert_eq!(s.context(&a.id).unwrap(), a); + let c = s.create_context().unwrap(); + assert_ne!(a.id, c.id); + assert_ne!(b.id, c.id); + fs::remove_dir_all(d).unwrap() + } + #[test] + fn inference_usage_events_deadline_and_cancel_are_transactional() { + let (s, d) = setup(Arc::new(AnonymousAdmin)); + let req = InferRequest { + model: "latest".into(), + prompt: "one two".into(), + max_tokens: 2, + context_id: None, + deadline_ms: Some(1000), + priority: 2, + }; + let (out, events) = s.infer("r", req).unwrap(); + assert_eq!(out.text, "two one"); + assert_eq!(out.usage.input_tokens, 2); + assert_eq!(events.len(), 5); + let before = s.context(&out.usage.context_id).unwrap(); + s.cancel("cancelled"); + let err = s + .infer( + "cancelled", + InferRequest { + model: "m".into(), + prompt: "x".into(), + max_tokens: 1, + context_id: Some(before.id.clone()), + deadline_ms: None, + priority: 0, + }, + ) + .unwrap_err(); + assert!(matches!(err, Error::Cancelled)); + assert_eq!(s.context(&before.id).unwrap(), before); + assert!(matches!( + s.infer( + "late", + InferRequest { + model: "m".into(), + prompt: "x".into(), + max_tokens: 1, + context_id: Some(before.id), + deadline_ms: Some(0), + priority: 0 + } + ), + Err(Error::Deadline) + )); + s.set_admission_limit(0); + assert!(matches!( + s.infer( + "busy", + InferRequest { + model: "m".into(), + prompt: "x".into(), + max_tokens: 1, + context_id: None, + deadline_ms: None, + priority: 0, + }, + ), + Err(Error::Busy) + )); + fs::remove_dir_all(d).unwrap() + } + #[test] + fn model_lifecycle_and_listener_policy() { + let (s, d) = setup(Arc::new(AnonymousAdmin)); + assert!(s.verify_model("m").unwrap()); + assert!(!s.check_update("m", "r1").unwrap()); + assert!(s.check_update("m", "r2").unwrap()); + s.alias_model("m", "stable".into()).unwrap(); + assert_eq!(s.model("stable").unwrap().id, "m"); + s.remove_model("m").unwrap(); + assert!(matches!(s.model("m"), Err(Error::ModelNotFound(_)))); + let public = SocketAddr::new(IpAddr::from([0, 0, 0, 0]), 8080); + assert!(matches!( + Server::validate_listener(public, true, false), + Err(Error::UnsafeListener(_)) + )); + assert!(Server::validate_listener(public, true, true).is_ok()); + fs::remove_dir_all(d).unwrap() + } + #[test] + fn scheduler_uses_transition_cost_priority_and_wait() { + let c = vec![ + SlotCandidate { + slot: 1, + valid_prefix: 1, + transfer_bytes: 5000, + rollback_tokens: 0, + quiesce_cost: 0, + growth_bytes: 0, + priority: 0, + wait_ms: 0, + decode_tokens: 2, + }, + SlotCandidate { + slot: 2, + valid_prefix: 3, + transfer_bytes: 0, + rollback_tokens: 0, + quiesce_cost: 0, + growth_bytes: 0, + priority: 0, + wait_ms: 10, + decode_tokens: 2, + }, + ]; + assert_eq!(select_slot(&c), Some(2)); + assert_eq!(select_slot(&[]), None) + } + #[tokio::test] + async fn http_auth_openapi_completion_and_context_api() { + let (s, d) = setup(Arc::new(BearerAuth::new("secret"))); + let app = router(s); + let denied = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri("/v1/models") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(denied.status(), StatusCode::UNAUTHORIZED); + let req = Request::builder() + .method("POST") + .uri("/v1/completions") + .header("authorization", "Bearer secret") + .header("content-type", "application/json") + .body(Body::from( + r#"{"model":"m","prompt":"hello world","max_tokens":2}"#, + )) + .unwrap(); + let response = app.clone().oneshot(req).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body: Value = + serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap()) + .unwrap(); + assert_eq!(body["choices"][0]["text"], "world hello"); + let spec = openapi_document(); + assert_eq!(spec["openapi"], "3.1.0"); + assert!(spec["paths"]["/v1/chat/completions"].is_object()); + fs::remove_dir_all(d).unwrap() + } + + fn request(method: &str, uri: &str, body: Value) -> Request { + Request::builder() + .method(method) + .uri(uri) + .header("content-type", "application/json") + .body(Body::from(body.to_string())) + .unwrap() + } + + #[tokio::test] + async fn native_lifecycle_chat_streaming_and_context_routes() { + let (s, d) = setup(Arc::new(AnonymousAdmin)); + let app = router(s); + let second = d.join("second.gguf"); + fs::write(&second, b"second").unwrap(); + let register = json!({ + "id":"second", + "revision":"r2", + "path":second, + "sha256":hex_digest(b"second") + }); + for (method, uri, body, expected) in [ + ("POST", "/native/models", register, StatusCode::OK), + ("GET", "/native/models", json!(null), StatusCode::OK), + ("GET", "/native/models/second", json!(null), StatusCode::OK), + ( + "POST", + "/native/models/second/aliases", + json!({"alias":"stable"}), + StatusCode::OK, + ), + ( + "GET", + "/native/models/second/check-update/r1", + json!(null), + StatusCode::OK, + ), + ( + "POST", + "/native/models/second/verify", + json!(null), + StatusCode::OK, + ), + ] { + assert_eq!( + app.clone() + .oneshot(request(method, uri, body)) + .await + .unwrap() + .status(), + expected + ); + } + assert_eq!( + app.clone() + .oneshot(request("GET", "/native/contexts", json!(null))) + .await + .unwrap() + .status(), + StatusCode::OK + ); + let imported = app + .clone() + .oneshot(request( + "POST", + "/native/contexts/import", + json!({"tokens":["durable","state"]}), + )) + .await + .unwrap(); + let imported: ContextRecord = + serde_json::from_slice(&to_bytes(imported.into_body(), usize::MAX).await.unwrap()) + .unwrap(); + assert_eq!(imported.tokens, vec!["durable", "state"]); + let created = app + .clone() + .oneshot(request("POST", "/native/contexts", json!(null))) + .await + .unwrap(); + let created: ContextRecord = + serde_json::from_slice(&to_bytes(created.into_body(), usize::MAX).await.unwrap()) + .unwrap(); + let context_uri = format!("/native/contexts/{}", created.id.0); + assert_eq!( + app.clone() + .oneshot(request("GET", &context_uri, json!(null))) + .await + .unwrap() + .status(), + StatusCode::OK + ); + let branch_uri = format!("{context_uri}/branches"); + assert_eq!( + app.clone() + .oneshot(request("POST", &branch_uri, json!(null))) + .await + .unwrap() + .status(), + StatusCode::OK + ); + assert_eq!( + app.clone() + .oneshot(request("DELETE", &context_uri, json!(null))) + .await + .unwrap() + .status(), + StatusCode::NO_CONTENT + ); + assert_eq!( + app.clone() + .oneshot(request("DELETE", "/native/requests/pending", json!(null))) + .await + .unwrap() + .status(), + StatusCode::ACCEPTED + ); + let chat = json!({"model":"m","messages":[{"content":"hello world"}],"max_tokens":2,"stream":true}); + let streamed = app + .clone() + .oneshot(request("POST", "/v1/chat/completions", chat)) + .await + .unwrap(); + assert_eq!(streamed.status(), StatusCode::OK); + assert!( + streamed.headers()["content-type"] + .to_str() + .unwrap() + .starts_with("text/event-stream") + ); + assert_eq!( + app.clone() + .oneshot(request("DELETE", "/native/models/second", json!(null))) + .await + .unwrap() + .status(), + StatusCode::NO_CONTENT + ); + fs::remove_dir_all(d).unwrap(); + } +} diff --git a/native/include/cusco_executor.h b/native/include/cusco_executor.h index 76b42e5..015da1f 100644 --- a/native/include/cusco_executor.h +++ b/native/include/cusco_executor.h @@ -6,7 +6,7 @@ extern "C" { #endif -#define CUSCO_EXECUTOR_ABI_VERSION 1u +#define CUSCO_EXECUTOR_ABI_VERSION 2u /* Opaque, uniquely owned handles. None is thread-safe. */ typedef struct cusco_executor cusco_executor; @@ -48,6 +48,11 @@ cusco_capabilities cusco_executor_capabilities(const cusco_executor *); * zero). Release it exactly once with cusco_executor_tokens_free. */ cusco_status cusco_executor_tokenize(cusco_executor *, const char *, int32_t ** tokens, size_t * count); void cusco_executor_tokens_free(int32_t * tokens); +/* Converts one token to an owned UTF-8 byte sequence. Release it exactly once + * with cusco_executor_piece_free. */ +cusco_status cusco_executor_token_to_piece( + cusco_executor *, int32_t token, char ** piece, size_t * size); +void cusco_executor_piece_free(char * piece); /* Mutates executor state. Input tokens are borrowed for the duration of the call. */ cusco_status cusco_executor_decode(cusco_executor *, const int32_t *, size_t, cusco_decode_result *); diff --git a/native/shim/cusco_executor.cpp b/native/shim/cusco_executor.cpp index c799a35..9bb441e 100644 --- a/native/shim/cusco_executor.cpp +++ b/native/shim/cusco_executor.cpp @@ -5,6 +5,7 @@ #include #include #include +#include #include struct cusco_checkpoint { @@ -190,6 +191,51 @@ void cusco_executor_tokens_free(int32_t * tokens) { delete[] tokens; } +cusco_status cusco_executor_token_to_piece( + cusco_executor * executor, + int32_t token, + char ** out, + size_t * size) try { + if (!executor || !out || !size) { + return CUSCO_INVALID; + } + *out = nullptr; + *size = 0; + std::string piece; + if (is_mock(executor)) { + piece = std::to_string(token); + } else { + int32_t required = + llama_token_to_piece(executor->vocab, token, nullptr, 0, 0, true); + if (required >= 0) { + return CUSCO_BACKEND; + } + piece.resize(static_cast(-required)); + const int32_t written = llama_token_to_piece( + executor->vocab, token, piece.data(), piece.size(), 0, true); + if (written < 0) { + return CUSCO_BACKEND; + } + piece.resize(static_cast(written)); + } + auto * bytes = new (std::nothrow) char[piece.size()]; + if (!bytes && !piece.empty()) { + return CUSCO_NOMEM; + } + memcpy(bytes, piece.data(), piece.size()); + *out = bytes; + *size = piece.size(); + return CUSCO_OK; +} catch (const std::bad_alloc &) { + return CUSCO_NOMEM; +} catch (...) { + return CUSCO_BACKEND; +} + +void cusco_executor_piece_free(char * piece) { + delete[] piece; +} + cusco_status cusco_executor_decode( cusco_executor * executor, const int32_t * tokens, From 32eaf250e4d9ea43eab06e0694a35fc1cd3ca250 Mon Sep 17 00:00:00 2001 From: Streaky Date: Wed, 5 Aug 2026 04:56:12 +0100 Subject: [PATCH 2/6] Add real-model server inference report --- README.md | 19 +++ compose.yaml | 18 +++ docs/dump/progress.md | 227 +++++++++++++++++++++++++++++++ tools/report-server-inference.py | 150 ++++++++++++++++++++ tools/server-inference-report.sh | 20 +++ 5 files changed, 434 insertions(+) create mode 100644 docs/dump/progress.md create mode 100755 tools/report-server-inference.py create mode 100755 tools/server-inference-report.sh diff --git a/README.md b/README.md index 24ac0a8..ae17ee7 100644 --- a/README.md +++ b/README.md @@ -28,6 +28,25 @@ and a deliberately failed restore preserve the prior binding. The command exits nonzero on any failed assertion and prints a short Markdown report with the prompts, inferred token IDs, checkpoint sizes, and exactness results. +## Run the real-model server report + +To exercise the authenticated Phase 4 HTTP surface with the same validation +GGUF and print a prompt, generated response, streaming assertions, usage, and +end-to-end timing data, run: + +```sh +tools/server-inference-report.sh +``` + +The report is also written to `results/phase4-server.json`. The workflow uses +GPU 1 and port 18082 by default; `CUSCO_GPU_DEVICE_ID`, +`CUSCO_SERVER_REPORT_PORT`, `CUSCO_MODEL_DIR`, and `CUSCO_RESULT_DIR` override +those defaults. Server timing includes model loading and generation, and the +current SSE adapter emits its buffered token events after generation completes, +so the reported first streamed token is an end-to-end observation rather than +decode-only time-to-first-token. + + ## Run the minimal server The unauthenticated development provider is restricted to loopback: diff --git a/compose.yaml b/compose.yaml index e703331..4e24de6 100644 --- a/compose.yaml +++ b/compose.yaml @@ -10,3 +10,21 @@ services: LD_LIBRARY_PATH: "/opt/llama-build/bin" volumes: ["${CUSCO_MODEL_DIR:-./models}:/models:ro", "${CUSCO_RESULT_DIR:-./results}:/results"] deploy: { resources: { reservations: { devices: [{ driver: nvidia, device_ids: ["${CUSCO_GPU_DEVICE_ID:-1}"], capabilities: [gpu] }] } } } + server-report: + build: . + command: + - sh + - -c + - >- + chmod a+rwx /results && + nvidia-smi --query-gpu=uuid,name,compute_cap --format=csv,noheader + > /results/phase4-gpu.csv && + cargo run -p cusco -- serve --listen 0.0.0.0:8080 + --state /results/phase4-state.json + --bearer-token phase4-report-token + environment: + CUSCO_REQUESTED_HOST_GPU: "${CUSCO_GPU_DEVICE_ID:-1}" + LD_LIBRARY_PATH: "/opt/llama-build/bin" + ports: ["127.0.0.1:${CUSCO_SERVER_REPORT_PORT:-18082}:8080"] + volumes: ["${CUSCO_MODEL_DIR:-./models}:/models:ro", "${CUSCO_RESULT_DIR:-./results}:/results"] + deploy: { resources: { reservations: { devices: [{ driver: nvidia, device_ids: ["${CUSCO_GPU_DEVICE_ID:-1}"], capabilities: [gpu] }] } } } diff --git a/docs/dump/progress.md b/docs/dump/progress.md new file mode 100644 index 0000000..24937dc --- /dev/null +++ b/docs/dump/progress.md @@ -0,0 +1,227 @@ +# Cusco progress and capability evidence + +_Last updated: 2026-08-05 on branch `phase-4-minimal-server`._ + +## Executive summary + +Cusco has completed the first four implementation gates in `docs/outline.md`: + +1. **Phase 1 — executor proof:** a real hybrid/recurrent Gemma checkpoint was captured, displaced, copied through host memory, restored, and continued exactly across four logical contexts. +2. **Phase 2 — Rust logical context store:** Rust represents logical contexts, structurally shared token branches, dependency-aware evaluated-prefix mappings, transactional publication, and logical reference accounting. +3. **Phase 3 — tiered physical manager:** Rust accounts for device, pinned-host, and storage representations; guarded capacity; asynchronous transfer ownership; transactional binding transitions; deterministic eviction; and capacity observability. +4. **Phase 4 — minimal server:** an authenticated HTTP service exposes OpenAI completion/chat adapters and native model/context lifecycle APIs over bounded admission, durable opaque contexts, canonical usage, cancellation/deadline handling, and the real llama.cpp executor. + +The strongest executor-boundary result remains the exact Phase 1 checkpoint proof. A second repeatable GPU-backed report now exercises the Phase 4 server end to end: authenticated model registration, a complex 72-token streaming completion through the real Gemma model, OpenAPI route checks, canonical usage, and client/server timing. + +Cusco remains experimental. The Phase 4 server persists its catalog and logical contexts across process restarts, but its SSE events are currently buffered until generation completes, and live executor slots are not yet integrated with the Phase 3 physical manager. It does not establish production readiness, sustained-load performance, or a stable public API. + +## Capability status + +| Capability | Status | Evidence | +|---|---|---| +| Versioned native C ABI around llama.cpp | Demonstrated | The current ABI is version 2; declarations and ownership contracts, including token-to-piece conversion, live in `native/include/cusco_executor.h`. | +| Composite checkpoint capture and restore | Demonstrated on the pinned Gemma artifact | `results/phase1.json` records four restored contexts with non-empty checkpoints. | +| Exact token continuation | Demonstrated | `token_equal: true` for all four contexts in `results/phase1.json`. | +| Bitwise-identical logit continuation | Demonstrated | `logits_equal: true` for all four contexts in `results/phase1.json`; the proof uses exact float-bit comparison rather than tolerance-based comparison. | +| Slot displacement before restore | Demonstrated | The Phase 1 proof replaces each executor's active state with unrelated tokens before restoration and comparison. | +| Device-to-host-to-device checkpoint movement | Demonstrated | `host_round_trip: true` in `results/phase1.json`. | +| Cancellation preserves the active binding | Demonstrated | `cancellation_preserved_binding: true` in `results/phase1.json`. | +| Failed promotion preserves the active binding | Demonstrated | `failed_promotion_preserved_binding: true` in `results/phase1.json`. | +| Concurrent logical-context proof | Demonstrated | Four independently owned executor contexts complete the proof; all four recorded comparisons pass. | +| Locked model identity and digest verification | Implemented and exercised | The proof artifact records the immutable `hf://` identity, expected SHA-256, local path, and file size. The CLI rejects a mismatched digest before executor startup. | +| Persistent/immutable token sequence structure | Implemented and tested | `crates/context-store` uses shared `Arc` sequence nodes; branch tests verify shared prefixes and isolated tails. Here “persistent” describes the immutable data-structure property, not disk persistence. | +| Stable content-derived branch identity | Implemented and tested | Branch IDs are SHA-256-derived from the prior branch hash, token, and fixed-width position encoding, so identities are portable across 32-bit and 64-bit targets. | +| Evaluated-prefix dependency identity | Implemented and tested | Mapping identity includes model epoch, adapter epoch, parent mapping, branch prefix, fixed-width represented end, required components, evaluation parameters, and lineage hash. | +| Longest valid evaluated-prefix lookup | Implemented and tested | Lookup uses an epoch-and-branch index, then checks branch identity, required checkpoint components, and the complete parent dependency chain without materializing token vectors during validation. | +| Transactional mapping publication | Implemented and tested | Publication is prepared separately and committed only if the context's active mapping is unchanged; stale concurrent publication returns `PublicationConflict`. | +| Logical reference accounting and reclamation | Implemented and tested | Catalog, context, and dependent references prevent premature reclamation; dependency-chain cleanup is iterative so reclamation depth does not consume call stack. | +| Per-file Rust coverage enforcement | Active | `tools/coverage.sh` generates workspace coverage and `tools/check_coverage.py` rejects any measured Rust source file below 80% line coverage. | +| Physical device/host/storage tier manager | Implemented and tested | `crates/physical-manager` accounts for residency and guarded capacity, models transfers separately from prepared transitions, commits bindings revision-transactionally, and emits capacity metrics and structured trace events. | +| Durable cross-process context persistence | Implemented and tested | Phase 4 atomically persists the model catalog, logical contexts, branches, revisions, and opaque UUIDs; restart tests verify recovery and non-reuse. Physical representation storage remains separate ownership/accounting metadata. | +| Inference server, scheduler, streaming API | Implemented and GPU-exercised | `cusco-server` provides bounded admission, transition-cost selection, OpenAI completion/chat adapters, native lifecycle routes, shared stream events, cancellation/deadline handling, canonical usage, and checked OpenAPI. `results/phase4-server.json` records a real-model authenticated streaming run. | + +## Phase 1 real-model proof + +### Validated model + +The locally retained proof artifact identifies the external validation model as: + +- Identity: `hf://models/unsloth/gemma-4-E2B-it-GGUF@0314792d7f1f7e229411f620751375812bb9faf2/gemma-4-E2B-it-Q3_K_M.gguf` +- SHA-256: `90293b8cdaf9c973012bf4df8a1e92bde7d74ad66a4fe56cf905ccd563d660c5` +- Size: `3,356,037,216` bytes +- Reported vocabulary: `262,144` tokens +- Reported checkpoint components: global KV, sliding-window attention state, and recurrent state +- llama.cpp source tag: read from the repository-wide `llama.cpp-version.txt` source of truth + +Model weights are external and are not committed to the repository. + +### Recorded integration report + +The machine-readable evidence is `results/phase1.json`. The human-readable report is generated and asserted by `tools/report-inference-proof.py`. + +| Context | Prompt | Input token | Inferred next token | Checkpoint bytes | Checksum | Exact after restore | +|---:|---|---:|---:|---:|---:|:---:| +| 0 | `The capital of France is [0]` | 236,842 | 107 | 166,506 | 15,855,946,795,732,230,871 | PASS | +| 1 | `The capital of France is [1]` | 236,842 | 9,079 | 166,506 | 7,424,184,562,009,432,005 | PASS | +| 2 | `The capital of France is [2]` | 236,842 | 9,079 | 166,506 | 10,193,984,636,459,084,218 | PASS | +| 3 | `The capital of France is [3]` | 236,842 | 9,079 | 166,506 | 17,792,484,377,050,697,218 | PASS | + +Additional observed outcomes: + +- Four independent contexts: **PASS** +- Real model vocabulary available (`262,144` tokens): **PASS** +- Restored tokens identical in every context: **PASS** +- Restored logits bitwise-identical in every context: **PASS** +- Device-to-host-to-device round trip: **PASS** +- Cancellation preserved the prior binding: **PASS** +- Deliberately failed restore preserved the prior binding: **PASS** +- Total proof time: **100,009 ms** +- CUDA device: **NVIDIA GeForce GTX 1080 Ti**, compute capability 6.1 +- Model placement: **36/36 layers offloaded to GPU** + +The differing checksums demonstrate four distinct captured states. The different next-token IDs also show that the report records real prompt-dependent model evaluation rather than only exercising checkpoint serialization. + +### Reproduction command + +With the verified model at `models/gemma-4-e2b-it.gguf`, Docker's NVIDIA runtime configured, and GPU 1 available: + +```sh +tools/inference-integration-test.sh +``` + +The script rebuilds the current executor image, runs the Compose proof service, verifies the configured model SHA-256, confirms GPU visibility, writes `results/phase1.json`, validates every assertion, prints the concise Markdown report above, and exits nonzero on failure. `CUSCO_GPU_DEVICE_ID`, `CUSCO_MODEL_DIR`, and `CUSCO_RESULT_DIR` override the defaults. + +## Phase 2 logical-context-store evidence + +The `cusco-context-store` crate implements: + +- logical context IDs independent of execution slots; +- immutable token sequences with structurally shared prefix nodes; +- cheap branching at any valid token boundary; +- stable SHA-256-derived branch, lineage, and evaluated-prefix identities; +- model and adapter epochs that invalidate incompatible evaluated state; +- composite checkpoint masks for global KV, SWA, and recurrent components; +- parent-linked evaluated-prefix mappings; +- deterministic longest-valid-prefix selection through an epoch-and-branch index; +- two-step prepared/committed publication; +- conflict rejection when another publication wins first; and +- catalog, active-context, and dependent reference counts with iterative reclamation. + +The focused test suite currently has nine tests, including a property test over arbitrary branch prefixes. On 2026-08-05 the following command passed: + +```sh +cargo +1.85.1 test -p cusco-context-store +``` + +Observed result: **9 passed, 0 failed**. + +The tests exercise: + +- exact shared-prefix structure and private branch tails; +- arbitrary prefix preservation; +- atomic publication and epoch invalidation; +- stale concurrent-publication rejection; +- longest-valid-prefix behavior across diverging branches; +- dependency-aware reference reclamation; and +- rejection of incomplete, out-of-bounds, and invalid-parent publications. + +## Phase 3 tiered-physical-manager evidence + +The new `cusco-physical-manager` crate implements: + +- content-independent physical representation IDs and explicit device, pinned-host, and storage residency; +- separate logical, active-binding, transition-reservation, transfer, and growth-reservation ownership; +- exact composite validation across global KV, sliding-window attention, and recurrent components; +- guarded growth and transition capacity plus opportunistic warm-device capacity; +- reference-only, non-destructive, quiesced, and recompute transition classes; +- asynchronous transfer completion whose ownership survives transition cancellation; +- revision-guarded atomic binding commits that preserve the prior binding on failure; +- device-to-host demotion and host/storage-to-device promotion; +- deterministic value/LRU warm-state eviction; and +- capacity metrics and structured prepared, committed, aborted, transfer, promotion, demotion, recomputation, and eviction events. + +The focused suite passed **9 tests, 0 failed**. It covers guarded admission, deterministic eviction, host promotion and demotion, reference-only switching, stale revision rejection, failed and cancelled transitions, detached-transfer capacity reservation, transfer lifetime, recomputation, explicit unbinding/reference release, and composite completeness and boundary agreement. + +## Phase 4 minimal-server evidence + +The `cusco-server` crate and `cusco serve` command implement: + +- authenticated OpenAI-compatible completion, chat, and model-list routes; +- native administrative model lifecycle and durable context/branch/import APIs; +- loopback-safe anonymous development mode and bearer authentication for public listeners; +- bounded admission and deterministic transition-cost slot selection; +- transactional durable context commits with cancellation and deadline rejection; +- canonical usage records and a shared started/token/usage/finished event sequence; +- checked OpenAPI publication; and +- real llama.cpp token generation and token-to-piece conversion through C ABI version 2. + +The reproducible GPU report is: + +```sh +tools/server-inference-report.sh +``` + +The latest observed run used the pinned Gemma artifact on an NVIDIA GeForce GTX 1080 Ti. It generated 72 tokens for a multi-sentence distributed-checkpoint prompt in **2,541 ms** of server-reported end-to-end time and **2,543.12 ms** of client wall time, or **28.335 generated tokens/s** when model loading and prompt evaluation are included. The first buffered SSE token was observed at **2,542.35 ms**. All report assertions passed: anonymous administration was rejected, authenticated registration and lookup succeeded, required OpenAPI routes were present, every generated token had one stream event, terminal usage/finished events were present, and model revision and timing were recorded. + +The machine-readable result is `results/phase4-server.json`. This measurement is an end-to-end integration observation, not a decode-only benchmark: the server currently opens the model per request and emits buffered SSE events after generation. + +## Test and coverage evidence + +The repository's containerized gate is: + +```sh +docker compose build test +docker compose run --rm test +``` + +The most recent observed full gate on 2026-08-05 passed all runnable workspace tests. The GPU-backed integration report was run separately because the coverage service intentionally has no GPU or external model requirement. + +Per-file line coverage reported by that run: + +| Rust source file | Line coverage | +|---|---:| +| `crates/cli/src/main.rs` | 85.82% | +| `crates/context-store/src/lib.rs` | 97.07% | +| `crates/executor/src/lib.rs` | 95.75% | +| `crates/model-registry/src/lib.rs` | 92.86% | +| `crates/physical-manager/src/lib.rs` | 98.82% | +| `crates/server/src/lib.rs` | 95.03% | + +Every measured Rust source file exceeded the required **80%** threshold. + +## Reproducibility and build constraints + +- `llama.cpp-version.txt` is the sole llama.cpp version source and contains a release tag rather than a duplicated commit hash. +- Docker and executor fetch/verification tooling consume that version file. +- Local CUDA builds default to `sm_61` and `sm_70`, covering the GTX 1080 Ti and Tesla GV100-class devices used for development. +- Broader CUDA architecture lists remain configurable through `CUSCO_CUDA_ARCHITECTURES` for release builds. +- CPU-only tests use the same CUDA-capable image without GPU passthrough, with CUDA's stub driver library available for linking/loading. +- Model assets and generated proof results are mounted separately from the image build and are not required for model-free lifecycle and logical-store tests. + +## Evidence index + +| Evidence | Location | +|---|---| +| Phase 1 machine-readable real-model result | `results/phase1.json` | +| Phase 1 integration scenario | `crates/executor/tests/phase1.rs` | +| Proof orchestration and result generation | `crates/cli/src/main.rs` | +| Safe Rust executor/checkpoint API | `crates/executor/src/lib.rs` | +| Native ABI contract | `native/include/cusco_executor.h` | +| Native llama.cpp implementation | `native/shim/cusco_executor.cpp` | +| Logical context store and behavioral tests | `crates/context-store/src/lib.rs` | +| Tiered physical manager and behavioral tests | `crates/physical-manager/src/lib.rs` | +| Runnable real-model report | `tools/inference-integration-test.sh`, `tools/report-inference-proof.py` | +| Phase 4 machine-readable real-model server result | `results/phase4-server.json` | +| Runnable Phase 4 server report | `tools/server-inference-report.sh`, `tools/report-server-inference.py` | +| Minimal server, HTTP adapters, scheduler, persistence, and tests | `crates/server/src/lib.rs` | +| Immutable model registry | `crates/model-registry/src/lib.rs` | +| Containerized proof/test services | `compose.yaml` | +| Per-file coverage gate | `tools/coverage.sh`, `tools/check_coverage.py` | +| llama.cpp release tag source | `llama.cpp-version.txt` | +| Full phased design and acceptance criteria | `docs/outline.md` | + +## Current boundary and next gate + +The evidence now supports the executor-boundary hypothesis, logical-state and transactional physical-tier models, durable minimal-server state, authenticated model/context APIs, and real-model HTTP generation. It does not establish production readiness, sustained-load behavior, true incremental token delivery, or a stable public API. The current server opens a model for each request, buffers generation before emitting SSE events, and does not yet bind Phase 3 physical-manager transitions to live executor slots. + +The next architectural gate is mapped execution if staged measurements justify it; otherwise the immediate work is integrating the minimal scheduler with the physical manager and then compatibility and production hardening. Any such work must preserve the transactional cancellation, capacity, persistence, and checkpoint guarantees already demonstrated. diff --git a/tools/report-server-inference.py b/tools/report-server-inference.py new file mode 100755 index 0000000..5ef2689 --- /dev/null +++ b/tools/report-server-inference.py @@ -0,0 +1,150 @@ +#!/usr/bin/env python3 +import hashlib +import json +import pathlib +import sys +import time +import urllib.error +import urllib.request + +BASE_URL = sys.argv[1] if len(sys.argv) > 1 else "http://127.0.0.1:18082" +RESULT_DIR = pathlib.Path(sys.argv[2] if len(sys.argv) > 2 else "results") +MODEL = pathlib.Path(sys.argv[3] if len(sys.argv) > 3 else "models/gemma-4-e2b-it.gguf") +TOKEN = "phase4-report-token" +PROMPT = ( + "A distributed inference system preserves model execution checkpoints across GPU and host memory. " + "In exactly three concise sentences, explain why transactional restore, cancellation safety, and " + "deterministic capacity accounting matter for correctness." +) + + +def request(method, path, body=None, authenticated=True): + headers = {"content-type": "application/json"} + if authenticated: + headers["authorization"] = f"Bearer {TOKEN}" + encoded = None if body is None else json.dumps(body).encode() + return urllib.request.urlopen( + urllib.request.Request(BASE_URL + path, data=encoded, headers=headers, method=method), + timeout=600, + ) + + +def wait_for_server(): + deadline = time.monotonic() + 180 + while time.monotonic() < deadline: + try: + with request("GET", "/openapi.json", authenticated=False): + return + except (OSError, urllib.error.URLError): + time.sleep(1) + raise SystemExit("server did not become ready within 180 seconds") + + +def digest(path): + value = hashlib.sha256() + with path.open("rb") as source: + while chunk := source.read(1024 * 1024): + value.update(chunk) + return value.hexdigest() + + +wait_for_server() +try: + request("GET", "/native/models", authenticated=False) + unauthorized = False +except urllib.error.HTTPError as error: + unauthorized = error.code == 401 + +model_sha256 = digest(MODEL) +model = { + "id": "gemma-phase4-report", + "revision": "0314792d7f1f7e229411f620751375812bb9faf2", + "path": "/models/gemma-4-e2b-it.gguf", + "sha256": model_sha256, +} +with request("POST", "/native/models", model) as response: + registered = json.load(response) +with request("GET", "/native/models") as response: + listed = json.load(response)["data"] +with request("GET", "/openapi.json", authenticated=False) as response: + openapi = json.load(response) + +started = time.perf_counter() +first_token_ms = None +events = [] +with request( + "POST", + "/v1/completions", + {"model": model["id"], "prompt": PROMPT, "max_tokens": 72, "stream": True}, +) as response: + for raw_line in response: + line = raw_line.decode().strip() + if not line.startswith("data: ") or line == "data: [DONE]": + continue + event = json.loads(line[6:]) + events.append(event) + if event.get("type") == "token" and first_token_ms is None: + first_token_ms = round((time.perf_counter() - started) * 1000, 2) +wall_ms = round((time.perf_counter() - started) * 1000, 2) + +token_events = [event for event in events if event.get("type") == "token"] +usage_events = [event for event in events if event.get("type") == "usage"] +response_text = "".join(event["token"] for event in token_events) +usage = usage_events[-1]["usage"] if usage_events else {} +generated = usage.get("generated_tokens", len(token_events)) +server_ms = usage.get("latency_ms", 0) +artifact = { + "model": registered, + "gpu": (RESULT_DIR / "phase4-gpu.csv").read_text(encoding="utf-8").strip(), + "prompt": PROMPT, + "response": response_text, + "timing": { + "client_wall_ms": wall_ms, + "first_streamed_token_ms": first_token_ms, + "server_end_to_end_ms": server_ms, + "generated_tokens_per_second": round(generated / (server_ms / 1000), 3) if server_ms else None, + }, + "usage": usage, + "stream": { + "event_count": len(events), + "token_event_count": len(token_events), + "event_kinds": [event.get("type") for event in events], + }, + "checks": { + "anonymous admin request rejected": unauthorized, + "registered model listed": any(row["id"] == model["id"] for row in listed), + "checked OpenAPI includes completion and native context routes": all( + route in openapi["paths"] for route in ("/v1/completions", "/native/contexts") + ), + "stream emitted one event per generated token": len(token_events) == generated and generated > 0, + "stream completed with usage": bool(usage_events) and events[-1].get("type") == "finished", + "response contains generated text": bool(response_text.strip()), + "server recorded timing and model identity": server_ms > 0 and usage.get("model_revision") == model["revision"], + }, +} +RESULT_DIR.mkdir(parents=True, exist_ok=True) +output = RESULT_DIR / "phase4-server.json" +output.write_text(json.dumps(artifact, indent=2) + "\n", encoding="utf-8") + +print("# Cusco Phase 4 real-model server report") +print(f"\nModel: `{model['id']}@{model['revision']}`") +print(f"Digest: `{model_sha256}`") +print(f"GPU: `{artifact['gpu']}`") +print("\n## Prompt\n") +print(PROMPT) +print("\n## Response\n") +print(response_text) +print("\n## Timing and usage\n") +print("| Client wall | First streamed token | Server end-to-end | Generated tokens | Tokens/s |") +print("|---:|---:|---:|---:|---:|") +rate = artifact["timing"]["generated_tokens_per_second"] +print(f"| {wall_ms:.2f} ms | {first_token_ms:.2f} ms | {server_ms} ms | {generated} | {rate:.3f} |") +print("\n## Assertions") +failed = [] +for name, passed in artifact["checks"].items(): + print(f"- {'PASS' if passed else 'FAIL'}: {name}") + if not passed: + failed.append(name) +if failed: + raise SystemExit("server report failed: " + ", ".join(failed)) +print(f"\nResult: PASS — authenticated HTTP streaming completed through the real llama.cpp executor. Artifact: `{output}`") diff --git a/tools/server-inference-report.sh b/tools/server-inference-report.sh new file mode 100755 index 0000000..98338e5 --- /dev/null +++ b/tools/server-inference-report.sh @@ -0,0 +1,20 @@ +#!/bin/sh +set -eu + +result_dir=${CUSCO_RESULT_DIR:-./results} +model_dir=${CUSCO_MODEL_DIR:-./models} +port=${CUSCO_SERVER_REPORT_PORT:-18082} +docker compose run --build --rm --no-deps --entrypoint chmod server-report a+rwx /results +mkdir -p "$result_dir" +rm -f "$result_dir/phase4-gpu.csv" "$result_dir/phase4-server.json" "$result_dir/phase4-state.json" +cleanup() { + docker compose stop server-report >/dev/null 2>&1 || true + docker compose rm -f server-report >/dev/null 2>&1 || true +} +trap cleanup EXIT INT TERM + +docker compose up --detach server-report +python3 tools/report-server-inference.py \ + "http://127.0.0.1:$port" \ + "$result_dir" \ + "$model_dir/gemma-4-e2b-it.gguf" From 199f931a6f5cc649d88ccf35bbc12f49da99e337 Mon Sep 17 00:00:00 2001 From: Streaky Date: Wed, 5 Aug 2026 05:11:53 +0100 Subject: [PATCH 3/6] Avoid scheduler priority overflow --- crates/server/src/lib.rs | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index a0222be..0b6ee89 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -254,7 +254,7 @@ pub fn select_slot(candidates: &[SlotCandidate]) -> Option { .saturating_add(c.growth_bytes); ( costs.saturating_sub(c.valid_prefix * 1024), - -c.priority, + std::cmp::Reverse(c.priority), std::cmp::Reverse(c.wait_ms), c.decode_tokens, c.slot, @@ -1043,6 +1043,12 @@ mod tests { }, ]; assert_eq!(select_slot(&c), Some(2)); + let mut low = c[1].clone(); + low.priority = i32::MIN; + let mut high = low.clone(); + high.slot = 3; + high.priority = i32::MAX; + assert_eq!(select_slot(&[low, high]), Some(3)); assert_eq!(select_slot(&[]), None) } #[tokio::test] From 717fa95c638c591615f2892287798eff348c0ae0 Mon Sep 17 00:00:00 2001 From: Streaky Date: Wed, 5 Aug 2026 05:12:17 +0100 Subject: [PATCH 4/6] Stream model digest verification --- crates/server/src/lib.rs | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index 0b6ee89..cee0f02 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -14,6 +14,7 @@ use std::{ collections::HashMap, convert::Infallible, fs, + io::Read, net::SocketAddr, path::{Path, PathBuf}, sync::Arc, @@ -461,8 +462,17 @@ impl Server { } pub fn verify_model(&self, id: &str) -> Result { let model = self.model(id)?; - let bytes = fs::read(&model.path).map_err(state_err)?; - Ok(hex_digest(&bytes) == model.sha256) + let mut file = fs::File::open(model.path).map_err(state_err)?; + let mut digest = Sha256::new(); + let mut buffer = [0; 1024 * 1024]; + loop { + let count = file.read(&mut buffer).map_err(state_err)?; + if count == 0 { + break; + } + digest.update(&buffer[..count]); + } + Ok(format!("{:x}", digest.finalize()) == model.sha256) } pub fn check_update(&self, id: &str, revision: &str) -> Result { Ok(self.model(id)?.revision != revision) From 0958ad30f368bb65568d4ba6f12b32c5c7e859a4 Mon Sep 17 00:00:00 2001 From: Streaky Date: Wed, 5 Aug 2026 05:13:42 +0100 Subject: [PATCH 5/6] Commit implicit contexts transactionally --- crates/server/src/lib.rs | 71 +++++++++++++++++++++++++++++++++------- 1 file changed, 60 insertions(+), 11 deletions(-) diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index cee0f02..f7708d9 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -512,9 +512,9 @@ impl Server { if self.inner.lock().cancelled.remove(request_id).is_some() { return Err(Error::Cancelled); } - let context = match req.context_id { - Some(ref id) => self.context(id)?, - None => self.create_context()?, + let context = match &req.context_id { + Some(id) => Some(self.context(id)?), + None => None, }; let generated = self.engine.generate(&model, &req.prompt, req.max_tokens)?; if self.inner.lock().cancelled.remove(request_id).is_some() { @@ -526,13 +526,32 @@ impl Server { { return Err(Error::Deadline); } - let mut guard = self.inner.lock(); - let stored = guard.durable.contexts.get_mut(&context.id).unwrap(); let input: Vec<_> = req.prompt.split_whitespace().map(str::to_owned).collect(); - stored.tokens.extend(input.iter().cloned()); - stored.tokens.extend(generated.iter().cloned()); - stored.revision += 1; - let revision = stored.revision; + let mut guard = self.inner.lock(); + let (context_id, revision) = if let Some(context) = context { + let stored = guard + .durable + .contexts + .get_mut(&context.id) + .ok_or(Error::ContextNotFound)?; + stored.tokens.extend(input.iter().cloned()); + stored.tokens.extend(generated.iter().cloned()); + stored.revision += 1; + (stored.id.clone(), stored.revision) + } else { + let id = ContextId::new(); + let mut tokens = input.clone(); + tokens.extend(generated.iter().cloned()); + guard.durable.contexts.insert( + id.clone(), + ContextRecord { + id: id.clone(), + revision: 1, + tokens, + }, + ); + (id, 1) + }; drop(guard); self.persist()?; let usage = Usage { @@ -542,13 +561,13 @@ impl Server { cached_tokens: 0, model: model.id, model_revision: model.revision, - context_id: context.id.clone(), + context_id: context_id.clone(), latency_ms: started.elapsed().as_millis(), status: "completed".into(), }; let mut events = vec![StreamEvent::Started { request_id: request_id.into(), - context_id: context.id, + context_id, }]; events.extend( generated @@ -926,6 +945,13 @@ mod tests { .unwrap(); (s, d) } + + struct FailingEngine; + impl InferenceEngine for FailingEngine { + fn generate(&self, _: &ModelRecord, _: &str, _: usize) -> Result, Error> { + Err(Error::State("generation failed".into())) + } + } #[test] fn persistence_branching_and_ids_survive_restart() { let (s, d) = setup(Arc::new(AnonymousAdmin)); @@ -1006,6 +1032,29 @@ mod tests { ), Err(Error::Busy) )); + let contexts_before = s.contexts(); + drop(s); + let failing = Server::open( + d.join("state.json"), + Arc::new(AnonymousAdmin), + Arc::new(FailingEngine), + ) + .unwrap(); + assert!(matches!( + failing.infer( + "failed", + InferRequest { + model: "m".into(), + prompt: "x".into(), + max_tokens: 1, + context_id: None, + deadline_ms: None, + priority: 0, + }, + ), + Err(Error::State(_)) + )); + assert_eq!(failing.contexts(), contexts_before); fs::remove_dir_all(d).unwrap() } #[test] From b6e2b92a43f227596b080fff01296853ad4b1ceb Mon Sep 17 00:00:00 2001 From: Streaky Date: Wed, 5 Aug 2026 05:14:35 +0100 Subject: [PATCH 6/6] Limit digest helper to tests --- crates/server/src/lib.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index f7708d9..e911b6c 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -597,6 +597,7 @@ impl Server { fn state_err(error: impl std::fmt::Display) -> Error { Error::State(error.to_string()) } +#[cfg(test)] fn hex_digest(bytes: &[u8]) -> String { format!("{:x}", Sha256::digest(bytes)) }