From bae084cd76ad7fe59dda27a8d89c064f289861c8 Mon Sep 17 00:00:00 2001 From: dtoro Date: Tue, 28 Apr 2026 11:53:11 +0200 Subject: [PATCH] Implement Sophia MVP stages 4-7 (introspection, queries, persistence, polish) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Brings the system from "engrams cluster + form synapses" to a complete end-to-end demo: ingest text, watch it cluster, ask questions, restart with state intact. - Stage 4: birth introspection (taxonomy/goals/open_questions via LLM, bounded by the global parallel-op semaphore), per-engram memory log, click-to-inspect side panel. - Stage 5: queries as conversations. POST /api/galaxy/:id/query embeds the question, materializes a pinned Query-Engram at the donut center, runs broadcast retrieval (global cosine scan + 1-hop synaptic expansion with attenuation) and fans out responder LLM calls. The integrator runs every 2s on accumulated snippets and streams the refining answer back over SSE; responders briefly transition to Conversing on the WS bus so the right dots light up. - Stage 6: snapshot persistence. sled-backed store keyed by galaxy id, JSON-encoded values (bincode chokes on internally-tagged enums like Manifest/MemoryKind), 60s periodic snapshot task, hydrate-on-boot, DELETE /api/galaxy/:id wired through. State survives kill -9. - Stage 7: HUD additions (sim ticks/sec, LLM queue depth, FPS) via a new GET /api/stats polled at 1Hz. `sophia demo` subcommand boots the server then auto-ingests a 50-paragraph corpus baked into the binary with include_str!. README quickstart added. Token caps for query_responder/integrator bumped (gemma-4-e4b is a thinking model — output budget must cover hidden reasoning + visible answer, otherwise content comes back empty). Pinned engrams skip physics; their tick scheduling is also skipped at materialization so they stay perfectly still at the donut center. Co-Authored-By: Claude Opus 4.7 (1M context) --- .gitignore | 2 + Cargo.lock | 172 ++++++- Cargo.toml | 8 + README.md | 156 ++++++ assets/demo_corpus.txt | 50 ++ config.toml | 29 +- crates/sophia-bin/Cargo.toml | 3 + crates/sophia-bin/src/main.rs | 222 +++++++- crates/sophia-core/src/dto.rs | 37 ++ crates/sophia-core/src/engram.rs | 17 +- crates/sophia-core/src/galaxy.rs | 5 +- crates/sophia-core/src/lib.rs | 6 + crates/sophia-core/src/memory.rs | 52 ++ crates/sophia-core/src/persistence.rs | 31 ++ crates/sophia-core/src/query.rs | 45 ++ crates/sophia-llm/src/client.rs | 14 + crates/sophia-llm/src/introspect.rs | 90 ++++ crates/sophia-llm/src/lib.rs | 4 + crates/sophia-llm/src/query.rs | 92 ++++ crates/sophia-server/Cargo.toml | 8 +- crates/sophia-server/src/lib.rs | 235 ++++++++- crates/sophia-server/src/query.rs | 393 ++++++++++++++ crates/sophia-sim/src/handle.rs | 703 +++++++++++++++++++++++++- crates/sophia-sim/src/lib.rs | 2 +- crates/sophia-sim/src/physics.rs | 8 + crates/sophia-sim/src/scheduler.rs | 15 +- crates/sophia-sim/src/world.rs | 78 ++- crates/sophia-store/Cargo.toml | 3 + crates/sophia-store/src/lib.rs | 112 +++- web/index.html | 145 ++++++ web/src/engram_mesh.ts | 40 ++ web/src/inspector.ts | 179 +++++++ web/src/main.ts | 31 ++ web/src/query_panel.ts | 128 +++++ web/src/scene.ts | 26 +- web/src/ws_client.ts | 54 ++ 36 files changed, 3137 insertions(+), 58 deletions(-) create mode 100644 README.md create mode 100644 assets/demo_corpus.txt create mode 100644 crates/sophia-core/src/memory.rs create mode 100644 crates/sophia-core/src/persistence.rs create mode 100644 crates/sophia-core/src/query.rs create mode 100644 crates/sophia-llm/src/introspect.rs create mode 100644 crates/sophia-llm/src/query.rs create mode 100644 crates/sophia-server/src/query.rs create mode 100644 web/src/inspector.ts create mode 100644 web/src/query_panel.ts diff --git a/.gitignore b/.gitignore index 91ff653..75d52d3 100644 --- a/.gitignore +++ b/.gitignore @@ -4,5 +4,7 @@ web/dist **/.DS_Store *.log sled_data/ +/data/ +.claude/ config.local.toml .env diff --git a/Cargo.lock b/Cargo.lock index 40afe28..734f98d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -122,6 +122,12 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" +[[package]] +name = "bitflags" +version = "1.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a" + [[package]] name = "bitflags" version = "2.11.1" @@ -212,6 +218,30 @@ dependencies = [ "libc", ] +[[package]] +name = "crc32fast" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9481c1c90cbf2ac953f07c8d4a58aa3945c425b7185c9154d67a65e4230da511" +dependencies = [ + "cfg-if", +] + +[[package]] +name = "crossbeam-epoch" +version = "0.9.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e" +dependencies = [ + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-utils" +version = "0.8.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28" + [[package]] name = "crunchy" version = "0.2.4" @@ -365,6 +395,16 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "fs2" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9564fc758e15025b46aa6643b1b77d047d1a56a1aea6e01002ac0c7026876213" +dependencies = [ + "libc", + "winapi", +] + [[package]] name = "futures-channel" version = "0.3.32" @@ -425,6 +465,15 @@ dependencies = [ "slab", ] +[[package]] +name = "fxhash" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c31b6d751ae2c7f11320402d34e41349dd1016f8d5d45e48c4312bc8625af50c" +dependencies = [ + "byteorder", +] + [[package]] name = "generator" version = "0.8.8" @@ -790,6 +839,15 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0175f63815ce00183bf755155ad0cb48c65226c5d17a724e369c25418d2b7699" +[[package]] +name = "instant" +version = "0.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e0242819d153cba4b4b05a5a8f2a7e9bbf97b6055b2a002b395c96b5ff3c0222" +dependencies = [ + "cfg-if", +] + [[package]] name = "ipnet" version = "2.12.0" @@ -996,7 +1054,7 @@ version = "0.10.78" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f38c4372413cdaaf3cc79dd92d29d7d9f5ab09b51b10dded508fb90bb70b9222" dependencies = [ - "bitflags", + "bitflags 2.11.1", "cfg-if", "foreign-types", "libc", @@ -1043,6 +1101,17 @@ dependencies = [ "num-traits", ] +[[package]] +name = "parking_lot" +version = "0.11.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d17b78036a60663b797adeaee46f5c9dfebb86948d1255007a1d6be0271ff99" +dependencies = [ + "instant", + "lock_api", + "parking_lot_core 0.8.6", +] + [[package]] name = "parking_lot" version = "0.12.5" @@ -1050,7 +1119,21 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a" dependencies = [ "lock_api", - "parking_lot_core", + "parking_lot_core 0.9.12", +] + +[[package]] +name = "parking_lot_core" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "60a2cfe6f0ad2bfc16aefa463b497d5c7a5ecd44a23efa72aa342d90177356dc" +dependencies = [ + "cfg-if", + "instant", + "libc", + "redox_syscall 0.2.16", + "smallvec", + "winapi", ] [[package]] @@ -1061,7 +1144,7 @@ checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" dependencies = [ "cfg-if", "libc", - "redox_syscall", + "redox_syscall 0.5.18", "smallvec", "windows-link", ] @@ -1166,13 +1249,22 @@ dependencies = [ "getrandom 0.2.17", ] +[[package]] +name = "redox_syscall" +version = "0.2.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb5a58c1855b4b6819d59012155603f0b22ad30cad752600aadfcb695265519a" +dependencies = [ + "bitflags 1.3.2", +] + [[package]] name = "redox_syscall" version = "0.5.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" dependencies = [ - "bitflags", + "bitflags 2.11.1", ] [[package]] @@ -1255,7 +1347,7 @@ version = "1.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190" dependencies = [ - "bitflags", + "bitflags 2.11.1", "errno", "libc", "linux-raw-sys", @@ -1328,7 +1420,7 @@ version = "3.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b7f4bc775c73d9a02cde8bf7b2ec4c9d12743edf609006c7facc23998404cd1d" dependencies = [ - "bitflags", + "bitflags 2.11.1", "core-foundation 0.10.1", "core-foundation-sys", "libc", @@ -1468,6 +1560,22 @@ version = "0.4.12" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" +[[package]] +name = "sled" +version = "0.34.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f96b4737c2ce5987354855aed3797279def4ebf734436c6aa4552cf8e169935" +dependencies = [ + "crc32fast", + "crossbeam-epoch", + "crossbeam-utils", + "fs2", + "fxhash", + "libc", + "log", + "parking_lot 0.11.2", +] + [[package]] name = "smallvec" version = "1.15.1" @@ -1489,11 +1597,14 @@ name = "sophia-bin" version = "0.0.1" dependencies = [ "anyhow", + "reqwest", "serde", + "serde_json", "sophia-core", "sophia-llm", "sophia-server", "sophia-sim", + "sophia-store", "tokio", "toml", "tracing", @@ -1538,7 +1649,9 @@ dependencies = [ "sophia-core", "sophia-llm", "sophia-sim", + "sophia-store", "tokio", + "tokio-stream", "tower", "tower-http", "tracing", @@ -1565,7 +1678,10 @@ version = "0.0.1" dependencies = [ "anyhow", "serde", + "serde_json", + "sled", "sophia-core", + "thiserror", "tracing", ] @@ -1624,7 +1740,7 @@ version = "0.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a13f3d0daba03132c0aa9767f98351b3488edc2c100cda2d2ec2b04f3d8d3c8b" dependencies = [ - "bitflags", + "bitflags 2.11.1", "core-foundation 0.9.4", "system-configuration-sys", ] @@ -1700,7 +1816,7 @@ dependencies = [ "bytes", "libc", "mio", - "parking_lot", + "parking_lot 0.12.5", "pin-project-lite", "signal-hook-registry", "socket2", @@ -1739,6 +1855,18 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-stream" +version = "0.1.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32da49809aab5c3bc678af03902d4ccddea2a87d028d86392a4b1560c6906c70" +dependencies = [ + "futures-core", + "pin-project-lite", + "tokio", + "tokio-util", +] + [[package]] name = "tokio-tungstenite" version = "0.24.0" @@ -1827,7 +1955,7 @@ version = "0.6.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d4e6559d53cc268e5031cd8429d05415bc4cb4aefc4aa5d6cc35fbf5b924a1f8" dependencies = [ - "bitflags", + "bitflags 2.11.1", "bytes", "futures-core", "futures-util", @@ -2166,7 +2294,7 @@ version = "0.244.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "47b807c72e1bac69382b3a6fb3dbe8ea4c0ed87ff5629b8685ae6b9a611028fe" dependencies = [ - "bitflags", + "bitflags 2.11.1", "hashbrown 0.15.5", "indexmap", "semver", @@ -2182,6 +2310,28 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "winapi" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419" +dependencies = [ + "winapi-i686-pc-windows-gnu", + "winapi-x86_64-pc-windows-gnu", +] + +[[package]] +name = "winapi-i686-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" + +[[package]] +name = "winapi-x86_64-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" + [[package]] name = "windows-link" version = "0.2.1" @@ -2372,7 +2522,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9d66ea20e9553b30172b5e831994e35fbde2d165325bec84fc43dbf6f4eb9cb2" dependencies = [ "anyhow", - "bitflags", + "bitflags 2.11.1", "indexmap", "log", "serde", diff --git a/Cargo.toml b/Cargo.toml index 0084dbc..512f36a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -56,11 +56,19 @@ rand = { version = "0.8", features = ["small_rng"] } # Async helpers futures-util = { version = "0.3", default-features = false, features = ["std", "sink"] } +tokio-stream = { version = "0.1", features = ["sync"] } bytes = "1" # Config toml = "0.8" +# Persistence (Stage 6) — embedded KV with serde_json-encoded values. We +# pick JSON over bincode because `Manifest` and `MemoryKind` are +# internally-tagged enums (`#[serde(tag = "kind")]`), which require a +# self-describing format on the wire. Snapshots are 60-second cadence; +# JSON's verbosity isn't a real cost at MVP scale. +sled = "0.34" + [profile.release] lto = "thin" codegen-units = 1 diff --git a/README.md b/README.md new file mode 100644 index 0000000..30b1799 --- /dev/null +++ b/README.md @@ -0,0 +1,156 @@ +# Sophia + +A knowledge engine where each idea is an autonomous agent — an *Engram* — that +moves, resonates with similar ideas, forms synapses with them, and answers +questions by gathering its own perspective into a single integrated reply. + +The whole system is a 3D simulation you can watch: ingested text becomes glowing +dots that fly out from the centre of a torus, drift toward similar peers, and +light up when a query's broadcast wave touches them. + +``` +ingest text → embed → spawn engram at center + ↓ + spiral into the tube + ↓ + gravity pulls toward similar engrams + ↓ + co-resonating pairs form synapses + ↓ + query broadcast → cosine + 1-hop synaptic → + N responders contribute snippets → + streamed integrated answer +``` + +## Stack + +- **Rust** workspace — domain types, event-driven sim, axum HTTP/WS server, + sled persistence, LM Studio client +- **Three.js + WebGL** — additive-blended particle cloud, post-processing + bloom, custom GLSL for per-engram pulse and hue variance +- **LM Studio** — local LLM (`gemma-4-e4b`) for introspection + query + responders + integrator; embeddings via `nomic-embed-text-v1.5` + +The full design lives in [`docs/system-analysis.md`](docs/system-analysis.md). + +## Quickstart + +You need: + +- Rust 1.80+ (via `rustup`) +- Node 20+ + `npm` +- LM Studio running locally with both `google/gemma-4-e4b` and + `text-embedding-nomic-embed-text-v1.5` loaded; copy your developer API + token from LM Studio into `config.local.toml`: + ```toml + [lm_studio] + api_token = "lms-..." + ``` + +Build the frontend once: + +```bash +(cd web && npm install && npm run build) +``` + +Then run the auto-ingesting demo: + +```bash +cargo run --release -- demo +# → http://127.0.0.1:7777 +``` + +The bin boots the server, hydrates any saved galaxies (`data/`), then spawns a +background task that ingests `assets/demo_corpus.txt` (~50 paragraphs across +cooking, ML, and history). Open the URL and you should see three jewel-tone +clusters form within ~30 s. + +Without `demo`, the server boots empty and you ingest manually via the bottom- +right text area, or: + +```bash +curl -X POST -H 'content-type: application/json' \ + -d '{"texts":["Pasta carbonara uses ...","..."]}' \ + http://127.0.0.1:7777/api/galaxy//ingest +``` + +## Asking a question + +Use the bottom-left **ask** box. The query materialises as a pinned engram at +the donut center; relevant engrams light up briefly and contribute a snippet; +the integrated answer streams back over SSE. Try: + +- `What did we ingest about pasta?` +- `How does gradient descent work?` +- `When did the Berlin Wall fall?` + +## API + +| route | what | +| --- | --- | +| `GET /healthz` | process status + LM Studio reachability | +| `GET /api/stats` | live HUD numbers (`ticks_per_sec`, `llm_queue_depth`, …) | +| `GET /api/galaxy` | list galaxies | +| `POST /api/galaxy` | `{name}` → new galaxy | +| `DELETE /api/galaxy/:id` | remove from sim + disk | +| `POST /api/galaxy/:id/seed?n=200` | spawn N synthetic engrams | +| `POST /api/galaxy/:id/ingest` | `{texts: [...]}` | +| `POST /api/galaxy/:id/resize` | `{major_radius, minor_radius}` | +| `POST /api/galaxy/:id/query` | `{text}` → `{query_id}` | +| `GET /api/query/:id` | latest snapshot | +| `GET /api/query/:id/stream` | SSE updates of running answer | +| `GET /api/galaxy/:gid/engrams/:eid` | inspector payload (manifest, memories…) | +| `WS /ws/galaxy/:id/events` | event stream (positions, synapses, query lifecycle) | + +## Layout + +``` +crates/ + sophia-core/ # domain types, no I/O + sophia-sim/ # event scheduler, physics, kiddo spatial index, broadcast + sophia-llm/ # LM Studio client + caveman prompts (introspection, query) + sophia-store/ # sled+JSON snapshot persistence + sophia-server/ # axum HTTP + WS, query orchestrator, SSE answer stream + sophia-bin/ # main; loads config, hydrates store, runs demo subcommand +web/ # Vite + TypeScript + Three.js +assets/ # demo corpus +data/ # sled DB (gitignored) +docs/ # design + plan +``` + +## Configuration + +[`config.toml`](config.toml) is versioned and contains defaults; copy what you +want to override into `config.local.toml` (gitignored — for the LM Studio +token, or for tweaking the snapshot interval, etc.). Keys: + +- `[server]` — host, port, static dir +- `[storage]` — `data_dir`, `snapshot_interval_secs` +- `[lm_studio]` — base URL, model names, API token, parallel-op cap +- `[galaxy_defaults]` — `major_radius`, `minor_radius` +- `[token_caps]` — per-call input/output caps for each LLM use site + (note: `gemma-4-e4b` is a thinking model; output caps must budget for + hidden reasoning *plus* visible content) + +## Persistence + +Each galaxy snapshots to one row in a sled DB at `data/`, JSON-encoded, every +60 s. Kill the server with `Ctrl-C` or `kill -9`; on restart it loads the +latest snapshot and resumes — same engram positions, slates, manifests, +memories, and synapses. + +## Stage map + +The implementation grew incrementally; each stage produced a runnable demo: + +0. Scaffolding — workspace + black canvas +1. Engrams in space — physics, scheduler, position WS stream +2. Slate + LM Studio — embed, ingest, manifest +3. Gravity + clustering + synapses +4. Birth introspection (taxonomy/goals) + memories + click-to-inspect +5. Queries as conversations — broadcast, responders, integrator, SSE +6. Persistence + replay-on-boot +7. Polish — HUD (FPS, ticks/sec, LLM queue), `demo` subcommand + +Each is verifiable end-to-end and the full plan is in +[`docs/system-analysis.md`](docs/system-analysis.md). diff --git a/assets/demo_corpus.txt b/assets/demo_corpus.txt new file mode 100644 index 0000000..d548778 --- /dev/null +++ b/assets/demo_corpus.txt @@ -0,0 +1,50 @@ +Pasta carbonara uses guanciale, eggs, pecorino romano, and freshly cracked black pepper. +Spaghetti aglio e olio combines garlic, olive oil, chili flakes, and parsley over al dente pasta. +Risotto requires arborio rice, slowly added warm broth, and constant stirring to release starch. +Pesto alla genovese blends basil, pine nuts, garlic, parmesan, pecorino, and olive oil. +Cacio e pepe is just pasta, pecorino, and pepper — emulsion comes from starchy pasta water. +Lasagne bolognese layers ragu, bechamel, and pasta sheets between melted parmesan. +Amatriciana sauces guanciale and tomato together with a touch of pecorino. +Gnocchi alla romana use semolina baked with butter and cheese instead of potato. +Tiramisu layers savoiardi soaked in espresso with mascarpone whipped with eggs and sugar. +Sourdough bread relies on a wild yeast starter and slow bulk fermentation for flavor. +Bechamel sauce is built from a roux of butter and flour cooked into milk. +Caramelization happens to sugars above 160 C; the Maillard reaction needs proteins and amines. +Sushi rice is seasoned with vinegar, sugar, and salt while still warm so seasoning soaks evenly. +Korean kimchi ferments napa cabbage with chili paste, garlic, ginger, and salted shrimp. +French croissants get their flake from laminated layers of cold butter folded into yeasted dough. +Backpropagation computes gradients of a loss with respect to neural network weights via the chain rule. +Gradient descent updates model parameters in the direction that reduces the loss the most. +Stochastic gradient descent samples mini-batches so each step costs less and adds useful noise. +Adam adapts a per-parameter learning rate using running moments of the gradients. +Convolutional neural networks share weights across spatial locations to exploit translation invariance. +Recurrent networks process sequences by feeding state from one step into the next. +Transformers attend over a whole sequence at once instead of recurring through it. +Self-attention computes a weighted sum of values where weights are softmax of query-key dot products. +Layer normalization stabilizes training by normalizing activations along the feature dimension. +Dropout randomly zeroes activations during training as a regularizer. +Residual connections let very deep networks train by giving gradients an additive path back. +Word embeddings map discrete tokens to dense vectors learned end-to-end with the task. +A loss function turns predictions and targets into a single scalar to minimize. +Cross-entropy loss compares a predicted distribution to a one-hot target. +Reinforcement learning learns a policy by trial-and-error rewards in an environment. +Markov decision processes describe RL with states, actions, transitions, and rewards. +Q-learning estimates the value of taking each action in each state. +Policy gradient methods directly adjust the policy to maximize expected reward. +The Roman Empire fell in 476 CE when Romulus Augustulus, the last western emperor, was deposed. +Constantinople fell to the Ottomans under Mehmed II in 1453, ending the Byzantine Empire. +The Magna Carta of 1215 limited the power of the English monarchy and protected baronial rights. +The printing press, perfected by Gutenberg around 1440, accelerated the spread of literacy in Europe. +The Treaty of Westphalia in 1648 ended the Thirty Years War and established sovereign-state diplomacy. +The French Revolution began in 1789 with the storming of the Bastille on July 14. +The Industrial Revolution started in late-eighteenth-century Britain with mechanized textile production. +The American Civil War ran from 1861 to 1865 over slavery, states' rights, and union. +World War I lasted from 1914 to 1918 and was triggered by the assassination of Archduke Franz Ferdinand. +The Russian Revolution of 1917 replaced the tsarist autocracy with Bolshevik rule. +The Great Depression began with the 1929 Wall Street Crash and lasted through the 1930s. +World War II ended in 1945 after Germany surrendered in May and Japan in September. +The United Nations was founded in 1945 in San Francisco to prevent another global war. +The Cold War divided the world between US and Soviet spheres from roughly 1947 to 1991. +The Berlin Wall fell on November 9, 1989, symbolizing the end of the Cold War in Europe. +The Apollo 11 mission landed humans on the Moon for the first time on July 20, 1969. +The European Union was established by the Maastricht Treaty in 1993, succeeding the EEC. diff --git a/config.toml b/config.toml index 9037c13..81ea2c2 100644 --- a/config.toml +++ b/config.toml @@ -6,6 +6,14 @@ host = "127.0.0.1" port = 7777 static_dir = "web/dist" +# Persistence (Stage 6). `data_dir` is a directory; sled creates and manages +# files inside it. Snapshots are written every `snapshot_interval_secs` +# while the server runs; on boot any existing galaxy snapshots are loaded +# and re-instantiated in the sim. +[storage] +data_dir = "data" +snapshot_interval_secs = 60 + [lm_studio] # LM Studio's OpenAI-compatible REST endpoint. base_url = "http://127.0.0.1:1234/v1" @@ -35,16 +43,21 @@ major_radius = 100.0 minor_radius = 30.0 # Caveman token caps per §13.5. +# Note: gemma-4-e4b is a *thinking* model — it emits a hidden reasoning +# stream before its visible content. Every `_out` cap therefore needs to +# budget for the hidden reasoning *plus* the visible answer. Numbers +# verified against logs: a 150-cap responder hit "response had no content" +# every time (all budget eaten by reasoning). [token_caps] -introspection_in = 400 -introspection_out = 200 +introspection_in = 600 +introspection_out = 900 peer_msg_in = 100 peer_msg_out = 100 synthesis_in = 300 synthesis_out = 80 -query_responder_in = 200 -query_responder_out = 150 -query_integrator_in = 800 -query_integrator_out = 300 -hard_ceiling_in = 1024 -hard_ceiling_out = 400 +query_responder_in = 400 +query_responder_out = 700 +query_integrator_in = 1200 +query_integrator_out = 1200 +hard_ceiling_in = 1500 +hard_ceiling_out = 1500 diff --git a/crates/sophia-bin/Cargo.toml b/crates/sophia-bin/Cargo.toml index 8182a4b..26cd08d 100644 --- a/crates/sophia-bin/Cargo.toml +++ b/crates/sophia-bin/Cargo.toml @@ -15,10 +15,13 @@ sophia-core = { workspace = true } sophia-server = { workspace = true } sophia-sim = { workspace = true } sophia-llm = { workspace = true } +sophia-store = { workspace = true } anyhow = { workspace = true } tokio = { workspace = true } tracing = { workspace = true } tracing-subscriber = { workspace = true } serde = { workspace = true } +serde_json = { workspace = true } toml = { workspace = true } +reqwest = { workspace = true } diff --git a/crates/sophia-bin/src/main.rs b/crates/sophia-bin/src/main.rs index 4d15c1e..0e99256 100644 --- a/crates/sophia-bin/src/main.rs +++ b/crates/sophia-bin/src/main.rs @@ -1,19 +1,31 @@ -//! Sophia binary. Reads `config.toml`, initializes tracing, starts the server. +//! Sophia binary. Reads `config.toml`, initializes tracing, opens the +//! persistence store, hydrates any saved galaxies into the sim, and starts +//! the server. A background task snapshots the sim to disk on a fixed +//! cadence (Stage 6). +//! +//! `sophia demo` (Stage 7) — same boot path, plus a background task that +//! waits for the server to be reachable then ingests a curated corpus +//! (compiled into the binary). Useful for screen recordings and first-run +//! demos: a single command takes you from cold boot to a populated galaxy. use std::path::{Path, PathBuf}; +use std::time::Duration; use anyhow::{Context, Result}; use serde::Deserialize; use sophia_core::GalaxyShape; use sophia_llm::{LmStudioClient, LmStudioConfig, TokenCapsAll}; use sophia_server::{serve, ServerConfig}; +use sophia_sim::SimHandle; +use sophia_store::Store; use tracing_subscriber::{fmt, prelude::*, EnvFilter}; #[derive(Debug, Deserialize)] struct Config { server: ServerSection, + #[serde(default)] + storage: StorageSection, lm_studio: LmStudioConfig, - #[allow(dead_code)] // wired in to LLM call sites in Stages 4+ token_caps: TokenCapsAll, #[serde(default)] galaxy_defaults: GalaxyDefaultsSection, @@ -26,6 +38,21 @@ struct ServerSection { static_dir: String, } +#[derive(Debug, Deserialize)] +struct StorageSection { + data_dir: String, + snapshot_interval_secs: u64, +} + +impl Default for StorageSection { + fn default() -> Self { + Self { + data_dir: "data".to_string(), + snapshot_interval_secs: 60, + } + } +} + #[derive(Debug, Deserialize)] struct GalaxyDefaultsSection { major_radius: f32, @@ -92,11 +119,175 @@ fn load_config() -> Result { Ok(cfg) } +/// Hydrate the sim from disk. Returns the count of galaxies restored. +/// If the store is empty, materializes a fresh "default" galaxy and writes +/// the first snapshot so a subsequent kill+restart already finds it. +async fn hydrate_or_seed(sim: &SimHandle, store: &Store) -> Result { + let saved = store.list_galaxies().context("listing stored galaxies")?; + if saved.is_empty() { + let info = sim.create_galaxy("default".to_string()).await?; + tracing::info!("no saved galaxies — created default {:?}", info.id); + // Persist the empty galaxy now so the very next restart rehydrates + // (with the same id) instead of creating a fresh one — keeps the + // browser's bookmarked galaxy id stable across restarts. + let snaps = sim.snapshot_all().await?; + for snap in &snaps { + store.save_galaxy(snap)?; + } + return Ok(0); + } + let n = saved.len(); + for snap in saved { + let id = snap.galaxy.id; + let engrams = snap.engrams.len(); + let synapses = snap.synapses.len(); + match sim.hydrate_galaxy(snap).await { + Ok(info) => { + tracing::info!( + "hydrated galaxy {:?} ({}, {} engrams, {} synapses)", + info.id, + info.name, + engrams, + synapses + ); + } + Err(e) => { + tracing::warn!("hydration failed for {:?}: {e}", id); + } + } + } + Ok(n) +} + +/// Stage 7: corpus to ingest when invoked as `sophia demo`. Compiled into +/// the binary so the command works from any cwd. +const DEMO_CORPUS: &str = include_str!("../../../assets/demo_corpus.txt"); + +/// Stage 7 demo task. Polls `/healthz` until the server is reachable, then +/// posts the curated corpus into the first galaxy. Logs progress; failures +/// are non-fatal — the server stays up either way. +fn spawn_demo_ingest(host: String, port: u16) { + tokio::spawn(async move { + let base = format!("http://{host}:{port}"); + let client = match reqwest::Client::builder() + .timeout(Duration::from_secs(60)) + .build() + { + Ok(c) => c, + Err(e) => { + tracing::warn!("demo: reqwest client failed: {e}"); + return; + } + }; + + // Wait for server reachable. Cap at 30 s so a misconfigured boot + // doesn't loop forever. + let deadline = std::time::Instant::now() + Duration::from_secs(30); + let mut reachable = false; + while std::time::Instant::now() < deadline { + if client.get(format!("{base}/healthz")).send().await.is_ok() { + reachable = true; + break; + } + tokio::time::sleep(Duration::from_millis(250)).await; + } + if !reachable { + tracing::warn!("demo: server never became reachable, aborting ingest"); + return; + } + + // Pick the first galaxy (the auto-created or just-hydrated default). + let galaxies: Vec = match client + .get(format!("{base}/api/galaxy")) + .send() + .await + .and_then(|r| r.error_for_status()) + { + Ok(r) => match r.json().await { + Ok(g) => g, + Err(e) => { + tracing::warn!("demo: parsing galaxies failed: {e}"); + return; + } + }, + Err(e) => { + tracing::warn!("demo: listing galaxies failed: {e}"); + return; + } + }; + let Some(first) = galaxies.first() else { + tracing::warn!("demo: no galaxies to ingest into"); + return; + }; + let Some(gid) = first.get("id").and_then(|v| v.as_str()) else { + tracing::warn!("demo: galaxy entry missing id"); + return; + }; + + let texts: Vec<&str> = DEMO_CORPUS + .lines() + .map(str::trim) + .filter(|l| !l.is_empty()) + .collect(); + tracing::info!("demo: ingesting {} paragraphs into galaxy {}", texts.len(), gid); + + let body = serde_json::json!({ "texts": texts }); + match client + .post(format!("{base}/api/galaxy/{gid}/ingest")) + .json(&body) + .send() + .await + .and_then(|r| r.error_for_status()) + { + Ok(_) => { + tracing::info!( + "demo: ingest succeeded — open {base} to watch the cluster form" + ); + } + Err(e) => { + tracing::warn!("demo: ingest failed: {e}"); + } + } + }); +} + +/// Background task: every `interval`, snapshot every galaxy and write to +/// disk. Errors are logged and ignored — we don't want a transient disk +/// hiccup to take down the simulation. +fn spawn_snapshot_task(sim: SimHandle, store: Store, interval: Duration) { + tokio::spawn(async move { + let mut ticker = tokio::time::interval(interval); + // First tick fires immediately by default; we want the first save + // *after* an interval has passed so boot doesn't double-write. + ticker.tick().await; + loop { + ticker.tick().await; + match sim.snapshot_all().await { + Ok(snaps) => { + for snap in &snaps { + if let Err(e) = store.save_galaxy(snap) { + tracing::warn!("snapshot save failed: {e}"); + } + } + tracing::debug!("snapshot tick: persisted {} galaxies", snaps.len()); + } + Err(e) => { + tracing::warn!("snapshot_all failed: {e}"); + } + } + } + }); +} + #[tokio::main] async fn main() -> Result<()> { init_tracing(); let cfg = load_config()?; + // Stage 7 CLI: only one subcommand right now (`demo`). Anything else + // boots the server normally. + let demo_mode = std::env::args().nth(1).as_deref() == Some("demo"); + let server_cfg = ServerConfig { host: cfg.server.host, port: cfg.server.port, @@ -125,9 +316,28 @@ async fn main() -> Result<()> { default_shape.minor_radius ); - let sim = sophia_sim::spawn_sim(default_shape); - let info = sim.create_galaxy("default".to_string()).await?; - tracing::info!("default galaxy ready: id={:?} name={}", info.id, info.name); + let store = Store::open(&cfg.storage.data_dir) + .with_context(|| format!("opening sled store at {}", cfg.storage.data_dir))?; + tracing::info!("persistence store open at {}", cfg.storage.data_dir); - serve(server_cfg, sim, llm).await + let sim = sophia_sim::spawn_sim(default_shape); + let restored = hydrate_or_seed(&sim, &store).await?; + tracing::info!("hydrated {restored} galaxies from disk"); + + spawn_snapshot_task( + sim.clone(), + store.clone(), + Duration::from_secs(cfg.storage.snapshot_interval_secs.max(1)), + ); + tracing::info!( + "snapshot task scheduled every {}s", + cfg.storage.snapshot_interval_secs + ); + + if demo_mode { + tracing::info!("demo mode: will auto-ingest curated corpus once server is reachable"); + spawn_demo_ingest(server_cfg.host.clone(), server_cfg.port); + } + + serve(server_cfg, sim, llm, cfg.token_caps, store).await } diff --git a/crates/sophia-core/src/dto.rs b/crates/sophia-core/src/dto.rs index b29cb6c..e36d50a 100644 --- a/crates/sophia-core/src/dto.rs +++ b/crates/sophia-core/src/dto.rs @@ -10,6 +10,8 @@ use serde::{Deserialize, Serialize}; use crate::engram::EngramState; use crate::ids::{EngramId, GalaxyId, SynapseId}; use crate::manifest::Manifest; +use crate::memory::{Introspection, Memory}; +use crate::query::{QueryId, QueryStatus}; use crate::synapse::Synapse; #[derive(Debug, Clone, Serialize, Deserialize)] @@ -62,6 +64,10 @@ pub struct EngramDetail { pub manifest: Option, pub slate_dim: Option, pub slate_norm: Option, + pub introspection: Introspection, + pub memories: Vec, + /// True for Query-Engrams (Stage 5). + pub pinned: bool, } /// Position frame: a downsampled bundle of all engram positions at a moment in @@ -97,6 +103,37 @@ pub enum SimEvent { major_radius: f32, minor_radius: f32, }, + /// An engram's lifecycle state changed (e.g., responder lit up while + /// answering a query, then returned to idle). Frontend uses this to + /// re-paint the engram's color in real time. + EngramStateChanged { + id: EngramId, + state: EngramState, + }, + /// A query was started — a Query-Engram materialized at `position` with + /// the question text. Frontend uses this to register the query for the + /// answer panel + visual highlight. + QueryStarted { + query: QueryId, + engram: EngramId, + position: [f32; 3], + text: String, + }, + /// The integrator produced a new running answer. `version` increments + /// monotonically per query; clients can ignore stale frames. + QueryAnswerUpdated { + query: QueryId, + version: u32, + status: QueryStatus, + answer: String, + }, + /// Terminal event for a query. The Query-Engram has been moved to the + /// `Memorize` state and the SSE stream is closed. + QueryFinished { + query: QueryId, + status: QueryStatus, + responder_count: u32, + }, /// Position frame is sent over the wire as a binary frame; this variant /// carries the in-process payload. #[serde(skip)] diff --git a/crates/sophia-core/src/engram.rs b/crates/sophia-core/src/engram.rs index 0061638..c7cb5e2 100644 --- a/crates/sophia-core/src/engram.rs +++ b/crates/sophia-core/src/engram.rs @@ -1,8 +1,11 @@ +use std::collections::VecDeque; + use glam::Vec3; use serde::{Deserialize, Serialize}; use crate::ids::EngramId; use crate::manifest::Manifest; +use crate::memory::{Introspection, Memory}; use crate::slate::Slate; /// Engram state machine — see §5 of `docs/system-analysis.md`. @@ -31,7 +34,9 @@ impl EngramState { /// /// Stage 2 adds `manifest` (the source content) and `slate` (an embedding for /// fast similarity). Stage 4 adds taxonomy/goals/memories on top of that. -#[derive(Debug, Clone)] +/// Stage 6 adds Serialize/Deserialize so engrams can round-trip through the +/// `sophia-store` snapshot file. +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct Engram { pub id: EngramId, /// Slot index for the visualization's `InstancedMesh`. Assigned at birth, @@ -51,4 +56,14 @@ pub struct Engram { /// Embedding vector — System-1 layer of the Universal Slate. `None` for /// synthetic seed engrams. pub slate: Option, + /// LLM-generated self-description (Stage 4). Filled async after birth; + /// empty until the introspection task completes. + pub introspection: Introspection, + /// Per-engram memory log (Stage 4). Bounded — older entries evicted from + /// the front when full. + pub memories: VecDeque, + /// Pinned engrams skip physics integration — they sit at their birth + /// position permanently. Used for Query-Engrams (Stage 5), which + /// materialize at the galaxy center and shouldn't drift into the tube. + pub pinned: bool, } diff --git a/crates/sophia-core/src/galaxy.rs b/crates/sophia-core/src/galaxy.rs index ecd6323..1cd45bd 100644 --- a/crates/sophia-core/src/galaxy.rs +++ b/crates/sophia-core/src/galaxy.rs @@ -59,8 +59,9 @@ pub struct Galaxy { pub id: GalaxyId, pub name: String, /// Galaxy origin in world coords. Always `Vec3::ZERO` in v1; reserved for - /// future multi-galaxy layouts. - #[serde(skip)] + /// future multi-galaxy layouts. Serialized so persistence round-trips + /// the full state (Stage 6) — defaults to ZERO if missing from disk. + #[serde(default)] pub center: Vec3, pub shape: GalaxyShape, } diff --git a/crates/sophia-core/src/lib.rs b/crates/sophia-core/src/lib.rs index 746521b..fccb925 100644 --- a/crates/sophia-core/src/lib.rs +++ b/crates/sophia-core/src/lib.rs @@ -8,6 +8,9 @@ pub mod engram; pub mod galaxy; pub mod ids; pub mod manifest; +pub mod memory; +pub mod persistence; +pub mod query; pub mod slate; pub mod synapse; @@ -16,6 +19,9 @@ pub use engram::{Engram, EngramState}; pub use galaxy::{Galaxy, GalaxyShape}; pub use ids::{EngramId, GalaxyId, SynapseId}; pub use manifest::Manifest; +pub use memory::{Introspection, Memory, MemoryKind, MAX_MEMORIES}; +pub use persistence::GalaxySnapshot; +pub use query::{QueryId, QueryStatus}; pub use slate::Slate; pub use synapse::{canonical_pair, Synapse}; diff --git a/crates/sophia-core/src/memory.rs b/crates/sophia-core/src/memory.rs new file mode 100644 index 0000000..ff48212 --- /dev/null +++ b/crates/sophia-core/src/memory.rs @@ -0,0 +1,52 @@ +//! Per-Engram memory entries — a running log of meaningful events from the +//! engram's own perspective. Stage 4 only records synapse formation; Stage 5 +//! will add conversation memorialization. + +use serde::{Deserialize, Serialize}; + +use crate::ids::EngramId; +use crate::query::QueryId; + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum MemoryKind { + Born, + Introspected, + /// A synapse formed with another engram. + SynapseFormed { with: EngramId, weight: f32 }, + /// This engram contributed a snippet to a query's running answer. + /// `snippet` is the engram's own LLM-generated POV on the question. + QueryParticipated { query: QueryId, snippet: String }, + /// (Query-engrams only.) The query completed and produced a final answer. + /// Stored on the Query-Engram itself so the inspector can show the + /// resolution after the fact. + QueryAnswered { answer: String, responder_count: u32 }, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Memory { + /// Time relative to the start of the simulation (ms). + pub at_ms: u32, + pub kind: MemoryKind, +} + +/// Hard cap on a single engram's memory list. Older entries are evicted from +/// the front of the deque when full. Stage 4 picks 64 — generous enough for +/// the demo, small enough not to bloat the Hello/inspector payloads. +pub const MAX_MEMORIES: usize = 64; + +/// Output of LLM-driven self-introspection at birth (Stage 4). Stored on the +/// `Engram` itself, not the `Manifest`, since it's the engram's *interpretation* +/// of its source content rather than the source itself. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct Introspection { + pub taxonomy: Vec, + pub goals: Vec, + pub open_questions: Vec, +} + +impl Introspection { + pub fn is_empty(&self) -> bool { + self.taxonomy.is_empty() && self.goals.is_empty() && self.open_questions.is_empty() + } +} diff --git a/crates/sophia-core/src/persistence.rs b/crates/sophia-core/src/persistence.rs new file mode 100644 index 0000000..b099025 --- /dev/null +++ b/crates/sophia-core/src/persistence.rs @@ -0,0 +1,31 @@ +//! Persistence types (Stage 6). +//! +//! `GalaxySnapshot` is the on-disk shape of a galaxy: its metadata plus the +//! full set of Engrams and Synapses it contains. This is what the periodic +//! snapshot task in `sophia-bin` writes to the sled store, and what the +//! `SimHandle::hydrate_galaxy` command reads back at boot. +//! +//! Pinned engrams (Stage 5 Query-Engrams) are intentionally excluded from +//! snapshots. Queries are ephemeral — their orchestrator task is gone after +//! a server restart, and a half-rendered query-engram with no SSE channel +//! would be confusing in the UI. Letting them die with the process is the +//! cleaner default. + +use serde::{Deserialize, Serialize}; + +use crate::engram::Engram; +use crate::galaxy::Galaxy; +use crate::synapse::Synapse; + +/// One galaxy's full persistent state. Encoded with `bincode` and stored +/// under the galaxy's id in the sled tree. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct GalaxySnapshot { + pub galaxy: Galaxy, + /// All non-pinned engrams. The order is preserved so `instance_idx` + /// values remain meaningful after a round-trip — the sim re-uses each + /// engram's stored `instance_idx` rather than re-numbering, so the + /// position-frame slot mapping stays stable across restarts. + pub engrams: Vec, + pub synapses: Vec, +} diff --git a/crates/sophia-core/src/query.rs b/crates/sophia-core/src/query.rs new file mode 100644 index 0000000..b01f696 --- /dev/null +++ b/crates/sophia-core/src/query.rs @@ -0,0 +1,45 @@ +//! Query domain types (Stage 5). +//! +//! A "query" is a question that materializes as a special pinned Engram at +//! the galaxy center. The Query-Engram broadcasts to nearby resonating +//! Engrams; their snippets are integrated into a streaming answer. +//! +//! `QueryId` is distinct from `EngramId` even though every query has a +//! backing query-engram — keeping them separate lets the orchestrator track +//! query state (responders, snippets, answer versions) independently of the +//! engram's lifecycle in the world. + +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(transparent)] +pub struct QueryId(pub Uuid); + +impl QueryId { + pub fn new() -> Self { + Self(Uuid::now_v7()) + } +} + +impl Default for QueryId { + fn default() -> Self { + Self::new() + } +} + +/// Lifecycle phase of a query, as exposed to SSE consumers. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum QueryStatus { + /// Query-Engram materialized; broadcast not yet started. + Pending, + /// Phase A/B retrieval in flight; responders being polled. + Responding, + /// All responses gathered; integrator producing final answer. + Integrating, + /// Final answer published; query-engram transitioned to Memorize. + Done, + /// Aborted (e.g. embedding failure or sim shutdown). + Failed, +} diff --git a/crates/sophia-llm/src/client.rs b/crates/sophia-llm/src/client.rs index aff5f63..edd77b2 100644 --- a/crates/sophia-llm/src/client.rs +++ b/crates/sophia-llm/src/client.rs @@ -96,6 +96,20 @@ impl LmStudioClient { &self.inner.cfg } + /// How many parallel-op permits are currently free. Equal to `parallel_ops` + /// when idle, drops as concurrent chat/embedding calls run. Used by the + /// HUD's "LLM queue" indicator — a number that consistently sits at 0 + /// means the LLM is the bottleneck. + pub fn available_permits(&self) -> usize { + self.inner.parallel.available_permits() + } + + /// The configured ceiling. Same as `config().parallel_ops`, exposed for + /// callers that already have the client handle but not the config. + pub fn parallel_ops(&self) -> usize { + self.inner.cfg.parallel_ops.max(1) + } + /// Apply Bearer auth header if a token is configured. LM Studio's REST /// server returns 401 if auth is enabled (default in 0.3.x+) and no token /// is sent. diff --git a/crates/sophia-llm/src/introspect.rs b/crates/sophia-llm/src/introspect.rs new file mode 100644 index 0000000..50787d8 --- /dev/null +++ b/crates/sophia-llm/src/introspect.rs @@ -0,0 +1,90 @@ +//! LLM-driven introspection: ask the chat model to summarise one piece of +//! text into a small structured `Introspection` record (taxonomy, goals, +//! open questions). Caveman budget per §13.5. + +use serde::Deserialize; +use sophia_core::Introspection; + +use crate::budget::TokenCaps; +use crate::client::{ChatMessage, LmError, LmStudioClient}; + +const SYSTEM_PROMPT: &str = "\ +You summarize one short piece of knowledge for a knowledge graph.\n\ +Output ONLY valid JSON matching this schema:\n\ +{\n \"taxonomy\": [string, ...],\n \"goals\": [string, ...],\n \"open_questions\": [string, ...]\n}\n\ +- taxonomy: 3 to 5 short topical labels, from general to specific.\n\ +- goals: 1 to 3 short statements of what this knowledge enables or answers.\n\ +- open_questions: 1 to 3 short questions this knowledge invites but doesn't answer.\n\ +Each item must be at most 12 words. No prose outside the JSON."; + +/// Truncate excessively long input so we stay inside the input cap. Caveman +/// budget tolerates ~400 input tokens total — content beyond ~1200 chars +/// (≈300 tokens) is unlikely to add useful introspection, so we cut. +const MAX_INPUT_CHARS: usize = 1200; + +#[derive(Debug, Deserialize)] +struct IntrospectionPayload { + #[serde(default)] + taxonomy: Vec, + #[serde(default)] + goals: Vec, + #[serde(default)] + open_questions: Vec, +} + +/// Introspect a single piece of text. Returns an empty `Introspection` if +/// the LLM responds with non-JSON or empty content; the caller can decide +/// whether to retry. Caps input length defensively. +pub async fn introspect_text( + client: &LmStudioClient, + text: &str, + caps: TokenCaps, +) -> Result { + let trimmed = if text.chars().count() > MAX_INPUT_CHARS { + let head: String = text.chars().take(MAX_INPUT_CHARS).collect(); + format!("{head}…") + } else { + text.to_string() + }; + let messages = [ + ChatMessage::system(SYSTEM_PROMPT), + ChatMessage::user(format!("Content:\n\"\"\"\n{trimmed}\n\"\"\"")), + ]; + let raw = client.chat(&messages, caps).await?; + Ok(parse_introspection(&raw)) +} + +/// Best-effort JSON extraction. Models occasionally wrap output in ```json +/// fences; we strip those before parsing. On failure returns empty. +fn parse_introspection(raw: &str) -> Introspection { + let json_text = strip_code_fences(raw.trim()); + let parsed: Result = serde_json::from_str(json_text); + match parsed { + Ok(p) => Introspection { + taxonomy: cleanup(p.taxonomy), + goals: cleanup(p.goals), + open_questions: cleanup(p.open_questions), + }, + Err(_) => Introspection::default(), + } +} + +fn strip_code_fences(s: &str) -> &str { + let s = s.trim(); + let stripped = s.strip_prefix("```json").or_else(|| s.strip_prefix("```")); + if let Some(rest) = stripped { + let rest = rest.trim_start_matches('\n'); + rest.strip_suffix("```").unwrap_or(rest).trim() + } else { + s + } +} + +fn cleanup(items: Vec) -> Vec { + items + .into_iter() + .map(|s| s.trim().to_string()) + .filter(|s| !s.is_empty()) + .take(8) // hard upper bound, just in case + .collect() +} diff --git a/crates/sophia-llm/src/lib.rs b/crates/sophia-llm/src/lib.rs index 9ff7108..331d26e 100644 --- a/crates/sophia-llm/src/lib.rs +++ b/crates/sophia-llm/src/lib.rs @@ -4,6 +4,10 @@ mod budget; mod client; +mod introspect; +mod query; pub use budget::{TokenCaps, TokenCapsAll}; pub use client::{ChatMessage, ChatRole, LmError, LmStudioClient, LmStudioConfig, EMBED_BATCH_LIMIT}; +pub use introspect::introspect_text; +pub use query::{integrate_responses, respond_to_query}; diff --git a/crates/sophia-llm/src/query.rs b/crates/sophia-llm/src/query.rs new file mode 100644 index 0000000..5bde085 --- /dev/null +++ b/crates/sophia-llm/src/query.rs @@ -0,0 +1,92 @@ +//! Query-time LLM prompts (Stage 5). +//! +//! Two distinct calls per query: +//! +//! - `respond_to_query`: each resonating engram (the "responder") is asked to +//! say what *its own* knowledge contributes to the question. Caveman cap +//! 200 in / 150 out — these are short, single-perspective snippets. +//! - `integrate_responses`: the Query-Engram folds the snippets it has +//! received so far into a single running answer. Cap 800 in / 300 out. +//! Re-runs every couple of seconds while new snippets arrive, so the +//! browser sees the answer refine. + +use crate::budget::TokenCaps; +use crate::client::{ChatMessage, LmError, LmStudioClient}; + +const RESPONDER_SYSTEM: &str = "\ +You are one fragment of knowledge inside a larger memory. \ +A query has been broadcast through the network. \ +Speak ONLY for what your own content contributes to the question. \ +Do not summarize the question. Do not speculate beyond your content. \ +If your content is irrelevant, reply with the single word: PASS. \ +Otherwise: 1–3 short sentences, plain prose, no headings, no bullets."; + +const INTEGRATOR_SYSTEM: &str = "\ +You are the answering process for a question. You receive snippets from \ +fragments of knowledge that resonated with the question. Synthesize them \ +into ONE coherent answer to the user's question. Cite no sources. \ +Do not list the snippets. Do not say things like \"based on the snippets\". \ +Plain prose, 2–6 sentences. If the snippets are insufficient, say so briefly."; + +/// Truncate one responder's source content to keep the per-call payload small. +const MAX_RESPONDER_INPUT_CHARS: usize = 600; + +/// Ask one resonating engram for its take on the question. Returns `None` if +/// the responder explicitly opts out (`PASS`) or returns empty content. +/// Keeping `Option` rather than `Result<...>` lets the orchestrator +/// distinguish "this engram had nothing useful to say" from "the LLM call +/// failed" — both are fine, but only the latter should be logged. +pub async fn respond_to_query( + client: &LmStudioClient, + question: &str, + responder_text: &str, + caps: TokenCaps, +) -> Result, LmError> { + let trimmed = if responder_text.chars().count() > MAX_RESPONDER_INPUT_CHARS { + let head: String = responder_text.chars().take(MAX_RESPONDER_INPUT_CHARS).collect(); + format!("{head}…") + } else { + responder_text.to_string() + }; + let user = format!( + "Question:\n{question}\n\nYour content:\n\"\"\"\n{trimmed}\n\"\"\"" + ); + let messages = [ + ChatMessage::system(RESPONDER_SYSTEM), + ChatMessage::user(user), + ]; + let raw = client.chat(&messages, caps).await?; + let cleaned = raw.trim(); + if cleaned.is_empty() || cleaned.eq_ignore_ascii_case("PASS") { + return Ok(None); + } + Ok(Some(cleaned.to_string())) +} + +/// Fold the current set of responder snippets into one running answer. The +/// orchestrator calls this repeatedly as more snippets arrive. +pub async fn integrate_responses( + client: &LmStudioClient, + question: &str, + snippets: &[String], + caps: TokenCaps, +) -> Result { + if snippets.is_empty() { + return Ok(String::new()); + } + // Number the snippets so the integrator can reason about distinct + // perspectives without us having to add ids. + let mut buf = String::new(); + for (i, s) in snippets.iter().enumerate() { + buf.push_str(&format!("[{}] {}\n", i + 1, s.trim())); + } + let user = format!( + "Question:\n{question}\n\nSnippets:\n{buf}\nAnswer the question now." + ); + let messages = [ + ChatMessage::system(INTEGRATOR_SYSTEM), + ChatMessage::user(user), + ]; + let raw = client.chat(&messages, caps).await?; + Ok(raw.trim().to_string()) +} diff --git a/crates/sophia-server/Cargo.toml b/crates/sophia-server/Cargo.toml index f835574..8e784af 100644 --- a/crates/sophia-server/Cargo.toml +++ b/crates/sophia-server/Cargo.toml @@ -7,9 +7,10 @@ rust-version.workspace = true authors.workspace = true [dependencies] -sophia-core = { workspace = true } -sophia-sim = { workspace = true } -sophia-llm = { workspace = true } +sophia-core = { workspace = true } +sophia-sim = { workspace = true } +sophia-llm = { workspace = true } +sophia-store = { workspace = true } anyhow = { workspace = true } tokio = { workspace = true } @@ -22,4 +23,5 @@ tower = { workspace = true } tower-http = { workspace = true } futures-util = { workspace = true } +tokio-stream = { workspace = true } bytes = { workspace = true } diff --git a/crates/sophia-server/src/lib.rs b/crates/sophia-server/src/lib.rs index 8696b88..2dcd8cd 100644 --- a/crates/sophia-server/src/lib.rs +++ b/crates/sophia-server/src/lib.rs @@ -1,23 +1,32 @@ //! HTTP / WebSocket server. +use std::convert::Infallible; use std::path::PathBuf; use std::sync::Arc; +use std::time::Duration; use anyhow::Result; use axum::extract::{Path, Query, State, WebSocketUpgrade}; use axum::http::StatusCode; +use axum::response::sse::{Event as SseEvent, KeepAlive, Sse}; use axum::response::{IntoResponse, Response}; use axum::routing::{delete, get, post}; use axum::{Json, Router}; +use futures_util::stream::{self, Stream, StreamExt}; use serde::Deserialize; -use sophia_core::{EngramId, GalaxyId, GalaxyShape, Manifest, Slate}; -use sophia_llm::{LmStudioClient, EMBED_BATCH_LIMIT}; +use sophia_core::{EngramId, GalaxyId, GalaxyShape, Manifest, QueryId, QueryStatus, Slate}; +use sophia_llm::{introspect_text, LmStudioClient, TokenCapsAll, EMBED_BATCH_LIMIT}; use sophia_sim::{IngestItem, SimHandle}; +use sophia_store::Store; +use tokio_stream::wrappers::BroadcastStream; use tower_http::services::{ServeDir, ServeFile}; use tower_http::trace::TraceLayer; +mod query; mod ws; +use query::QueryRegistry; + #[derive(Debug, Clone)] pub struct ServerConfig { pub host: String, @@ -29,29 +38,54 @@ pub struct ServerConfig { struct AppState { sim: SimHandle, llm: LmStudioClient, + token_caps: TokenCapsAll, + queries: Arc, + store: Store, } -pub fn build_router(cfg: &ServerConfig, sim: SimHandle, llm: LmStudioClient) -> Router { +pub fn build_router( + cfg: &ServerConfig, + sim: SimHandle, + llm: LmStudioClient, + token_caps: TokenCapsAll, + store: Store, +) -> Router { let index = cfg.static_dir.join("index.html"); let static_service = ServeDir::new(&cfg.static_dir).fallback(ServeFile::new(&index)); - let state = Arc::new(AppState { sim, llm }); + let state = Arc::new(AppState { + sim, + llm, + token_caps, + queries: Arc::new(QueryRegistry::new()), + store, + }); Router::new() .route("/healthz", get(healthz)) .route("/api/galaxy", post(create_galaxy).get(list_galaxies)) - .route("/api/galaxy/:id", delete(delete_galaxy_unimpl)) + .route("/api/galaxy/:id", delete(delete_galaxy)) .route("/api/galaxy/:id/seed", post(seed_galaxy)) .route("/api/galaxy/:id/ingest", post(ingest_galaxy)) .route("/api/galaxy/:id/resize", post(resize_galaxy)) + .route("/api/galaxy/:id/query", post(start_query)) .route("/api/galaxy/:gid/engrams/:eid", get(get_engram)) + .route("/api/query/:id", get(get_query)) + .route("/api/query/:id/stream", get(query_stream)) + .route("/api/stats", get(stats)) .route("/ws/galaxy/:id/events", get(ws_events)) .with_state(state) .fallback_service(static_service) .layer(TraceLayer::new_for_http()) } -pub async fn serve(cfg: ServerConfig, sim: SimHandle, llm: LmStudioClient) -> Result<()> { - let app = build_router(&cfg, sim, llm); +pub async fn serve( + cfg: ServerConfig, + sim: SimHandle, + llm: LmStudioClient, + token_caps: TokenCapsAll, + store: Store, +) -> Result<()> { + let app = build_router(&cfg, sim, llm, token_caps, store); let addr = format!("{}:{}", cfg.host, cfg.port); let listener = tokio::net::TcpListener::bind(&addr).await?; tracing::info!("sophia-server listening on http://{}", addr); @@ -86,6 +120,26 @@ async fn healthz(State(s): State>) -> Response { .into_response() } +/// `GET /api/stats` — cheap snapshot for the HUD. Polled at ~1 Hz by the +/// browser; every field is integer so the panel stays readable. The +/// `llm_queue_depth` is `parallel_ops - available_permits` (i.e. how many +/// LLM calls are currently in flight). +async fn stats(State(s): State>) -> Response { + let sim_stats = s.sim.stats().await; + let permits_free = s.llm.available_permits(); + let parallel_ops = s.llm.parallel_ops(); + let in_flight = parallel_ops.saturating_sub(permits_free); + Json(serde_json::json!({ + "ticks_per_sec": sim_stats.ticks_per_sec, + "engrams_total": sim_stats.engrams_total, + "synapses_total": sim_stats.synapses_total, + "galaxies": sim_stats.galaxies, + "llm_queue_depth": in_flight, + "llm_parallel_cap": parallel_ops, + })) + .into_response() +} + #[derive(Deserialize)] struct CreateGalaxyBody { name: String, @@ -96,7 +150,22 @@ async fn create_galaxy( Json(body): Json, ) -> Response { match s.sim.create_galaxy(body.name).await { - Ok(info) => Json(info).into_response(), + Ok(info) => { + // Persist immediately so a kill+restart in the first 60 s + // doesn't lose the new galaxy. Failure here is non-fatal — + // the next periodic snapshot tick will pick it up. + if let Ok(snaps) = s.sim.snapshot_all().await { + for snap in &snaps { + if snap.galaxy.id == info.id { + if let Err(e) = s.store.save_galaxy(snap) { + tracing::warn!("initial save of new galaxy failed: {e}"); + } + break; + } + } + } + Json(info).into_response() + } Err(e) => sim_error(e), } } @@ -108,9 +177,22 @@ async fn list_galaxies(State(s): State>) -> Response { } } -async fn delete_galaxy_unimpl(Path(_id): Path) -> Response { - // Stage 6 will wire deletion to the persistence layer. - (StatusCode::NOT_IMPLEMENTED, "galaxy deletion arrives in Stage 6").into_response() +async fn delete_galaxy( + State(s): State>, + Path(id): Path, +) -> Response { + if let Err(e) = s.sim.delete_galaxy(id).await { + return sim_error(e); + } + if let Err(e) = s.store.delete_galaxy(id) { + // Sim succeeded but disk didn't — log and report; the galaxy is + // gone from memory so this is a 500-ish situation, but the next + // periodic snapshot will repair (it just won't write the deleted + // galaxy back since it's not in `snapshot_all`). + tracing::warn!("sim deleted galaxy {:?} but store delete failed: {e}", id); + return (StatusCode::INTERNAL_SERVER_ERROR, format!("store: {e}")).into_response(); + } + (StatusCode::NO_CONTENT, "").into_response() } #[derive(Deserialize)] @@ -171,6 +253,9 @@ async fn ingest_galaxy( } } + // Pair (text, slate) → IngestItem, but keep texts so we can launch a + // post-ingest introspection task per engram below. + let texts_for_introspection: Vec = texts.clone(); let items: Vec = texts .into_iter() .zip(all_vecs) @@ -181,7 +266,40 @@ async fn ingest_galaxy( .collect(); tracing::info!("ingest: {} items into galaxy {:?}", items.len(), id); - match s.sim.ingest(id, items).await { + let result = s.sim.ingest(id, items).await; + + // On success, fire-and-forget one LLM introspection task per engram. + // The LM Studio client's semaphore (parallel_ops) bounds total + // concurrency, so even 200 spawned futures only run N at a time. + // Updates trickle in over the next several seconds via + // `SimHandle::update_introspection`. + if let Ok(ids) = &result { + for (engram_id, text) in ids.iter().zip(texts_for_introspection) { + let llm = s.llm.clone(); + let sim = s.sim.clone(); + let caps = s.token_caps.introspection(); + let galaxy_id = id; + let engram_id = *engram_id; + tokio::spawn(async move { + match introspect_text(&llm, &text, caps).await { + Ok(intro) => { + if intro.is_empty() { + tracing::debug!(?engram_id, "introspection returned empty"); + } else if let Err(e) = + sim.update_introspection(galaxy_id, engram_id, intro).await + { + tracing::warn!(?engram_id, "update_introspection failed: {e}"); + } + } + Err(e) => { + tracing::warn!(?engram_id, "introspect_text failed: {e}"); + } + } + }); + } + } + + match result { Ok(ids) => Json(serde_json::json!({ "engram_ids": ids })).into_response(), Err(e) => sim_error(e), } @@ -230,6 +348,99 @@ async fn ws_events( upgrade.on_upgrade(move |socket| ws::run_galaxy_socket(socket, s.sim.clone(), id)) } +#[derive(Deserialize)] +struct QueryBody { + text: String, +} + +/// `POST /api/galaxy/:id/query` — embed the question, materialize a Query-Engram, +/// kick off the orchestrator. Returns 202 + `{query_id}` immediately so the +/// client can SSE-subscribe before the first responder fires. +async fn start_query( + State(s): State>, + Path(id): Path, + Json(body): Json, +) -> Response { + match query::start_query( + s.sim.clone(), + s.llm.clone(), + s.token_caps.clone(), + s.queries.clone(), + id, + body.text, + ) + .await + { + Ok(qid) => ( + StatusCode::ACCEPTED, + Json(serde_json::json!({ "query_id": qid })), + ) + .into_response(), + Err(msg) => (StatusCode::BAD_REQUEST, msg).into_response(), + } +} + +/// `GET /api/query/:id` — current snapshot. Useful for clients that miss the +/// initial SSE event or want a one-shot read. +async fn get_query( + State(s): State>, + Path(id): Path, +) -> Response { + match s.queries.get(id).await { + Some(entry) => Json(entry.snapshot().await).into_response(), + None => (StatusCode::NOT_FOUND, "query not found").into_response(), + } +} + +/// `GET /api/query/:id/stream` — Server-Sent Events of `QueryUpdate` JSON. +/// First event is always the latest known snapshot (so a late subscriber +/// catches up without polling); subsequent events come from the per-query +/// broadcast channel. The stream ends when the orchestrator publishes a +/// terminal status (`Done` or `Failed`). +async fn query_stream( + State(s): State>, + Path(id): Path, +) -> Response { + let Some(entry) = s.queries.get(id).await else { + return (StatusCode::NOT_FOUND, "query not found").into_response(); + }; + let initial = entry.snapshot().await; + let receiver = entry.subscribe(); + let live = BroadcastStream::new(receiver).filter_map(|res| async move { + match res { + Ok(update) => Some(update), + Err(e) => { + tracing::debug!("sse lagged: {e}"); + None + } + } + }); + // Emit the initial snapshot first, then forward live updates until a + // terminal frame (`done`/`failed`) — the terminal frame is INCLUDED, + // and the stream closes immediately after. Standard `take_while` drops + // the matching item; `unfold` gives us "include and then end" by + // collapsing the state to None after emitting the terminal frame. + let combined = stream::once(async move { initial }).chain(live); + let stream = stream::unfold(Some(Box::pin(combined)), |state| async move { + let mut s = state?; + let item = s.next().await?; + let next_state = if matches!(item.status, QueryStatus::Done | QueryStatus::Failed) { + None + } else { + Some(s) + }; + Some((item, next_state)) + }); + let sse_stream: std::pin::Pin> + Send>> = + Box::pin(stream.map(|u| { + let payload = serde_json::to_string(&u).unwrap_or_else(|_| "{}".into()); + Ok(SseEvent::default().data(payload)) + })); + Sse::new(sse_stream) + .keep_alive(KeepAlive::new().interval(Duration::from_secs(15))) + .into_response() +} + fn sim_error(e: sophia_sim::SimError) -> Response { use sophia_sim::SimError::*; let status = match e { diff --git a/crates/sophia-server/src/query.rs b/crates/sophia-server/src/query.rs new file mode 100644 index 0000000..fced7fd --- /dev/null +++ b/crates/sophia-server/src/query.rs @@ -0,0 +1,393 @@ +//! Stage 5 query orchestration. +//! +//! A "query" is a conversation initiated by the user. Lifecycle: +//! +//! 1. `POST /api/galaxy/:id/query {text}` — embed the question, materialize a +//! pinned Query-Engram in the sim, register a per-query SSE bus, spawn the +//! background orchestrator, return `{query_id}` immediately (HTTP 202). +//! 2. Orchestrator (this module): `sim.broadcast_query(...)` chooses the +//! responders (Phase A cosine + Phase B 1-hop synaptic). Each responder +//! is asked, in parallel under the LM Studio semaphore, what *its* content +//! contributes to the question. As snippets arrive they're recorded on +//! the responder (visual light-up) and integrated into a running answer +//! every `INTEGRATION_INTERVAL`. +//! 3. When all responses are gathered (or the timeout hits), one final +//! integration produces the canonical answer; `sim.finish_query(...)` +//! transitions the Query-Engram to Memorize and the SSE stream closes. +//! +//! Decisions worth noting: +//! +//! - Responder cap (`MAX_RESPONDERS = 12`): keeps the LLM queue from being +//! monopolized by one query and bounds the integrator's input length. +//! - Re-integration is *time-driven* (every 2 s) rather than per-snippet — +//! batching snippets into a single integrator call costs less and gives +//! the answer time to refine. +//! - Per-query SSE channel buffer is small (16); the latest snapshot is +//! stored separately so a slow client can resync to current state on +//! connect rather than replaying every interim version. + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use serde::{Deserialize, Serialize}; +use sophia_core::{GalaxyId, QueryId, QueryStatus, Slate}; +use sophia_llm::{integrate_responses, respond_to_query, LmStudioClient, TokenCapsAll}; +use sophia_sim::SimHandle; +use tokio::sync::{broadcast, Mutex, RwLock}; +use tokio::task::JoinSet; +use tracing::{debug, warn}; + +/// Maximum number of responders we ask per query. Larger numbers don't help +/// the answer much (the integrator can only fit ~5–10 distinct viewpoints +/// in its 800-token input cap) and starve the LLM queue. +const MAX_RESPONDERS: usize = 12; +/// How often the integrator re-runs while new snippets are arriving. Every +/// 2 s gives the user a visible refinement cadence without burning LLM +/// budget on near-identical re-integrations. +const INTEGRATION_INTERVAL: Duration = Duration::from_millis(2_000); +/// Hard cap on time spent waiting for responders. After this we stop polling, +/// run a final integration on whatever snippets we have, and finish. Tuned +/// for `gemma-4-e4b` thinking-mode latency: each responder call takes +/// ~20–40 s once the hidden reasoning budget is large enough to actually +/// produce visible content, and LM Studio serialises some of them despite +/// the parallel-op semaphore on our side. +const RESPONDER_TIMEOUT: Duration = Duration::from_secs(120); +/// SSE channel buffer per query. SSE consumers that lag past this will see +/// old versions dropped; the registry's `latest` snapshot covers the rest. +const SSE_CAPACITY: usize = 16; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct QueryUpdate { + pub query_id: QueryId, + pub version: u32, + pub status: QueryStatus, + pub answer: String, + pub responder_count: u32, +} + +#[derive(Debug)] +pub struct QueryEntry { + sender: broadcast::Sender, + latest: Mutex, +} + +/// Process-wide registry. Cheaply cloneable Arc holds an `RwLock` over the +/// `HashMap>` — typical access is read-heavy +/// (subscribers + GET /api/query/:id), with a single write per new query. +#[derive(Default)] +pub struct QueryRegistry { + inner: RwLock>>, +} + +impl QueryRegistry { + pub fn new() -> Self { + Self::default() + } + + async fn register(&self, query_id: QueryId, initial: QueryUpdate) -> Arc { + let (sender, _) = broadcast::channel(SSE_CAPACITY); + let entry = Arc::new(QueryEntry { + sender, + latest: Mutex::new(initial), + }); + self.inner.write().await.insert(query_id, entry.clone()); + entry + } + + pub async fn get(&self, query_id: QueryId) -> Option> { + self.inner.read().await.get(&query_id).cloned() + } +} + +impl QueryEntry { + pub fn subscribe(&self) -> broadcast::Receiver { + self.sender.subscribe() + } + + pub async fn snapshot(&self) -> QueryUpdate { + self.latest.lock().await.clone() + } + + async fn publish(&self, update: QueryUpdate) { + *self.latest.lock().await = update.clone(); + // Fan-out send; drop silently if no one is subscribed. + let _ = self.sender.send(update); + } +} + +/// Start a query. Embeds the question, materializes the Query-Engram, kicks +/// off the orchestrator. Returns the new `QueryId` immediately so the caller +/// can SSE-subscribe before the first responder fires. +pub async fn start_query( + sim: SimHandle, + llm: LmStudioClient, + caps: TokenCapsAll, + registry: Arc, + galaxy: GalaxyId, + text: String, +) -> Result { + let trimmed = text.trim().to_string(); + if trimmed.is_empty() { + return Err("query text must not be empty".into()); + } + + // Embed the question. This blocks the request for the duration of one + // LM Studio embedding call (typically <100 ms). Doing it inline keeps + // the orchestrator simple — no "pending embedding" state. + let embedding = llm + .embed(&trimmed) + .await + .map_err(|e| format!("embed: {e}"))?; + let slate = Slate(embedding); + + let query_id = QueryId::new(); + + let query_engram = sim + .start_query_engram(galaxy, query_id, trimmed.clone(), slate) + .await + .map_err(|e| format!("start_query_engram: {e}"))?; + + // Seed registry with a Pending snapshot so SSE subscribers get something + // immediate even before broadcast finishes. + let initial = QueryUpdate { + query_id, + version: 0, + status: QueryStatus::Pending, + answer: String::new(), + responder_count: 0, + }; + let entry = registry.register(query_id, initial.clone()).await; + + tokio::spawn(orchestrate_query(OrchestrateCtx { + sim, + llm, + caps, + entry, + galaxy, + query_id, + query_engram, + question: trimmed, + })); + + Ok(query_id) +} + +struct OrchestrateCtx { + sim: SimHandle, + llm: LmStudioClient, + caps: TokenCapsAll, + entry: Arc, + galaxy: GalaxyId, + query_id: QueryId, + query_engram: sophia_core::EngramId, + question: String, +} + +/// The orchestrator. Owns the per-query lifecycle from broadcast to finish. +async fn orchestrate_query(ctx: OrchestrateCtx) { + let OrchestrateCtx { + sim, llm, caps, entry, galaxy, query_id, query_engram, question, + } = ctx; + + let responders = match sim.broadcast_query(galaxy, query_engram, MAX_RESPONDERS).await { + Ok(rs) => rs, + Err(e) => { + warn!(?query_id, "broadcast_query failed: {e}"); + publish_terminal(&entry, query_id, QueryStatus::Failed, String::new(), 0).await; + return; + } + }; + + if responders.is_empty() { + let answer = "No engrams resonated with this question.".to_string(); + let _ = sim + .publish_query_answer(galaxy, query_id, 1, QueryStatus::Done, answer.clone()) + .await; + let _ = sim + .finish_query(galaxy, query_id, query_engram, answer.clone(), 0) + .await; + publish_terminal(&entry, query_id, QueryStatus::Done, answer, 0).await; + return; + } + + debug!( + ?query_id, + responders = responders.len(), + "broadcasting query to responders" + ); + + // Fan out one responder LLM task per pick. Each task fetches the + // responder's source text from the sim, asks the LLM for that engram's + // POV, then records the snippet. The LM Studio semaphore inside + // `respond_to_query` is what keeps concurrent calls bounded — we do + // *not* serialize here. + let snippets: Arc>> = Arc::new(Mutex::new(Vec::new())); + let mut tasks = JoinSet::new(); + for (responder_id, score) in responders.iter().copied() { + let sim = sim.clone(); + let llm = llm.clone(); + let caps = caps.query_responder(); + let question = question.clone(); + let snippets = Arc::clone(&snippets); + tasks.spawn(async move { + let text = match sim.get_engram_text(galaxy, responder_id).await { + Ok(Some(t)) => t, + _ => return, + }; + let snippet = match respond_to_query(&llm, &question, &text, caps).await { + Ok(Some(s)) => s, + Ok(None) => { + debug!(?responder_id, score, "responder passed"); + return; + } + Err(e) => { + warn!(?responder_id, "responder failed: {e}"); + return; + } + }; + // Record on the engram (visual light-up + memory entry). + if let Err(e) = sim + .record_query_participation(galaxy, query_id, responder_id, snippet.clone()) + .await + { + warn!(?responder_id, "record_query_participation: {e}"); + } + snippets.lock().await.push(snippet); + }); + } + + // Run the integrator on a fixed cadence while responders are in flight. + // We don't try to "wake on snippet" — a 2 s tick is plenty fast for the + // user, and lets multiple snippets batch into one integrator call. + let mut version: u32 = 0; + let deadline = tokio::time::Instant::now() + RESPONDER_TIMEOUT; + let mut ticker = tokio::time::interval(INTEGRATION_INTERVAL); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + + loop { + tokio::select! { + biased; + // All responder tasks have completed (or panicked). + _ = wait_for_join_set(&mut tasks) => { + break; + } + _ = ticker.tick() => { + let snapshot = snippets.lock().await.clone(); + if snapshot.is_empty() { + continue; + } + version += 1; + run_integration( + &sim, &llm, caps.query_integrator(), &entry, galaxy, + query_id, version, &question, &snapshot, QueryStatus::Responding, + ).await; + } + _ = tokio::time::sleep_until(deadline) => { + warn!(?query_id, "responder timeout — finishing with partial set"); + tasks.shutdown().await; + break; + } + } + } + + let final_snippets = snippets.lock().await.clone(); + let responder_count = final_snippets.len() as u32; + + // Final integration. If we already have a recent answer and no new + // snippets arrived in the final window we still re-run once — the + // integrator is allowed to refine even with the same input. + version += 1; + let final_answer = if final_snippets.is_empty() { + "No responder produced a usable snippet.".to_string() + } else { + match integrate_responses( + &llm, + &question, + &final_snippets, + caps.query_integrator(), + ).await { + Ok(a) if !a.is_empty() => a, + Ok(_) => "Integrator returned an empty answer.".to_string(), + Err(e) => { + warn!(?query_id, "final integration failed: {e}"); + entry.snapshot().await.answer + } + } + }; + + let _ = sim + .publish_query_answer(galaxy, query_id, version, QueryStatus::Done, final_answer.clone()) + .await; + let _ = sim + .finish_query(galaxy, query_id, query_engram, final_answer.clone(), responder_count) + .await; + publish_terminal(&entry, query_id, QueryStatus::Done, final_answer, responder_count).await; +} + +/// Run one integration pass and publish the result to both the sim's WS bus +/// and the per-query SSE bus. Increments `responder_count` from the snapshot +/// length so the UI can show "synthesizing 5 perspectives…". +#[allow(clippy::too_many_arguments)] +async fn run_integration( + sim: &SimHandle, + llm: &LmStudioClient, + caps: sophia_llm::TokenCaps, + entry: &QueryEntry, + galaxy: GalaxyId, + query_id: QueryId, + version: u32, + question: &str, + snippets: &[String], + status: QueryStatus, +) { + let answer = match integrate_responses(llm, question, snippets, caps).await { + Ok(a) => a, + Err(e) => { + warn!(?query_id, "integrator failed: {e}"); + return; + } + }; + if answer.is_empty() { + return; + } + let responder_count = snippets.len() as u32; + let _ = sim + .publish_query_answer(galaxy, query_id, version, status, answer.clone()) + .await; + entry + .publish(QueryUpdate { + query_id, + version, + status, + answer, + responder_count, + }) + .await; +} + +async fn publish_terminal( + entry: &QueryEntry, + query_id: QueryId, + status: QueryStatus, + answer: String, + responder_count: u32, +) { + let snapshot = entry.snapshot().await; + let version = snapshot.version.saturating_add(1); + entry + .publish(QueryUpdate { + query_id, + version, + status, + answer, + responder_count, + }) + .await; +} + +/// Drain a `JoinSet` until empty. Used as a `select!` arm to wait for "all +/// spawned tasks done" without holding a mutable borrow across awaits. +async fn wait_for_join_set(tasks: &mut JoinSet<()>) { + while tasks.join_next().await.is_some() {} +} diff --git a/crates/sophia-sim/src/handle.rs b/crates/sophia-sim/src/handle.rs index 02b3e42..95d44b4 100644 --- a/crates/sophia-sim/src/handle.rs +++ b/crates/sophia-sim/src/handle.rs @@ -12,8 +12,10 @@ use tracing::warn; use sophia_core::{ Engram, EngramDetail, EngramId, EngramSnapshot, EngramState, GalaxyId, GalaxyInfo, GalaxyShape, - Manifest, PositionFrame, SimEvent, Slate, SynapseDto, Vec3, + GalaxySnapshot, Introspection, Manifest, Memory, MemoryKind, PositionFrame, QueryId, + QueryStatus, SimEvent, Slate, SynapseDto, Vec3, MAX_MEMORIES, }; +use std::collections::{HashMap, HashSet, VecDeque}; /// One item to ingest into the sim. The server pre-computes the slate via the /// LM Studio embedding endpoint and hands the sim a fully-formed payload. @@ -70,6 +72,23 @@ const GRAVITY_MAX_ACC: f32 = 30.0; /// than the gravity threshold so weak co-residence doesn't link everyone. const SYNAPSE_THRESHOLD: f32 = 0.62; +// ---- Stage 5: query broadcast tuning ---- +/// Minimum cosine similarity for an engram to qualify as a Phase A responder. +/// Tuned for `nomic-embed-text-v1.5`: directly relevant text typically scores +/// 0.55+, weakly related 0.35–0.50, unrelated 0.30 or below. +const QUERY_PHASE_A_THRESHOLD: f32 = 0.45; +/// Synapse weight cutoff for Phase B 1-hop expansion. A synapse weaker than +/// this isn't strong enough evidence to drag its endpoint into the conversation. +const QUERY_PHASE_B_SYNAPSE_THRESHOLD: f32 = 0.55; +/// Fraction of `max_responders` reserved for Phase A (spatial). Phase B fills +/// whatever's left so synaptic expansion always gets at least *some* slots. +const QUERY_PHASE_A_FRAC: f32 = 0.75; +/// Per-responder duration of the "lit up" Conversing visual on the WS bus. +/// The orchestrator does NOT explicitly turn the responder back to Idle — +/// instead it schedules a deferred event after this delay, keeping the sim +/// authoritative for the transition (no client-side timers required). +const RESPONDER_LIGHTUP_MS: u64 = 4_000; + #[derive(Debug, Error)] pub enum SimError { #[error("simulation has shut down")] @@ -104,6 +123,12 @@ enum SimCmd { engram: EngramId, reply: oneshot::Sender, SimError>>, }, + UpdateIntrospection { + galaxy: GalaxyId, + engram: EngramId, + introspection: Introspection, + reply: oneshot::Sender>, + }, Subscribe { galaxy: GalaxyId, reply: oneshot::Sender>, @@ -113,6 +138,107 @@ enum SimCmd { shape: GalaxyShape, reply: oneshot::Sender>, }, + /// Stage 5: materialize a pinned Query-Engram at the galaxy center. + /// Returns the new engram's id so the orchestrator can refer to it. + StartQueryEngram { + galaxy: GalaxyId, + query: QueryId, + text: String, + slate: Slate, + reply: oneshot::Sender>, + }, + /// Stage 5: run Phase A (cosine scan) + Phase B (1-hop synaptic) and + /// return the chosen responders sorted by score (high→low). Capped at + /// `max_responders`. + BroadcastQuery { + galaxy: GalaxyId, + query_engram: EngramId, + max_responders: usize, + reply: oneshot::Sender, SimError>>, + }, + /// Stage 5: fetch the source text for a responder so the orchestrator + /// can build the per-responder LLM prompt. Returns `None` for synthetic + /// engrams (no manifest). + GetEngramText { + galaxy: GalaxyId, + engram: EngramId, + reply: oneshot::Sender, SimError>>, + }, + /// Stage 5: a responder produced a snippet for a query. Append a memory + /// on the responder, briefly flip its state to `Conversing` (auto-reverts + /// after `RESPONDER_LIGHTUP_MS`). + RecordQueryParticipation { + galaxy: GalaxyId, + query: QueryId, + engram: EngramId, + snippet: String, + reply: oneshot::Sender>, + }, + /// Stage 5: publish a new running answer over the WS bus. The orchestrator + /// also pushes to its per-query SSE channel separately — this event is + /// only for clients that want to subscribe over the global galaxy bus. + PublishQueryAnswer { + galaxy: GalaxyId, + query: QueryId, + version: u32, + status: QueryStatus, + answer: String, + reply: oneshot::Sender>, + }, + /// Stage 5: query is done. Transition the Query-Engram to `Memorize`, + /// store the final answer in its memory log, emit `QueryFinished`. + FinishQuery { + galaxy: GalaxyId, + query: QueryId, + query_engram: EngramId, + final_answer: String, + responder_count: u32, + reply: oneshot::Sender>, + }, + /// Stage 6: snapshot every galaxy to a serializable form. Used by the + /// periodic snapshot task in the bin to write to the sled store. Pinned + /// (query) engrams are excluded — they're ephemeral. + SnapshotAll { + reply: oneshot::Sender>, + }, + /// Stage 6: re-create a galaxy in the world from a stored snapshot. + /// Re-uses every engram's stored `instance_idx` so the dense slot table + /// is identical to pre-shutdown. Re-schedules `BroadcastFrame`, + /// `RebuildIndex`, and one `EngramTick` per (non-pinned) engram so motion + /// resumes immediately. Returns `InvalidShape` if the snapshot's shape + /// fails validation. + HydrateGalaxy { + snapshot: GalaxySnapshot, + reply: oneshot::Sender>, + }, + /// Stage 6: remove a galaxy from the world. Caller is responsible for + /// also removing its on-disk snapshot (see `Store::delete_galaxy`). + DeleteGalaxy { + galaxy: GalaxyId, + reply: oneshot::Sender>, + }, + /// Stage 7: cheap stats snapshot for the HUD. Returns total events + /// processed per galaxy in the last second + total engram count. + /// Polled at 1 Hz from the browser. + GetStats { + reply: oneshot::Sender, + }, +} + +/// Lightweight stats payload returned by `SimHandle::stats`. Used by the +/// HUD's "events/sec" + "engrams" counters. Numbers are best-effort +/// snapshots, not exact (the sim counts as it processes; the read happens +/// asynchronously). +#[derive(Debug, Clone, Default)] +pub struct SimStats { + /// Sum of ticks across all galaxies in the last 1-second window. + pub ticks_per_sec: u32, + /// Sum of all non-pinned engrams across galaxies. + pub engrams_total: u32, + /// Number of live galaxies. + pub galaxies: u32, + /// Sum of all synapses across galaxies. + pub synapses_total: u32, } #[derive(Clone)] @@ -165,6 +291,22 @@ impl SimHandle { rx.await.map_err(|_| SimError::Shutdown)? } + /// Apply async-LLM-generated introspection to an existing engram. Called + /// from the server after a background introspection task completes. + pub async fn update_introspection( + &self, + galaxy: GalaxyId, + engram: EngramId, + introspection: Introspection, + ) -> Result<(), SimError> { + let (reply, rx) = oneshot::channel(); + self.tx + .send(SimCmd::UpdateIntrospection { galaxy, engram, introspection, reply }) + .await + .map_err(|_| SimError::Shutdown)?; + rx.await.map_err(|_| SimError::Shutdown)? + } + /// Subscribe to a galaxy's event stream. Returns initial state plus the /// live receiver so the WS client can replay then follow. pub async fn subscribe( @@ -190,6 +332,151 @@ impl SimHandle { .map_err(|_| SimError::Shutdown)?; rx.await.map_err(|_| SimError::Shutdown)? } + + /// Stage 5: materialize a Query-Engram at the galaxy center. + pub async fn start_query_engram( + &self, + galaxy: GalaxyId, + query: QueryId, + text: String, + slate: Slate, + ) -> Result { + let (reply, rx) = oneshot::channel(); + self.tx + .send(SimCmd::StartQueryEngram { galaxy, query, text, slate, reply }) + .await + .map_err(|_| SimError::Shutdown)?; + rx.await.map_err(|_| SimError::Shutdown)? + } + + /// Stage 5: select up to `max_responders` engrams whose slates resonate + /// with the query, optionally extending via 1-hop synaptic neighbours. + pub async fn broadcast_query( + &self, + galaxy: GalaxyId, + query_engram: EngramId, + max_responders: usize, + ) -> Result, SimError> { + let (reply, rx) = oneshot::channel(); + self.tx + .send(SimCmd::BroadcastQuery { galaxy, query_engram, max_responders, reply }) + .await + .map_err(|_| SimError::Shutdown)?; + rx.await.map_err(|_| SimError::Shutdown)? + } + + /// Stage 5: fetch the source text of an engram (for building the + /// per-responder LLM prompt). Returns `None` for synthetic engrams. + pub async fn get_engram_text( + &self, + galaxy: GalaxyId, + engram: EngramId, + ) -> Result, SimError> { + let (reply, rx) = oneshot::channel(); + self.tx + .send(SimCmd::GetEngramText { galaxy, engram, reply }) + .await + .map_err(|_| SimError::Shutdown)?; + rx.await.map_err(|_| SimError::Shutdown)? + } + + /// Stage 5: record a responder's contribution + light it up. + pub async fn record_query_participation( + &self, + galaxy: GalaxyId, + query: QueryId, + engram: EngramId, + snippet: String, + ) -> Result<(), SimError> { + let (reply, rx) = oneshot::channel(); + self.tx + .send(SimCmd::RecordQueryParticipation { galaxy, query, engram, snippet, reply }) + .await + .map_err(|_| SimError::Shutdown)?; + rx.await.map_err(|_| SimError::Shutdown)? + } + + /// Stage 5: publish an updated running answer to WS subscribers. + pub async fn publish_query_answer( + &self, + galaxy: GalaxyId, + query: QueryId, + version: u32, + status: QueryStatus, + answer: String, + ) -> Result<(), SimError> { + let (reply, rx) = oneshot::channel(); + self.tx + .send(SimCmd::PublishQueryAnswer { galaxy, query, version, status, answer, reply }) + .await + .map_err(|_| SimError::Shutdown)?; + rx.await.map_err(|_| SimError::Shutdown)? + } + + /// Stage 5: terminal step — store the final answer in the Query-Engram's + /// memory log, transition to Memorize, emit QueryFinished. + pub async fn finish_query( + &self, + galaxy: GalaxyId, + query: QueryId, + query_engram: EngramId, + final_answer: String, + responder_count: u32, + ) -> Result<(), SimError> { + let (reply, rx) = oneshot::channel(); + self.tx + .send(SimCmd::FinishQuery { + galaxy, + query, + query_engram, + final_answer, + responder_count, + reply, + }) + .await + .map_err(|_| SimError::Shutdown)?; + rx.await.map_err(|_| SimError::Shutdown)? + } + + /// Stage 6: snapshot every galaxy. Used by the periodic snapshot task. + pub async fn snapshot_all(&self) -> Result, SimError> { + let (reply, rx) = oneshot::channel(); + self.tx + .send(SimCmd::SnapshotAll { reply }) + .await + .map_err(|_| SimError::Shutdown)?; + rx.await.map_err(|_| SimError::Shutdown) + } + + /// Stage 6: re-create a galaxy from a stored snapshot. + pub async fn hydrate_galaxy(&self, snapshot: GalaxySnapshot) -> Result { + let (reply, rx) = oneshot::channel(); + self.tx + .send(SimCmd::HydrateGalaxy { snapshot, reply }) + .await + .map_err(|_| SimError::Shutdown)?; + rx.await.map_err(|_| SimError::Shutdown)? + } + + /// Stage 6: drop a galaxy from the world. + pub async fn delete_galaxy(&self, galaxy: GalaxyId) -> Result<(), SimError> { + let (reply, rx) = oneshot::channel(); + self.tx + .send(SimCmd::DeleteGalaxy { galaxy, reply }) + .await + .map_err(|_| SimError::Shutdown)?; + rx.await.map_err(|_| SimError::Shutdown)? + } + + /// Stage 7: snapshot of recent activity for the HUD. Cheap — never + /// blocks the sim; falls back to defaults if the sim is shutting down. + pub async fn stats(&self) -> SimStats { + let (reply, rx) = oneshot::channel(); + if self.tx.send(SimCmd::GetStats { reply }).await.is_err() { + return SimStats::default(); + } + rx.await.unwrap_or_default() + } } /// Spawn the simulation task. `default_shape` is used for any new galaxy @@ -203,10 +490,22 @@ pub fn spawn_sim(default_shape: GalaxyShape) -> SimHandle { let mut indexes: std::collections::HashMap = Default::default(); let mut rng = SmallRng::seed_from_u64(0xC0DE_5071); let started = Instant::now(); + // Stage 7 stats: rolling 1-second tick counter. We bump + // `tick_window_count` for every event we drain, and roll it into + // `ticks_per_sec` once a second has elapsed since `tick_window_start`. + let mut tick_window_start = Instant::now(); + let mut tick_window_count: u32 = 0; + let mut ticks_per_sec: u32 = 0; loop { // Pick whichever happens first: a new command or the next due event. let now = Instant::now(); + // Roll the tick window if a second has elapsed. + if now.duration_since(tick_window_start) >= Duration::from_secs(1) { + ticks_per_sec = tick_window_count; + tick_window_count = 0; + tick_window_start = now; + } let next_at = scheduler.next_at(); let timeout = match next_at { Some(at) => at.saturating_duration_since(now), @@ -215,13 +514,17 @@ pub fn spawn_sim(default_shape: GalaxyShape) -> SimHandle { tokio::select! { cmd = rx.recv() => { let Some(cmd) = cmd else { break }; // all handles dropped - handle_cmd(cmd, &mut world, &mut scheduler, &mut indexes, &mut rng, default_shape); + handle_cmd( + cmd, &mut world, &mut scheduler, &mut indexes, + &mut rng, default_shape, started, ticks_per_sec, + ); } _ = tokio::time::sleep(timeout) => { // Drain all due events. let now = Instant::now(); while let Some(event) = scheduler.pop_due(now) { handle_event(event, &mut world, &mut scheduler, &mut indexes, &mut rng, started); + tick_window_count = tick_window_count.saturating_add(1); } } } @@ -230,6 +533,7 @@ pub fn spawn_sim(default_shape: GalaxyShape) -> SimHandle { SimHandle { tx } } +#[allow(clippy::too_many_arguments)] fn handle_cmd( cmd: SimCmd, world: &mut World, @@ -237,6 +541,8 @@ fn handle_cmd( indexes: &mut std::collections::HashMap, _rng: &mut SmallRng, default_shape: GalaxyShape, + started: Instant, + ticks_per_sec: u32, ) { match cmd { SimCmd::CreateGalaxy { name, reply } => { @@ -280,6 +586,9 @@ fn handle_cmd( manifest: e.manifest.clone(), slate_dim: e.slate.as_ref().map(|s| s.dim()), slate_norm: e.slate.as_ref().map(|s| s.norm()), + introspection: e.introspection.clone(), + memories: e.memories.iter().cloned().collect(), + pinned: e.pinned, }) }); match detail { @@ -291,6 +600,19 @@ fn handle_cmd( } } } + SimCmd::UpdateIntrospection { galaxy, engram, introspection, reply } => { + let res = world + .galaxies + .get_mut(&galaxy) + .ok_or(SimError::UnknownGalaxy) + .and_then(|g| { + let e = g.engrams.get_mut(&engram).ok_or(SimError::UnknownGalaxy)?; + e.introspection = introspection; + push_memory(&mut e.memories, ms_since(started), MemoryKind::Introspected); + Ok(()) + }); + let _ = reply.send(res); + } SimCmd::Subscribe { galaxy, reply } => { let res = world .galaxies @@ -308,6 +630,186 @@ fn handle_cmd( }; let _ = reply.send(res); } + SimCmd::StartQueryEngram { galaxy, query, text, slate, reply } => { + let res = world + .galaxies + .get_mut(&galaxy) + .ok_or(SimError::UnknownGalaxy) + .map(|g| materialize_query_engram(g, galaxy, query, text, slate, started)); + let _ = reply.send(res); + } + SimCmd::BroadcastQuery { galaxy, query_engram, max_responders, reply } => { + let res = world + .galaxies + .get(&galaxy) + .ok_or(SimError::UnknownGalaxy) + .map(|g| compute_query_responders(g, query_engram, max_responders)); + let _ = reply.send(res); + } + SimCmd::GetEngramText { galaxy, engram, reply } => { + let res = world + .galaxies + .get(&galaxy) + .ok_or(SimError::UnknownGalaxy) + .map(|g| { + g.engrams.get(&engram).and_then(|e| { + e.manifest + .as_ref() + .map(|Manifest::Text { content }| content.clone()) + }) + }); + let _ = reply.send(res); + } + SimCmd::RecordQueryParticipation { galaxy, query, engram, snippet, reply } => { + let res = world + .galaxies + .get_mut(&galaxy) + .ok_or(SimError::UnknownGalaxy) + .map(|g| { + if let Some(e) = g.engrams.get_mut(&engram) { + push_memory( + &mut e.memories, + ms_since(started), + MemoryKind::QueryParticipated { query, snippet }, + ); + e.state = EngramState::Conversing; + g.emit(SimEvent::EngramStateChanged { + id: engram, + state: EngramState::Conversing, + }); + // Schedule the auto-revert. If the engram gets re-lit + // for a different query before this fires, the + // `expected` guard will skip the revert. + scheduler.schedule( + Instant::now() + Duration::from_millis(RESPONDER_LIGHTUP_MS), + Event::RevertState { + galaxy, + engram, + expected: EngramState::Conversing, + revert_to: EngramState::Idle, + }, + ); + } + }); + let _ = reply.send(res); + } + SimCmd::PublishQueryAnswer { galaxy, query, version, status, answer, reply } => { + let res = world + .galaxies + .get(&galaxy) + .ok_or(SimError::UnknownGalaxy) + .map(|g| { + g.emit(SimEvent::QueryAnswerUpdated { query, version, status, answer }); + }); + let _ = reply.send(res); + } + SimCmd::FinishQuery { + galaxy, query, query_engram, final_answer, responder_count, reply, + } => { + let res = world + .galaxies + .get_mut(&galaxy) + .ok_or(SimError::UnknownGalaxy) + .map(|g| { + if let Some(e) = g.engrams.get_mut(&query_engram) { + e.state = EngramState::Memorize; + push_memory( + &mut e.memories, + ms_since(started), + MemoryKind::QueryAnswered { + answer: final_answer, + responder_count, + }, + ); + g.emit(SimEvent::EngramStateChanged { + id: query_engram, + state: EngramState::Memorize, + }); + } + g.emit(SimEvent::QueryFinished { + query, + status: QueryStatus::Done, + responder_count, + }); + }); + let _ = reply.send(res); + } + SimCmd::SnapshotAll { reply } => { + let snaps = world + .galaxies + .values() + .map(|g| g.to_snapshot()) + .collect(); + let _ = reply.send(snaps); + } + SimCmd::HydrateGalaxy { snapshot, reply } => { + let res = match snapshot.galaxy.shape.validate() { + Err(msg) => Err(SimError::InvalidShape(msg)), + Ok(()) => { + let id = snapshot.galaxy.id; + let state = GalaxyState::from_snapshot(snapshot); + let info = state.info(); + state.emit(state.torus_event()); + // Re-schedule the periodic galaxy events first so they + // start firing at their normal cadence. + let now = Instant::now(); + scheduler.schedule(now + FRAME_INTERVAL, Event::BroadcastFrame { galaxy: id }); + scheduler.schedule(now + REBUILD_INTERVAL, Event::RebuildIndex { galaxy: id }); + // One EngramTick per non-pinned engram. Stagger them + // across one tick interval so they don't all fire + // simultaneously and clobber the scheduler heap on + // boot — also gives the spatial index time to rebuild + // before gravity kicks in. + let n = state.engrams.len().max(1) as u64; + let stagger_step = TICK_INTERVAL.as_micros() as u64 / n.max(1); + for (i, engram_id) in state + .engrams + .keys() + .copied() + .enumerate() + { + let offset = Duration::from_micros(stagger_step * i as u64); + scheduler.schedule( + now + TICK_INTERVAL + offset, + Event::EngramTick { galaxy: id, engram: engram_id }, + ); + } + indexes.insert(id, KiddoIndex::empty()); + world.galaxies.insert(id, state); + Ok(info) + } + }; + let _ = reply.send(res); + } + SimCmd::DeleteGalaxy { galaxy, reply } => { + let res = if world.galaxies.remove(&galaxy).is_some() { + indexes.remove(&galaxy); + Ok(()) + } else { + Err(SimError::UnknownGalaxy) + }; + let _ = reply.send(res); + } + SimCmd::GetStats { reply } => { + let mut engrams_total: u32 = 0; + let mut synapses_total: u32 = 0; + for g in world.galaxies.values() { + // Pinned (query) engrams are excluded from the public count + // so the HUD doesn't blink during queries. + engrams_total += g + .engrams + .values() + .filter(|e| !e.pinned) + .count() as u32; + synapses_total += g.synapses.len() as u32; + } + let _ = reply.send(SimStats { + ticks_per_sec, + engrams_total, + galaxies: world.galaxies.len() as u32, + synapses_total, + }); + } } } @@ -387,6 +889,56 @@ fn ingest_galaxy( Ok(ids) } +/// Materialize a Query-Engram (Stage 5): pinned at the galaxy center with the +/// question text + slate, state `Searching`. Returns the new engram's id. No +/// `EngramTick` is scheduled — pinned engrams don't move and don't form +/// spontaneous synapses; their behaviour is driven entirely by the orchestrator. +fn materialize_query_engram( + g: &mut GalaxyState, + _galaxy: GalaxyId, + _query: QueryId, + text: String, + slate: Slate, + started: Instant, +) -> EngramId { + let id = EngramId::new(); + let instance_idx = g.slot_to_id.len() as u32; + let position = g.galaxy.center; + let mut memories = VecDeque::new(); + push_memory(&mut memories, ms_since(started), MemoryKind::Born); + let engram = Engram { + id, + instance_idx, + position, + velocity: Vec3::ZERO, + size: 1.0, + state: EngramState::Searching, + age: 0, + manifest: Some(Manifest::Text { content: text.clone() }), + slate: Some(slate), + introspection: Introspection::default(), + memories, + pinned: true, + }; + g.slot_to_id.push(id); + g.engrams.insert(id, engram.clone()); + let snapshot = EngramSnapshot { + id, + instance_idx, + position: position.to_array(), + size: engram.size, + state: engram.state, + }; + g.emit(SimEvent::EngramCreated { snapshot }); + g.emit(SimEvent::QueryStarted { + query: _query, + engram: id, + position: position.to_array(), + text, + }); + id +} + /// Materialize one engram with the pre-allocated id at the galaxy's birth /// point with a random initial velocity, emit `EngramCreated`, and schedule /// its first `EngramTick`. @@ -397,6 +949,7 @@ fn materialize_engram( payload: SpawnPayload, scheduler: &mut Scheduler, rng: &mut SmallRng, + started: Instant, ) { let instance_idx = g.slot_to_id.len() as u32; let birth = g.galaxy.shape.birth_point(g.galaxy.center); @@ -406,6 +959,8 @@ fn materialize_engram( SpawnPayload::Synthetic => (None, None), SpawnPayload::Manifested { manifest, slate } => (Some(manifest), Some(slate)), }; + let mut memories = VecDeque::new(); + push_memory(&mut memories, ms_since(started), MemoryKind::Born); let engram = Engram { id, instance_idx, @@ -416,6 +971,9 @@ fn materialize_engram( age: 0, manifest, slate, + introspection: Introspection::default(), + memories, + pinned: false, }; g.slot_to_id.push(id); g.engrams.insert(id, engram.clone()); @@ -433,6 +991,19 @@ fn materialize_engram( ); } +/// Append a memory to the front-bounded VecDeque, evicting the oldest entry +/// once `MAX_MEMORIES` is reached. +fn push_memory(memories: &mut VecDeque, at_ms: u32, kind: MemoryKind) { + while memories.len() >= MAX_MEMORIES { + memories.pop_front(); + } + memories.push_back(Memory { at_ms, kind }); +} + +fn ms_since(started: Instant) -> u32 { + started.elapsed().as_millis() as u32 +} + /// Walk the kiddo index for `engram_id`'s neighbours and return: /// - `gravity_acc`: cosine-weighted attraction toward similar peers (only /// contributes inside the tube; physics gates this further by position). @@ -492,6 +1063,97 @@ fn compute_gravity_and_candidates( (acc, candidates) } +/// Stage 5 broadcast retrieval. Two passes: +/// +/// **Phase A — global cosine scan.** Score every non-pinned engram's slate +/// against the query, keep those above `QUERY_PHASE_A_THRESHOLD`, take the +/// top `QUERY_PHASE_A_FRAC * max_responders` by score. +/// +/// **Phase B — 1-hop synaptic expansion.** For each Phase A pick, walk its +/// synapses and add any neighbour we haven't already chosen. The neighbour's +/// effective score is `cosine_to_query * synapse_weight` (synaptic +/// attenuation per the system-analysis doc). Fill the remaining +/// `max_responders - phase_a` slots in score-sorted order. +/// +/// We do NOT use the spatial kiddo index here. Query-Engrams sit at the +/// galaxy center while normal engrams cluster in the tube ~major_radius +/// away — a single radius doesn't cover both. A linear scan over engrams +/// is fine at MVP scale (a few thousand) and avoids per-query index +/// rebuilds. The kiddo index stays where it earns its keep: per-tick +/// gravity for engrams already in the tube. +fn compute_query_responders( + g: &GalaxyState, + query_engram: EngramId, + max_responders: usize, +) -> Vec<(EngramId, f32)> { + if max_responders == 0 { + return Vec::new(); + } + let Some(q) = g.engrams.get(&query_engram) else { return Vec::new(); }; + let Some(q_slate) = q.slate.as_ref() else { return Vec::new(); }; + + // Phase A. + let mut scored: Vec<(EngramId, f32)> = g + .engrams + .values() + .filter(|e| e.id != query_engram && !e.pinned) + .filter_map(|e| { + let s = e.slate.as_ref()?; + let cos = q_slate.cosine(s); + (cos >= QUERY_PHASE_A_THRESHOLD).then_some((e.id, cos)) + }) + .collect(); + scored.sort_by(|a, b| { + b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal) + }); + let phase_a_budget = ((max_responders as f32) * QUERY_PHASE_A_FRAC).ceil() as usize; + let phase_a_budget = phase_a_budget.min(max_responders).max(1); + let mut chosen: Vec<(EngramId, f32)> = scored.into_iter().take(phase_a_budget).collect(); + let mut chosen_set: HashSet = + chosen.iter().map(|(id, _)| *id).collect(); + // Always exclude the query-engram itself from synaptic expansion just in + // case some legacy synapse pointed at it. + chosen_set.insert(query_engram); + + // Phase B: build adjacency once, then walk. + let phase_b_budget = max_responders.saturating_sub(chosen.len()); + if phase_b_budget == 0 || g.synapses.is_empty() { + return chosen; + } + let mut adjacency: HashMap> = + HashMap::with_capacity(g.engrams.len()); + for syn in g.synapses.values() { + if syn.weight < QUERY_PHASE_B_SYNAPSE_THRESHOLD { + continue; + } + adjacency.entry(syn.a).or_default().push((syn.b, syn.weight)); + adjacency.entry(syn.b).or_default().push((syn.a, syn.weight)); + } + let mut additions: Vec<(EngramId, f32)> = Vec::new(); + for (seed_id, _) in chosen.iter() { + if let Some(neighbours) = adjacency.get(seed_id) { + for (neighbour, weight) in neighbours { + if !chosen_set.insert(*neighbour) { + continue; + } + let Some(n) = g.engrams.get(neighbour) else { continue; }; + if n.pinned { + continue; + } + let Some(n_slate) = n.slate.as_ref() else { continue; }; + let effective = q_slate.cosine(n_slate) * weight; + additions.push((*neighbour, effective)); + } + } + } + // Take the strongest Phase B additions to fill remaining slots. + additions.sort_by(|a, b| { + b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal) + }); + chosen.extend(additions.into_iter().take(phase_b_budget)); + chosen +} + /// Closest point on the torus centerline to `p` in xy-plane (mirrors the /// math in `physics::spine_point` but doesn't pull `physics` in here). fn nearest_spine_xy(p: Vec3, major_r: f32) -> Vec3 { @@ -515,7 +1177,7 @@ fn handle_event( match event { Event::Spawn { galaxy, id, payload } => { let Some(g) = world.galaxies.get_mut(&galaxy) else { return }; - materialize_engram(g, galaxy, id, payload, scheduler, rng); + materialize_engram(g, galaxy, id, payload, scheduler, rng, started); } Event::EngramTick { galaxy, engram } => { let Some(g) = world.galaxies.get_mut(&galaxy) else { return }; @@ -535,11 +1197,27 @@ fn handle_event( return; } - // After the tick, materialise any qualifying synapses and emit - // events for the newly-formed ones. + // After the tick, materialise any qualifying synapses, emit + // events for the newly-formed ones, and append a memory entry + // on both endpoints so the inspector can show them. for (peer, weight) in synapse_candidates { if let Some(syn) = g.try_form_synapse(engram, peer, weight) { g.emit(SimEvent::SynapseCreated { synapse: SynapseDto::from(&syn) }); + let now_ms = ms_since(started); + if let Some(e_a) = g.engrams.get_mut(&syn.a) { + push_memory( + &mut e_a.memories, + now_ms, + MemoryKind::SynapseFormed { with: syn.b, weight }, + ); + } + if let Some(e_b) = g.engrams.get_mut(&syn.b) { + push_memory( + &mut e_b.memories, + now_ms, + MemoryKind::SynapseFormed { with: syn.a, weight }, + ); + } } } @@ -568,11 +1246,14 @@ fn handle_event( } Event::RebuildIndex { galaxy } => { // Index rebuild only — no bbox recompute (the torus is fixed-size, - // resized only via SimCmd::Resize). + // resized only via SimCmd::Resize). Pinned (query) engrams are + // skipped: the index drives gravity + per-tick neighbour search, + // and queries don't participate in either. if let Some(g) = world.galaxies.get(&galaxy) { let points: Vec<(EngramId, [f32; 3])> = g .engrams .values() + .filter(|e| !e.pinned) .map(|e| (e.id, e.position.to_array())) .collect(); if let Some(idx) = indexes.get_mut(&galaxy) { @@ -584,5 +1265,15 @@ fn handle_event( Event::RebuildIndex { galaxy }, ); } + Event::RevertState { galaxy, engram, expected, revert_to } => { + if let Some(g) = world.galaxies.get_mut(&galaxy) { + if let Some(e) = g.engrams.get_mut(&engram) { + if e.state == expected { + e.state = revert_to; + g.emit(SimEvent::EngramStateChanged { id: engram, state: revert_to }); + } + } + } + } } } diff --git a/crates/sophia-sim/src/lib.rs b/crates/sophia-sim/src/lib.rs index 75b9536..f0fa4da 100644 --- a/crates/sophia-sim/src/lib.rs +++ b/crates/sophia-sim/src/lib.rs @@ -9,4 +9,4 @@ mod physics; mod scheduler; mod world; -pub use handle::{spawn_sim, IngestItem, SimError, SimHandle}; +pub use handle::{spawn_sim, IngestItem, SimError, SimHandle, SimStats}; diff --git a/crates/sophia-sim/src/physics.rs b/crates/sophia-sim/src/physics.rs index 1d74dc6..6ed0942 100644 --- a/crates/sophia-sim/src/physics.rs +++ b/crates/sophia-sim/src/physics.rs @@ -75,6 +75,14 @@ pub fn tick( gravity_acc: Vec3, rng: &mut SmallRng, ) { + // Pinned engrams (Stage 5 Query-Engrams) sit perfectly still at their + // birth position — skip the entire integration. Age still advances so + // memory timestamps and any future age-gated logic stay consistent. + if engram.pinned { + engram.age = engram.age.saturating_add(1); + return; + } + let curiosity_factor = (-(engram.age as f32) / CURIOSITY_TAU_TICKS).exp(); // Curiosity: random impulse, decaying with age. diff --git a/crates/sophia-sim/src/scheduler.rs b/crates/sophia-sim/src/scheduler.rs index 36d9956..3688dc0 100644 --- a/crates/sophia-sim/src/scheduler.rs +++ b/crates/sophia-sim/src/scheduler.rs @@ -4,7 +4,7 @@ use std::cmp::Ordering; use std::collections::BinaryHeap; use std::time::Instant; -use sophia_core::{EngramId, GalaxyId, Manifest, Slate}; +use sophia_core::{EngramId, EngramState, GalaxyId, Manifest, Slate}; /// Payload for a queued spawn. Synthetic seed engrams have no manifest/slate; /// ingested engrams carry the pre-embedded text. @@ -28,6 +28,19 @@ pub enum Event { /// Recompute density-driven bbox + rebuild spatial index. Fires every /// ~500 ms and reschedules itself. RebuildIndex { galaxy: GalaxyId }, + /// Stage 5: revert an engram's state (typically Conversing → Idle) + /// after a responder's light-up window expires. Fired once and not + /// rescheduled. The `expected` state guards against racing transitions + /// — only revert if the engram is still in the state the lit-up event + /// originally set (the orchestrator may light up the same engram twice + /// for two queries; we don't want the older revert to clobber a fresher + /// one). + RevertState { + galaxy: GalaxyId, + engram: EngramId, + expected: EngramState, + revert_to: EngramState, + }, } #[derive(Debug)] diff --git a/crates/sophia-sim/src/world.rs b/crates/sophia-sim/src/world.rs index 1615b91..e818534 100644 --- a/crates/sophia-sim/src/world.rs +++ b/crates/sophia-sim/src/world.rs @@ -4,7 +4,7 @@ use std::collections::{HashMap, HashSet}; use sophia_core::{ canonical_pair, Engram, EngramId, EngramSnapshot, Galaxy, GalaxyId, GalaxyInfo, GalaxyShape, - SimEvent, Synapse, SynapseDto, SynapseId, + GalaxySnapshot, SimEvent, Synapse, SynapseDto, SynapseId, }; use tokio::sync::broadcast; @@ -120,6 +120,82 @@ impl GalaxyState { minor_radius: self.galaxy.shape.minor_radius, } } + + /// Serialize the persistent half of this galaxy's state. Pinned engrams + /// (Stage 5 Query-Engrams) are dropped — queries are ephemeral and + /// shouldn't outlive the orchestrator that owns them. Synapses that + /// pointed at a pinned engram (none currently exist; future-proofing) + /// are also dropped. + pub fn to_snapshot(&self) -> GalaxySnapshot { + let engrams: Vec = self + .slot_to_id + .iter() + .filter_map(|id| self.engrams.get(id)) + .filter(|e| !e.pinned) + .cloned() + .collect(); + let kept_ids: HashSet = engrams.iter().map(|e| e.id).collect(); + let synapses: Vec = self + .synapses + .values() + .filter(|s| kept_ids.contains(&s.a) && kept_ids.contains(&s.b)) + .cloned() + .collect(); + GalaxySnapshot { + galaxy: self.galaxy.clone(), + engrams, + synapses, + } + } + + /// Re-instantiate a galaxy from a stored snapshot. Re-uses each engram's + /// stored `instance_idx` so the dense slot table matches pre-shutdown + /// (the WS PositionFrame slot ordering depends on it). The caller (sim + /// loop) is responsible for re-scheduling per-engram ticks and the + /// recurring `BroadcastFrame` / `RebuildIndex` events. + pub fn from_snapshot(snap: GalaxySnapshot) -> Self { + let (bus, _) = broadcast::channel(BROADCAST_CAPACITY); + let mut engrams: HashMap = HashMap::with_capacity(snap.engrams.len()); + let mut max_idx: u32 = 0; + for e in &snap.engrams { + if e.instance_idx + 1 > max_idx { + max_idx = e.instance_idx + 1; + } + } + // Build a dense slot_to_id keyed by the stored instance_idx. Any + // gaps (shouldn't happen but defensive) get a placeholder UUID that + // will simply never resolve to an engram in `engrams.get(...)`, + // keeping the indexing math correct. + let mut slot_to_id: Vec = vec![EngramId::default(); max_idx as usize]; + for e in snap.engrams { + let idx = e.instance_idx as usize; + if idx < slot_to_id.len() { + slot_to_id[idx] = e.id; + } + engrams.insert(e.id, e); + } + let mut synapses: HashMap = HashMap::with_capacity(snap.synapses.len()); + let mut synapse_pairs: HashSet<(EngramId, EngramId)> = HashSet::with_capacity(snap.synapses.len()); + let mut synapse_count: HashMap = HashMap::new(); + for syn in snap.synapses { + let pair = canonical_pair(syn.a, syn.b); + if !synapse_pairs.insert(pair) { + continue; + } + *synapse_count.entry(pair.0).or_insert(0) += 1; + *synapse_count.entry(pair.1).or_insert(0) += 1; + synapses.insert(syn.id, syn); + } + Self { + galaxy: snap.galaxy, + engrams, + slot_to_id, + synapses, + synapse_pairs, + synapse_count, + bus, + } + } } pub struct World { diff --git a/crates/sophia-store/Cargo.toml b/crates/sophia-store/Cargo.toml index 9737062..e2df0ab 100644 --- a/crates/sophia-store/Cargo.toml +++ b/crates/sophia-store/Cargo.toml @@ -10,5 +10,8 @@ authors.workspace = true sophia-core = { workspace = true } anyhow = { workspace = true } +thiserror = { workspace = true } tracing = { workspace = true } serde = { workspace = true } +serde_json = { workspace = true } +sled = { workspace = true } diff --git a/crates/sophia-store/src/lib.rs b/crates/sophia-store/src/lib.rs index a9b65ae..be7e815 100644 --- a/crates/sophia-store/src/lib.rs +++ b/crates/sophia-store/src/lib.rs @@ -1,7 +1,109 @@ -//! Persistence: sled-backed event log + snapshot. +//! Persistence: sled-backed snapshot store (Stage 6). //! -//! See `docs/system-analysis.md` §13.3 (event-sourced + graph read model). -//! Per the doc: "the event log == the canonical memory store" — anything -//! memorable is persisted; transient state is not. +//! Each galaxy is one key in a sled `Db`: key = the galaxy's UUID bytes, +//! value = a JSON-encoded `GalaxySnapshot`. The bin's snapshot task +//! overwrites the value every 60 s; on boot we iterate the tree, decode +//! each entry, and hand them to the sim for hydration. //! -//! Stage 0: stub. Real content arrives in Stage 6. +//! **Why JSON, not bincode**: `Manifest` and `MemoryKind` are +//! internally-tagged enums (`#[serde(tag = "kind")]`). Internally tagged +//! enums require a self-describing format on the wire — bincode calls +//! `deserialize_any` to read the tag, which it doesn't support. JSON is +//! self-describing and handles them natively. At MVP scale (a few hundred +//! engrams) the verbosity is irrelevant; if storage grows we can swap in +//! msgpack/cbor without touching the sim. +//! +//! See `docs/system-analysis.md` §13.3: the design calls for an event log +//! with periodic snapshots, but per the same section "no +//! event-replay-from-genesis." For MVP that means we only ever need the +//! snapshot side — no event log, no per-event append. If the model later +//! needs time-travel debugging we can add the log without disturbing the +//! snapshot path (different sled tree). +//! +//! Sled defaults to `flush_every_ms = 500`, so even a hard kill loses at +//! most ~half a second of writes — acceptable for snapshot data that's +//! already 60 s stale by definition. + +use std::path::Path; + +use sophia_core::{GalaxyId, GalaxySnapshot}; +use thiserror::Error; +use tracing::{debug, warn}; + +#[derive(Debug, Error)] +pub enum StoreError { + #[error("sled error: {0}")] + Sled(#[from] sled::Error), + #[error("json error: {0}")] + Json(#[from] serde_json::Error), +} + +/// Cheaply cloneable handle to the on-disk store. Internally holds a +/// `sled::Db`; cloning shares the same database connection. +#[derive(Clone)] +pub struct Store { + db: sled::Db, +} + +impl Store { + /// Open or create the store at `path`. The path is treated as a + /// directory — sled creates it (and the sled internal files inside) if + /// needed. + pub fn open(path: impl AsRef) -> Result { + let db = sled::open(path.as_ref())?; + Ok(Self { db }) + } + + /// Iterate every stored galaxy snapshot. Decode failures are logged and + /// skipped — a corrupt entry shouldn't take down the boot path. + pub fn list_galaxies(&self) -> Result, StoreError> { + let mut out = Vec::new(); + for kv in self.db.iter() { + let (k, v) = kv?; + match serde_json::from_slice::(&v) { + Ok(snap) => out.push(snap), + Err(e) => { + warn!("skipping corrupt galaxy entry (key={} bytes): {e}", k.len()); + } + } + } + Ok(out) + } + + /// Insert or overwrite the stored snapshot for one galaxy. Sled flushes + /// asynchronously (`flush_every_ms`); we don't `flush()` here because + /// snapshot writes happen on a 60 s cadence and a half-second of in-mem + /// buffering is fine. + pub fn save_galaxy(&self, snapshot: &GalaxySnapshot) -> Result<(), StoreError> { + let bytes = serde_json::to_vec(snapshot)?; + self.db.insert(galaxy_key(snapshot.galaxy.id), bytes)?; + debug!( + "saved galaxy {:?} ({} engrams, {} synapses)", + snapshot.galaxy.id, + snapshot.engrams.len(), + snapshot.synapses.len() + ); + Ok(()) + } + + /// Remove a galaxy from disk. Idempotent — missing keys are not an error. + pub fn delete_galaxy(&self, id: GalaxyId) -> Result<(), StoreError> { + self.db.remove(galaxy_key(id))?; + Ok(()) + } + + /// Force a synchronous flush. Called on graceful shutdown (and useful + /// in tests). Otherwise sled's background flusher is sufficient. + pub fn flush(&self) -> Result<(), StoreError> { + self.db.flush()?; + Ok(()) + } +} + +/// 16-byte key for one galaxy. Using the raw UUID bytes (rather than the +/// hyphenated string) keeps the key compact and lexicographically equivalent +/// to UUID v7's time order — which makes `Db::iter()` walk galaxies in +/// creation order without us doing any sorting. +fn galaxy_key(id: GalaxyId) -> [u8; 16] { + *id.0.as_bytes() +} diff --git a/web/index.html b/web/index.html index 24bfa27..3f847e2 100644 --- a/web/index.html +++ b/web/index.html @@ -94,6 +94,127 @@ } #ingest button:hover { background: #232c36; } #ingest button:disabled { opacity: 0.5; cursor: progress; } + #query { + position: fixed; + bottom: 12px; + left: 12px; + width: 380px; + background: rgba(0,0,0,0.55); + padding: 10px; + border-radius: 4px; + font-size: 12px; + } + #query-form { display: flex; gap: 6px; } + #query-input { + flex: 1; + font: inherit; + background: #0c1014; + color: #cfd6df; + border: 1px solid #2a323d; + border-radius: 3px; + padding: 5px 8px; + } + #query-submit { + font: inherit; + background: #1a2129; + color: #cfd6df; + border: 1px solid #2a323d; + padding: 4px 12px; + border-radius: 3px; + cursor: pointer; + } + #query-submit:hover { background: #232c36; } + #query-submit:disabled { opacity: 0.5; cursor: progress; } + #query-status { + margin-top: 6px; + opacity: 0.55; + font-size: 11px; + min-height: 1.2em; + } + #query-answer { + margin-top: 4px; + white-space: pre-wrap; + line-height: 1.45; + max-height: 180px; + overflow-y: auto; + padding: 6px 8px; + background: rgba(255,255,255,0.03); + border-radius: 3px; + min-height: 2em; + } + #query-answer:empty::before { + content: "ask what your knowledge knows…"; + opacity: 0.4; + } + #inspector { + position: fixed; + top: 12px; + left: 50%; + transform: translateX(-50%); + width: 460px; + max-height: 70vh; + overflow-y: auto; + background: rgba(0,0,0,0.78); + padding: 12px 14px; + border-radius: 4px; + border: 1px solid #2a323d; + font-size: 12px; + line-height: 1.45; + display: none; + } + #inspector.open { display: block; } + #inspector h3 { + margin: 0 0 6px; + font-size: 12px; + text-transform: uppercase; + letter-spacing: 0.06em; + opacity: 0.6; + font-weight: 500; + } + #inspector .field { margin-bottom: 8px; } + #inspector .label { + opacity: 0.45; + font-size: 10px; + text-transform: uppercase; + letter-spacing: 0.08em; + margin-bottom: 2px; + } + #inspector .value { color: #e1e6ec; } + #inspector .muted { color: #6b7480; } + #inspector ul { + margin: 2px 0 0 14px; + padding: 0; + } + #inspector ul li { margin-bottom: 1px; } + #inspector .row { + display: flex; + justify-content: space-between; + align-items: center; + } + #inspector .close { + cursor: pointer; + background: transparent; + border: none; + color: #cfd6df; + font-size: 16px; + line-height: 1; + padding: 0 4px; + opacity: 0.7; + } + #inspector .close:hover { opacity: 1; } + #inspector .manifest { + white-space: pre-wrap; + max-height: 8em; + overflow-y: auto; + font-size: 11px; + opacity: 0.85; + } + #inspector .id { + font-family: ui-monospace, monospace; + font-size: 10px; + opacity: 0.5; + word-break: break-all; + } @@ -105,6 +226,9 @@
torus …
engrams 0
synapses 0
+
sim …
+
llm queue …
+
fps …
lm studio …
health …
@@ -119,6 +243,27 @@ +
+
+

engram

+ +
+
+
state
+
manifest
+
taxonomy
+
goals
+
open questions
+
memories
+
+
+
+ + +
+
+
+
diff --git a/web/src/engram_mesh.ts b/web/src/engram_mesh.ts index 1400ac1..12b3996 100644 --- a/web/src/engram_mesh.ts +++ b/web/src/engram_mesh.ts @@ -105,6 +105,18 @@ export class EngramMesh { /** Optional hook fired when an engram's base color is set/updated. The * trail renderer subscribes so head + tail share the same colour. */ public onColorAssigned: ((idx: number, r: number, g: number, b: number) => void) | null = null; + /** Reverse lookup: instance_idx → engram UUID. Filled on upsert; used by + * the click-picker to map a raycast hit back to an engram id. */ + public readonly idxToId: string[] = []; + /** Forward lookup: engram UUID → instance_idx. Mirrors `idxToId` so live + * state-change events (which carry the UUID, not the slot index) can find + * the right vertex to re-paint. */ + private readonly idToIdx = new Map(); + + /** Expose the underlying `THREE.Points` so the scene can raycast against it. */ + pointsObject(): THREE.Points { + return this.points; + } constructor(scene: THREE.Scene, capacity = 5000) { this.capacity = capacity; @@ -125,6 +137,12 @@ export class EngramMesh { geom.setAttribute("aSize", this.sizeAttr); geom.setAttribute("aHash", this.hashAttr); geom.setDrawRange(0, 0); + // Set a permanent oversized bounding sphere. Without this, three.js + // computes one once based on the initial all-zero positions (radius 0) + // and Points.raycast() short-circuits — every click misses. Recomputing + // per frame is expensive; a giant fixed sphere always passes the early + // reject and the per-vertex test then runs normally. + geom.boundingSphere = new THREE.Sphere(new THREE.Vector3(0, 0, 0), 100_000); this.material = new THREE.ShaderMaterial({ vertexShader: VERT_SHADER, @@ -197,6 +215,28 @@ export class EngramMesh { if (idx + 1 > this.maxIdx) this.maxIdx = idx + 1; this.points.geometry.setDrawRange(0, this.maxIdx); + this.idxToId[idx] = e.id; + this.idToIdx.set(e.id, idx); + } + + /** + * Re-paint the dot for an existing engram when its lifecycle state + * changes (Stage 5: responder lights up to Conversing while answering a + * query, Query-Engram transitions Searching→Memorize at completion). + * Silently ignored if the engram isn't known yet — state-change events + * for unfamiliar ids can race ahead of the corresponding `engram_created` + * over the WS bus during a reconnect window. + */ + setStateById(id: string, state: string): void { + const idx = this.idToIdx.get(id); + if (idx === undefined) return; + const color = STATE_COLOR[state] ?? STATE_COLOR.idle; + const colorArr = this.colorAttr.array as Float32Array; + colorArr[idx * 3] = color[0]; + colorArr[idx * 3 + 1] = color[1]; + colorArr[idx * 3 + 2] = color[2]; + this.colorAttr.needsUpdate = true; + this.onColorAssigned?.(idx, color[0], color[1], color[2]); } applyPositionFrame(frame: PositionFrame): void { diff --git a/web/src/inspector.ts b/web/src/inspector.ts new file mode 100644 index 0000000..302e956 --- /dev/null +++ b/web/src/inspector.ts @@ -0,0 +1,179 @@ +import * as THREE from "three"; + +import type { EngramMesh } from "./engram_mesh"; + +/** + * Click-to-inspect: raycast the mouse into the engram point-cloud, fetch + * `/api/galaxy/:gid/engrams/:eid` for the picked engram, and render the + * result in the side panel. Re-clicking an already-open engram refreshes + * the panel (handy because introspection arrives async after birth). + */ + +type EngramDetail = { + id: string; + instance_idx: number; + position: [number, number, number]; + size: number; + state: string; + age: number; + manifest: { kind: string; content?: string } | null; + slate_dim: number | null; + slate_norm: number | null; + introspection: { + taxonomy: string[]; + goals: string[]; + open_questions: string[]; + }; + memories: Array<{ + at_ms: number; + kind: { kind: string; with?: string; weight?: number }; + }>; +}; + +// World-space radius around each engram vertex that counts as a "hit" when +// raycasting. Has to be generous because the rendered dot is enlarged by +// bloom + sprite scaling — the user expects to be able to click anywhere +// the dot *appears*, not the exact 1-unit vertex position. +const PICK_THRESHOLD = 12; +// Pixels of cursor movement between mousedown and mouseup that still count +// as a "click" rather than a drag. Trackpads in particular need slack. +const CLICK_DRAG_TOLERANCE_PX = 8; + +const panel = document.getElementById("inspector"); +const idEl = document.getElementById("inspector-id"); +const stateEl = document.getElementById("inspector-state"); +const manifestEl = document.getElementById("inspector-manifest"); +const taxonomyEl = document.getElementById("inspector-taxonomy"); +const goalsEl = document.getElementById("inspector-goals"); +const questionsEl = document.getElementById("inspector-questions"); +const memoriesEl = document.getElementById("inspector-memories"); +const closeBtn = document.getElementById("inspector-close"); + +closeBtn?.addEventListener("click", () => { + panel?.classList.remove("open"); +}); + +let lastPickedId: string | null = null; + +function shortId(id: string): string { + return id.length > 12 ? `${id.slice(0, 8)}…${id.slice(-4)}` : id; +} + +function bullets(items: string[]): string { + if (!items || items.length === 0) return `…awaiting LLM…`; + return `
    ${items.map((s) => `
  • ${escapeHtml(s)}
    • `).join("")}
`; +} + +function escapeHtml(s: string): string { + return s + .replace(/&/g, "&") + .replace(//g, ">") + .replace(/"/g, """); +} + +function fmtMemories(mems: EngramDetail["memories"]): string { + if (!mems || mems.length === 0) return `none yet`; + const items = mems.map((m) => { + const t = (m.at_ms / 1000).toFixed(1) + "s"; + let body: string; + switch (m.kind.kind) { + case "born": + body = "born"; + break; + case "introspected": + body = "introspected (LLM)"; + break; + case "synapse_formed": + body = `↔ ${shortId(m.kind.with ?? "")} (sim ${(m.kind.weight ?? 0).toFixed(2)})`; + break; + default: + body = m.kind.kind; + } + return `
  • +${t} ${escapeHtml(body)}
    • `; + }); + return `
      ${items.join("")}
    `; +} + +function render(d: EngramDetail): void { + if (!panel) return; + if (idEl) idEl.textContent = d.id; + if (stateEl) stateEl.textContent = `${d.state} · age ${d.age} · slot ${d.instance_idx}`; + if (manifestEl) { + manifestEl.textContent = d.manifest?.content ?? "(synthetic seed — no manifest)"; + } + if (taxonomyEl) taxonomyEl.innerHTML = bullets(d.introspection?.taxonomy ?? []); + if (goalsEl) goalsEl.innerHTML = bullets(d.introspection?.goals ?? []); + if (questionsEl) questionsEl.innerHTML = bullets(d.introspection?.open_questions ?? []); + if (memoriesEl) memoriesEl.innerHTML = fmtMemories(d.memories ?? []); + panel.classList.add("open"); +} + +async function fetchAndRender(galaxyId: string, engramId: string): Promise { + try { + const res = await fetch(`/api/galaxy/${galaxyId}/engrams/${engramId}`); + if (!res.ok) { + console.warn(`engram fetch ${res.status}`); + return; + } + const detail = (await res.json()) as EngramDetail; + render(detail); + } catch (e) { + console.warn("engram fetch failed", e); + } +} + +/** + * Wire up click-picking on the canvas. The engram point cloud is already in + * the scene; we raycast against it and look up the engram id from the hit + * vertex index. + */ +export function attachInspector(opts: { + canvas: HTMLCanvasElement; + camera: THREE.Camera; + engrams: EngramMesh; + galaxyId: string; +}): void { + const raycaster = new THREE.Raycaster(); + raycaster.params.Points = { threshold: PICK_THRESHOLD }; + const mouse = new THREE.Vector2(); + let downX = 0, downY = 0; + + // Use mousedown + mouseup with a tiny movement tolerance so dragging the + // OrbitControls doesn't trigger a pick. + opts.canvas.addEventListener("mousedown", (ev) => { + downX = ev.clientX; + downY = ev.clientY; + }); + opts.canvas.addEventListener("mouseup", (ev) => { + const moved = Math.hypot(ev.clientX - downX, ev.clientY - downY); + if (moved > CLICK_DRAG_TOLERANCE_PX) return; + const rect = opts.canvas.getBoundingClientRect(); + mouse.x = ((ev.clientX - rect.left) / rect.width) * 2 - 1; + mouse.y = -((ev.clientY - rect.top) / rect.height) * 2 + 1; + raycaster.setFromCamera(mouse, opts.camera); + const hits = raycaster.intersectObject(opts.engrams.pointsObject(), false); + console.debug(`[inspector] click ndc=(${mouse.x.toFixed(2)},${mouse.y.toFixed(2)}) hits=${hits.length}`); + if (hits.length === 0) return; + // Pick the closest hit. + const hit = hits[0]; + const idx = hit.index; + if (idx === undefined) return; + const id = opts.engrams.idxToId[idx]; + if (!id) { + console.warn(`[inspector] hit at idx ${idx} has no engram id`); + return; + } + console.debug(`[inspector] picked engram ${id}`); + lastPickedId = id; + void fetchAndRender(opts.galaxyId, id); + }); + + // Light auto-refresh — re-fetch the currently-open engram every 3s so the + // user sees introspection populate and memories grow without re-clicking. + setInterval(() => { + if (!panel?.classList.contains("open")) return; + if (!lastPickedId) return; + void fetchAndRender(opts.galaxyId, lastPickedId); + }, 3000); +} diff --git a/web/src/main.ts b/web/src/main.ts index a4944e9..6c5cb23 100644 --- a/web/src/main.ts +++ b/web/src/main.ts @@ -11,6 +11,9 @@ const ingestBtn = document.getElementById("ingest-btn") as HTMLButtonElement | n const ingestText = document.getElementById("ingest-text") as HTMLTextAreaElement | null; const torusEl = document.getElementById("hud-torus"); const synapsesEl = document.getElementById("hud-synapses"); +const ticksEl = document.getElementById("hud-ticks"); +const llmQueueEl = document.getElementById("hud-llm-queue"); +const fpsEl = document.getElementById("hud-fps"); const resizeBtn = document.getElementById("resize-btn") as HTMLButtonElement | null; const majorInput = document.getElementById("resize-major") as HTMLInputElement | null; const minorInput = document.getElementById("resize-minor") as HTMLInputElement | null; @@ -93,8 +96,36 @@ async function bootstrap(): Promise { if (minorInput && document.activeElement !== minorInput) minorInput.value = String(minorR); }, onSynapses: (n) => { if (synapsesEl) synapsesEl.textContent = String(n); }, + onFps: (fps) => { if (fpsEl) fpsEl.textContent = fps.toFixed(0); }, }); + // Poll /api/stats once a second for sim ticks/sec + LLM queue depth. + // The HUD numbers it drives (sim, llm queue) are server-side state, so + // the browser can't compute them locally — keep this loop lightweight + // and silent on transient fetch errors so a brief network hiccup doesn't + // pollute the console. + type StatsResponse = { + ticks_per_sec: number; + engrams_total: number; + synapses_total: number; + galaxies: number; + llm_queue_depth: number; + llm_parallel_cap: number; + }; + setInterval(async () => { + try { + const r = await fetch("/api/stats"); + if (!r.ok) return; + const s = (await r.json()) as StatsResponse; + if (ticksEl) ticksEl.textContent = `${s.ticks_per_sec}/s`; + if (llmQueueEl) { + llmQueueEl.textContent = `${s.llm_queue_depth} / ${s.llm_parallel_cap}`; + } + } catch { + // ignore — server might be restarting + } + }, 1000); + if (seedBtn && seedNInput) { seedBtn.addEventListener("click", async () => { const n = Math.max(1, Math.min(5000, parseInt(seedNInput.value, 10) || 200)); diff --git a/web/src/query_panel.ts b/web/src/query_panel.ts new file mode 100644 index 0000000..ffc9a9b --- /dev/null +++ b/web/src/query_panel.ts @@ -0,0 +1,128 @@ +/** + * Stage 5 query panel. + * + * Owns the bottom-left text input + answer area + SSE consumer. Submission + * does `POST /api/galaxy/:id/query` and immediately opens an EventSource on + * `/api/query/:id/stream`; the running answer streams in as the integrator + * re-runs. Closes the EventSource on terminal status (`done` / `failed`). + * + * The visual feedback in the 3D scene (responder light-up, Query-Engram + * appearing at the donut center) lives in `scene.ts` — driven by separate + * WS events. This module only manages the side panel. + */ + +import type { QueryStatus } from "./ws_client"; + +type QueryUpdate = { + query_id: string; + version: number; + status: QueryStatus; + answer: string; + responder_count: number; +}; + +export type QueryPanelOptions = { + galaxyId: string; +}; + +const TERMINAL: QueryStatus[] = ["done", "failed"]; + +export function attachQueryPanel(opts: QueryPanelOptions): void { + const form = document.getElementById("query-form") as HTMLFormElement | null; + const input = document.getElementById("query-input") as HTMLInputElement | null; + const submit = document.getElementById("query-submit") as HTMLButtonElement | null; + const answerEl = document.getElementById("query-answer"); + const statusEl = document.getElementById("query-status"); + if (!form || !input || !submit || !answerEl || !statusEl) { + console.warn("query panel markup missing — skipping wireup"); + return; + } + + let activeStream: EventSource | null = null; + + function setBusy(busy: boolean): void { + submit!.disabled = busy; + input!.disabled = busy; + } + + function renderUpdate(u: QueryUpdate): void { + answerEl!.textContent = u.answer || "(thinking…)"; + const phase = + u.status === "pending" + ? "broadcasting" + : u.status === "responding" + ? `synthesising · ${u.responder_count} perspective${u.responder_count === 1 ? "" : "s"}` + : u.status === "integrating" + ? "integrating" + : u.status === "done" + ? `done · ${u.responder_count} perspective${u.responder_count === 1 ? "" : "s"}` + : "failed"; + statusEl!.textContent = `v${u.version} · ${phase}`; + } + + function closeStream(): void { + if (activeStream) { + activeStream.close(); + activeStream = null; + } + } + + async function submitQuery(text: string): Promise { + setBusy(true); + closeStream(); + answerEl!.textContent = "(asking…)"; + statusEl!.textContent = "submitting"; + + let queryId: string; + try { + const res = await fetch(`/api/galaxy/${opts.galaxyId}/query`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ text }), + }); + if (!res.ok) { + const body = await res.text(); + throw new Error(`HTTP ${res.status}: ${body}`); + } + const body = (await res.json()) as { query_id: string }; + queryId = body.query_id; + } catch (e) { + statusEl!.textContent = "failed"; + answerEl!.textContent = `submit failed: ${(e as Error).message}`; + setBusy(false); + return; + } + + activeStream = new EventSource(`/api/query/${queryId}/stream`); + activeStream.onmessage = (ev) => { + try { + const update = JSON.parse(ev.data) as QueryUpdate; + renderUpdate(update); + if (TERMINAL.includes(update.status)) { + closeStream(); + setBusy(false); + } + } catch (err) { + console.warn("query SSE decode failed", err); + } + }; + activeStream.onerror = () => { + // EventSource auto-reconnects on transient errors; but once the + // server has finished and closed the channel, the reconnect loop + // would just pile up 404s. The terminal-status check above already + // closed the stream in the success path; this handler covers the + // failure path where the channel never opened cleanly. + if (activeStream && activeStream.readyState === EventSource.CLOSED) { + statusEl!.textContent = "stream closed"; + setBusy(false); + } + }; + } + + form.addEventListener("submit", (ev) => { + ev.preventDefault(); + const text = input!.value.trim(); + if (!text) return; + void submitQuery(text); + }); +} diff --git a/web/src/scene.ts b/web/src/scene.ts index 4f6c6cf..c4f70f8 100644 --- a/web/src/scene.ts +++ b/web/src/scene.ts @@ -8,6 +8,8 @@ import { OutputPass } from "three/examples/jsm/postprocessing/OutputPass.js"; import { EngramMesh } from "./engram_mesh"; import { EngramTrails } from "./engram_trails"; import { SynapseMesh } from "./synapse_mesh"; +import { attachInspector } from "./inspector"; +import { attachQueryPanel } from "./query_panel"; import type { EngramSnapshot, EventHandlers, SynapseDto } from "./ws_client"; import { connectGalaxy } from "./ws_client"; @@ -16,6 +18,8 @@ export type SceneHooks = { onGalaxyName: (name: string) => void; onTorus: (majorRadius: number, minorRadius: number) => void; onSynapses?: (count: number) => void; + /** Rolling FPS, updated about once per second from inside the rAF loop. */ + onFps?: (fps: number) => void; }; // State-color table mirrored from EngramMesh; needed here so we can pre-paint @@ -174,6 +178,11 @@ export function startScene(canvas: HTMLCanvasElement, galaxyId: string, hooks: S synapses.applyPositionFrame(frame); hooks.onCount(engrams.count()); }, + onEngramStateChanged: (id, state) => { + // Re-paint the dot for an engram whose state flipped (Stage 5 + // responder light-up + Query-Engram → Memorize on completion). + engrams.setStateById(id, state); + }, onClose: () => { console.warn("ws closed — refresh to reconnect"); }, @@ -181,6 +190,9 @@ export function startScene(canvas: HTMLCanvasElement, galaxyId: string, hooks: S }; connectGalaxy(galaxyId, handlers); + attachInspector({ canvas, camera, engrams, galaxyId }); + attachQueryPanel({ galaxyId }); + function onResize(): void { const w = window.innerWidth; const h = window.innerHeight; @@ -193,10 +205,22 @@ export function startScene(canvas: HTMLCanvasElement, galaxyId: string, hooks: S window.addEventListener("resize", onResize); const startedAt = performance.now(); + // FPS: rolling counter — accumulate frames, emit roughly once a second. + let fpsWindowStart = startedAt; + let fpsFrames = 0; function animate(): void { controls.update(); - engrams.tick((performance.now() - startedAt) / 1000); + const nowMs = performance.now(); + engrams.tick((nowMs - startedAt) / 1000); composer.render(); + fpsFrames += 1; + const elapsed = nowMs - fpsWindowStart; + if (elapsed >= 1000) { + const fps = (fpsFrames * 1000) / elapsed; + hooks.onFps?.(fps); + fpsFrames = 0; + fpsWindowStart = nowMs; + } requestAnimationFrame(animate); } animate(); diff --git a/web/src/ws_client.ts b/web/src/ws_client.ts index d0da073..29333b0 100644 --- a/web/src/ws_client.ts +++ b/web/src/ws_client.ts @@ -33,6 +33,13 @@ export type SynapseDto = { weight: number; }; +export type QueryStatus = + | "pending" + | "responding" + | "integrating" + | "done" + | "failed"; + export type SimEventMsg = | { type: "hello"; @@ -47,6 +54,27 @@ export type SimEventMsg = center: [number, number, number]; major_radius: number; minor_radius: number; + } + | { type: "engram_state_changed"; id: string; state: string } + | { + type: "query_started"; + query: string; + engram: string; + position: [number, number, number]; + text: string; + } + | { + type: "query_answer_updated"; + query: string; + version: number; + status: QueryStatus; + answer: string; + } + | { + type: "query_finished"; + query: string; + status: QueryStatus; + responder_count: number; }; export type PositionFrame = { @@ -69,6 +97,20 @@ export type EventHandlers = { majorRadius: number, minorRadius: number, ) => void; + onEngramStateChanged?: (id: string, state: string) => void; + onQueryStarted?: ( + queryId: string, + engramId: string, + position: [number, number, number], + text: string, + ) => void; + onQueryAnswerUpdated?: ( + queryId: string, + version: number, + status: QueryStatus, + answer: string, + ) => void; + onQueryFinished?: (queryId: string, status: QueryStatus, responderCount: number) => void; onPositionFrame?: (frame: PositionFrame) => void; onClose?: (ev: CloseEvent) => void; onError?: (ev: Event) => void; @@ -97,6 +139,18 @@ export function connectGalaxy(galaxyId: string, handlers: EventHandlers): WebSoc case "torus_updated": handlers.onTorusUpdated?.(msg.center, msg.major_radius, msg.minor_radius); break; + case "engram_state_changed": + handlers.onEngramStateChanged?.(msg.id, msg.state); + break; + case "query_started": + handlers.onQueryStarted?.(msg.query, msg.engram, msg.position, msg.text); + break; + case "query_answer_updated": + handlers.onQueryAnswerUpdated?.(msg.query, msg.version, msg.status, msg.answer); + break; + case "query_finished": + handlers.onQueryFinished?.(msg.query, msg.status, msg.responder_count); + break; } } else { const frame = decodePositionFrame(ev.data as ArrayBuffer);