Implement Sophia MVP stages 4-7 (introspection, queries, persistence, polish)
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) <noreply@anthropic.com>
This commit is contained in:
@@ -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 }
|
||||
|
||||
@@ -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<QueryRegistry>,
|
||||
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<Arc<AppState>>) -> 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<Arc<AppState>>) -> 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<CreateGalaxyBody>,
|
||||
) -> 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<Arc<AppState>>) -> Response {
|
||||
}
|
||||
}
|
||||
|
||||
async fn delete_galaxy_unimpl(Path(_id): Path<GalaxyId>) -> 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<Arc<AppState>>,
|
||||
Path(id): Path<GalaxyId>,
|
||||
) -> 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<String> = texts.clone();
|
||||
let items: Vec<IngestItem> = 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<Arc<AppState>>,
|
||||
Path(id): Path<GalaxyId>,
|
||||
Json(body): Json<QueryBody>,
|
||||
) -> 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<Arc<AppState>>,
|
||||
Path(id): Path<QueryId>,
|
||||
) -> 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<Arc<AppState>>,
|
||||
Path(id): Path<QueryId>,
|
||||
) -> 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<Box<dyn Stream<Item = Result<SseEvent, Infallible>> + 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 {
|
||||
|
||||
393
crates/sophia-server/src/query.rs
Normal file
393
crates/sophia-server/src/query.rs
Normal file
@@ -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<QueryUpdate>,
|
||||
latest: Mutex<QueryUpdate>,
|
||||
}
|
||||
|
||||
/// Process-wide registry. Cheaply cloneable Arc holds an `RwLock` over the
|
||||
/// `HashMap<QueryId, Arc<QueryEntry>>` — typical access is read-heavy
|
||||
/// (subscribers + GET /api/query/:id), with a single write per new query.
|
||||
#[derive(Default)]
|
||||
pub struct QueryRegistry {
|
||||
inner: RwLock<HashMap<QueryId, Arc<QueryEntry>>>,
|
||||
}
|
||||
|
||||
impl QueryRegistry {
|
||||
pub fn new() -> Self {
|
||||
Self::default()
|
||||
}
|
||||
|
||||
async fn register(&self, query_id: QueryId, initial: QueryUpdate) -> Arc<QueryEntry> {
|
||||
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<Arc<QueryEntry>> {
|
||||
self.inner.read().await.get(&query_id).cloned()
|
||||
}
|
||||
}
|
||||
|
||||
impl QueryEntry {
|
||||
pub fn subscribe(&self) -> broadcast::Receiver<QueryUpdate> {
|
||||
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<QueryRegistry>,
|
||||
galaxy: GalaxyId,
|
||||
text: String,
|
||||
) -> Result<QueryId, String> {
|
||||
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<QueryEntry>,
|
||||
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<Mutex<Vec<String>>> = 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() {}
|
||||
}
|
||||
Reference in New Issue
Block a user