diff --git a/Cargo.lock b/Cargo.lock index 3538ff44..369617f3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5974,7 +5974,7 @@ dependencies = [ [[package]] name = "pumas-library" -version = "0.7.0" +version = "0.8.0-rc.1" dependencies = [ "anyhow", "async-trait", diff --git a/crates/server/src/README.md b/crates/server/src/README.md index 45adbd41..c6c2d7a5 100644 --- a/crates/server/src/README.md +++ b/crates/server/src/README.md @@ -63,8 +63,11 @@ domain model in `eidetic-core`. | `agent_workflow_harness_tests.rs` | Provider, manifest, executor, cancellation and exact-budget lifecycle regressions against SQLite history. | | `agent_premise_workflow.rs` | First premise graph-context workflow slice over backend graph reads, reviewable proposals, and harness history. | | `export_service.rs` | Host-neutral PDF export behavior consumed by Tauri commands. | -| `reference_service.rs` | Canonical reference list/upload/delete and source-bound asynchronous embedding publication. | -| `embeddings.rs` | Transitional HTTP embedding adapter validating returned model, input index and finite nonzero vectors. | +| `reference_service.rs` | Canonical reference list/upload/delete, explicit index status/reindex and all-or-nothing source-bound embedding publication; reopen rebuilds disposable index state. | +| `embeddings.rs` | Independently configured real Pumas typed embeddings and authenticated OpenAI-compatible embedding adapter, binding provider/model/profile/load revision and finite nonzero vectors. | +| `pumas_inference.rs` | Fenced Pumas discovery, typed operation v1/capability reads, model metadata revision and explicit serving lifecycle RPC consumer. | +| `pumas_inference_tests.rs` | Controlled HTTP fixtures proving load/use/unload, reference upload/reopen/model-revision refusal, visible errors and retrieved prompt contributions; no real-model inference claim. | +| `pumas_runtime.rs` | Optional selected-root Pumas control-plane bootstrap and generation-fenced owned shutdown. Explicit drain retains child/output completion; cancelled construction and last-Drop observation remain gated by the lifecycle report. Borrowed services are never stopped. | | `vector_store.rs` | Disposable exact-source index with publication tickets, representation-space filtering and stable ranking. | | `vector_store_tests.rs` | Source custody, stale lifecycle, incompatible representation and numeric ranking regressions. | | `reference_retrieval_tests.rs` | Actual RAG attach-boundary regression coverage for queued/reopened sessions and late query results. | @@ -305,10 +308,23 @@ Saved-project listings read modification metadata from project.db, so editing an queued retrieval. Create/load publish path, project and index under the project guard. Source checks happen again before ranking under project-then-index lock order. No lock crosses model I/O. + Retrieval verifies the captured index scope under that same guard before any + compatibility/status mutation; a late query cannot mark a newer reindexed or + same-path reopened document stale. Regression checks assert the complete fresh + status remains unchanged and its current query still retrieves the source. - Only nonempty finite nonzero vectors with matching configured endpoint/model and dimension are ranked. Returned HTTP model identity must match the request. - This does not attest immutable model weights; Pumas revision-aware embeddings, - live runtime qualification and saved-reference re-indexing remain M3 work. + Pumas embeddings now bind producer instance/service generation, provider, + selected profile/model, monotonic serving cursor, optional load timestamp and actual library metadata, including + upstream revision when present. Dimension mismatch excludes candidates. + The pinned producer omits load timestamps; its service-scoped cursor fences + same-model reloads with unchanged metadata. Any serving mutation, including an + unrelated chat load, conservatively requires reindexing. Before/after inference + observations reject an operation that crosses a serving-state change. + This is session/provenance identity, not an immutable weight attestation; + out-of-band file mutation without metadata/reload is outside the producer's + current contract. Saved sources rebuild on reopen; failed/stale/indexing states + are projected by `reference_index_status`. See `docs/pumas-local-inference.md`. - Generation and preview accept an optional explicit fictional `story_time_ms`. Snapshot fields resolve independently at or before it. Conflicting values at the latest effective timestamp stop context construction. A cleared value is @@ -699,6 +715,40 @@ recap titles actually supplied before provider I/O. Existing graph edges and proposal bindings support exact impact, stale/ABA refusal, historical/current or owned-deletion evidence, and explicit acceptance with refreshed consumed lineage. +Local inference uses the existing configuration facade plus `ai_config_get`, +`pumas_catalog`, `pumas_load_model`, `pumas_unload_model`, +`reference_index_status` and `reference_reindex` desktop commands. Chat Pumas +selection requires explicit installed model/profile; embedding selection has +separate provider, endpoint, model/profile and credentials. Typed stream terminal +failures never publish partial screenplay. Reference retrieval emits source names +or a visible error before the actual generation context event. Disabled or failed +retrieval permits ordinary generation with the error displayed; it does not claim +references contributed. Configuration is application-session state, as before; +references are canonical saved SQLite project content. + +The Linux cancelled-startup fixture observes actual `/proc` child disappearance +after the owned wait receipt before deleting its signal files or ending the test +runtime; a recorded shutdown RPC alone is not process-cessation evidence. + +Explicit Pumas shutdown retains the child and stdout-drain completion together, +including I/O errors and panics. Cancelling a shutdown waiter leaves that output +handle available to the next waiter. A hung borrowed bootstrap is terminated +and reaped as our helper only, without sending shutdown/unload to the borrowed +service. Full cancelled-construction and last-Drop completion ownership still +requires a retained composition owner; see the +[lifecycle repair and remaining decision](../../../docs/reports/pr29-lifecycle-ownership.md). + +Explicit selected-root bootstrap failures preserve Pumas as the selected backend +and expose the actual startup error in AI status. They never silently revert to +a direct llama.cpp endpoint; an operator can supply a valid explicit Pumas URL +to retry connection through the existing configuration control. + +Unload decodes Pumas's actual `UnserveModelResponse`: an `error` is failure even +when `success` is true; `unloaded: false` without an error is an already-unserved +outcome rather than a successful unload receipt. Missing required receipt fields +are rejected. HTTP fixtures retain the loaded model after ONNX/Ollama-shaped +unload failures and verify the actual provider diagnostic reaches the caller. + ## Canonical arc assignment and sibling reorder `timeline_arc_membership` validates assigned IDs and owned field reads in the diff --git a/crates/server/src/ai_backends/README.md b/crates/server/src/ai_backends/README.md index 842f4b6e..689943cd 100644 --- a/crates/server/src/ai_backends/README.md +++ b/crates/server/src/ai_backends/README.md @@ -7,7 +7,8 @@ This directory implements server-side adapters that translate core AI requests i | File/Folder | Description | |-------------|-------------| | `mod.rs` | Shared backend trait and adapter selection. | -| `llamacpp.rs` | Local Pumas llama.cpp OpenAI-compatible adapter. | +| `llamacpp.rs` | Existing direct local llama.cpp OpenAI-compatible adapter. | +| `pumas.rs` | Managed Pumas typed chat operation v1 with capability checks, fenced transport and JSON-output validation. | | `openrouter.rs` | OpenRouter HTTP adapter. | | `sse.rs` | Shared stateful SSE framing, completion validation and transport error propagation. | | `sse_tests.rs` | Buffer boundaries, fragmented framing, malformed input and bounded provider diagnostics. | @@ -66,3 +67,11 @@ use crate::ai_backends::AiBackend; ## Structured Producer Contract - Backend adapters must preserve the field semantics defined by core AI request/response types. - Changes to response interpretation require coordinated route and frontend progress-handling updates. + +Pumas typed SSE uses `started`, `delta`, `completed` and `failed` JSON events, +checks contract version and request identity, and requires a valid terminal event +plus clean HTTP EOF. The existing OpenAI-compatible providers continue to require +`[DONE]`. Typed Pumas v1 has no JSON-mode option: structured calls explicitly +request JSON in their prompt and validate JSON before the established parsers. +This follows the producer contract at `ad31e391dcd94c204b3fcd748cc98d27b3281e65`; +no direct ONNX session or fake backend is selected in production. diff --git a/crates/server/src/ai_backends/mod.rs b/crates/server/src/ai_backends/mod.rs index cab9a475..ac88fda1 100644 --- a/crates/server/src/ai_backends/mod.rs +++ b/crates/server/src/ai_backends/mod.rs @@ -1,5 +1,6 @@ pub(crate) mod llamacpp; pub(crate) mod openrouter; +mod pumas; mod sse; use serde::Serialize; @@ -13,11 +14,13 @@ use eidetic_core::error::Error; pub(crate) enum Backend { LlamaCpp(llamacpp::LlamaCppBackend), OpenRouter(openrouter::OpenRouterBackend), + Pumas(pumas::PumasBackend), } impl Backend { pub fn from_config(config: &AiConfig) -> Self { match config.backend_type { + BackendType::Pumas => Backend::Pumas(pumas::PumasBackend::new(config)), BackendType::LlamaCpp => Backend::LlamaCpp(llamacpp::LlamaCppBackend::new(config)), BackendType::OpenRouter => { Backend::OpenRouter(openrouter::OpenRouterBackend::new(config)) @@ -33,6 +36,7 @@ impl Backend { match self { Backend::LlamaCpp(b) => b.generate(prompt, config).await, Backend::OpenRouter(b) => b.generate(prompt, config).await, + Backend::Pumas(b) => b.generate(prompt, config).await, } } @@ -61,6 +65,7 @@ impl Backend { match self { Backend::LlamaCpp(b) => b.generate_json(prompt, config).await, Backend::OpenRouter(b) => b.generate_json(prompt, config).await, + Backend::Pumas(b) => b.generate_json(prompt, config).await, } } @@ -68,6 +73,7 @@ impl Backend { match self { Backend::LlamaCpp(b) => b.health_check().await, Backend::OpenRouter(b) => b.health_check().await, + Backend::Pumas(b) => b.health_check().await, } } } diff --git a/crates/server/src/ai_backends/pumas.rs b/crates/server/src/ai_backends/pumas.rs new file mode 100644 index 00000000..6663cfde --- /dev/null +++ b/crates/server/src/ai_backends/pumas.rs @@ -0,0 +1,90 @@ +use eidetic_core::{Error, ai::backend::GenerateStream}; +use serde_json::json; + +use super::BackendStatus; +use crate::{ + prompt_format::ChatPrompt, + pumas_inference::{PumasClient, PumasOperation}, + state::{AiConfig, BackendType}, +}; + +pub(crate) struct PumasBackend { + config: Box, +} +impl PumasBackend { + pub(crate) fn new(config: &AiConfig) -> Self { + Self { + config: Box::new(config.clone()), + } + } + pub(crate) async fn generate( + &self, + prompt: &ChatPrompt, + config: &AiConfig, + ) -> Result { + let client = PumasClient::connect(&config.base_url) + .await + .map_err(Error::AiBackend)?; + let (id, response) = client.operation(PumasOperation { + model: &config.model, + profile: &config.pumas_profile, + capability: "chat_generation", + input: json!({"kind":"messages","messages":[{"role":"system","content":prompt.system},{"role":"user","content":prompt.user}]}), + output: "text", + options: json!({"kind":"text_generation","max_tokens":config.max_tokens,"temperature":config.temperature}), + stream: true, + }).await.map_err(Error::AiBackend)?; + Ok(super::sse::pumas_tokens( + response, + id, + config.model.clone(), + config.pumas_profile.clone(), + )) + } + pub(crate) async fn generate_json( + &self, + prompt: &ChatPrompt, + config: &AiConfig, + ) -> Result { + // Typed v1 has no JSON-mode option. Require JSON in the prompt and validate + // it here before the existing structured-output parsers see it. + use futures::StreamExt; + let prompt = ChatPrompt { + system: format!( + "{}\nReturn only valid JSON without Markdown fences.", + prompt.system + ), + user: prompt.user.clone(), + }; + let mut stream = self.generate(&prompt, config).await?; + let mut full = String::new(); + while let Some(token) = stream.next().await { + full.push_str(&token?); + } + serde_json::from_str::(&full) + .map_err(|e| Error::AiBackend(format!("Pumas returned invalid JSON: {e}")))?; + Ok(full) + } + pub(crate) async fn health_check(&self) -> Result { + let client = PumasClient::connect(&self.config.base_url) + .await + .map_err(Error::AiBackend)?; + client + .selected( + &self.config.model, + &self.config.pumas_profile, + "chat_generation", + ) + .await + .map_err(Error::AiBackend)?; + Ok(BackendStatus { + connected: true, + model: self.config.model.clone(), + backend_type: BackendType::Pumas, + message: format!( + "Pumas model {} loaded on {}", + self.config.model, self.config.pumas_profile + ), + }) + } +} diff --git a/crates/server/src/ai_backends/sse.rs b/crates/server/src/ai_backends/sse.rs index 3ff362da..6b21709e 100644 --- a/crates/server/src/ai_backends/sse.rs +++ b/crates/server/src/ai_backends/sse.rs @@ -20,6 +20,8 @@ struct Reader { data: Vec, completed: bool, provider: &'static str, + pumas_request: Option<(String, String, String)>, + started: bool, } impl Reader { @@ -43,12 +45,47 @@ impl Reader { if self.completed { return Err(self.error("received data after completion")); } - if text.trim() == "[DONE]" { + if self.pumas_request.is_none() && text.trim() == "[DONE]" { self.completed = true; return Ok(None); } let value: serde_json::Value = serde_json::from_str(text).map_err(|error| self.error(error))?; + if let Some((id, model, profile)) = &self.pumas_request { + if value["request_id"] != *id { + return Err(self.error("Pumas stream correlation mismatch")); + } + match value["kind"].as_str() { + Some("started") + if !self.started + && value["contract_version"] == 1 + && value["capability"] == "chat_generation" + && value["model"] == *model + && value["profile"] == *profile => + { + self.started = true; + return Ok(None); + } + Some("delta") if self.started => { + return value["text"] + .as_str() + .map(|text| Some(text.to_owned())) + .ok_or_else(|| self.error("invalid Pumas text delta")); + } + Some("completed") + if self.started + && matches!( + value["finish_reason"].as_str(), + Some("stop" | "length" | "content_filter") + ) => + { + self.completed = true; + return Ok(None); + } + Some("failed") => return Err(self.error(&value["error"])), + _ => return Err(self.error("invalid Pumas stream sequence")), + } + } if let Some(error) = value.get("error") { return Err(self.error(error)); } @@ -126,6 +163,23 @@ impl std::fmt::Write for ErrorDetail { } pub(super) fn tokens(response: reqwest::Response, provider: &'static str) -> GenerateStream { + stream_tokens(response, provider, None) +} + +pub(super) fn pumas_tokens( + response: reqwest::Response, + request_id: String, + model: String, + profile: String, +) -> GenerateStream { + stream_tokens(response, "Pumas", Some((request_id, model, profile))) +} + +fn stream_tokens( + response: reqwest::Response, + provider: &'static str, + pumas_request: Option<(String, String, String)>, +) -> GenerateStream { let reader = Reader { body: response .bytes_stream() @@ -137,6 +191,8 @@ pub(super) fn tokens(response: reqwest::Response, provider: &'static str) -> Gen data: Vec::new(), completed: false, provider, + pumas_request, + started: false, }; Box::pin(stream::try_unfold(reader, |mut reader| async move { loop { diff --git a/crates/server/src/ai_backends/sse_tests.rs b/crates/server/src/ai_backends/sse_tests.rs index e0bdf397..129bcaa0 100644 --- a/crates/server/src/ai_backends/sse_tests.rs +++ b/crates/server/src/ai_backends/sse_tests.rs @@ -9,6 +9,8 @@ fn reader() -> Reader { data: Vec::new(), completed: false, provider: "fixture", + pumas_request: None, + started: false, } } @@ -146,3 +148,83 @@ fn a_chunk_is_consumed_incrementally_without_a_token_queue() { assert!(reader.scan > consumed); assert_eq!(reader.lines().unwrap(), None); } + +#[test] +fn pumas_typed_stream_checks_order_contract_selection_and_correlation() { + let typed = || { + let mut reader = reader(); + reader.pumas_request = Some(("request".into(), "model".into(), "profile".into())); + reader + }; + let started = br#"data: {"kind":"started","contract_version":1,"request_id":"request","capability":"chat_generation","model":"model","profile":"profile"} + +"#; + let delta = br#"data: {"kind":"delta","request_id":"request","text":"hello"} + +"#; + let completed = br#"data: {"kind":"completed","request_id":"request","finish_reason":"stop"} + +"#; + let mut reader = typed(); + assert!(feed(&mut reader, started).unwrap().is_empty()); + assert_eq!(feed(&mut reader, delta).unwrap(), ["hello"]); + assert!(feed(&mut reader, completed).unwrap().is_empty()); + assert!(reader.completed); + assert!(feed(&mut reader, delta).is_err()); + assert!(feed(&mut typed(), delta).is_err()); + assert!(feed(&mut typed(), completed).is_err()); + for replacement in [ + String::from_utf8_lossy(started) + .replace("\"contract_version\":1", "\"contract_version\":2"), + String::from_utf8_lossy(started).replace("\"model\":\"model\"", "\"model\":\"other\""), + String::from_utf8_lossy(started) + .replace("\"profile\":\"profile\"", "\"profile\":\"other\""), + String::from_utf8_lossy(started) + .replace("\"request_id\":\"request\"", "\"request_id\":\"other\""), + ] { + assert!(feed(&mut typed(), replacement.as_bytes()).is_err()); + } + let mut reader = typed(); + feed(&mut reader, started).unwrap(); + assert!(feed(&mut reader, br#"data: {"kind":"failed","request_id":"request","error":{"code":"transport_lost","outcome":"unknown"}} + +"#).unwrap_err().to_string().contains("transport_lost")); +} + +#[tokio::test] +async fn dropping_pumas_stream_closes_pending_transport_without_fabricating_completion() { + use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::TcpListener, + }; + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let peer = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.unwrap(); + let mut request = vec![0; 4096]; + assert!(socket.read(&mut request).await.unwrap() > 0); + let body = concat!( + "data: {\"kind\":\"started\",\"contract_version\":1,\"request_id\":\"request\",\"capability\":\"chat_generation\",\"model\":\"model\",\"profile\":\"profile\"}\n\n", + "data: {\"kind\":\"delta\",\"request_id\":\"request\",\"text\":\"partial\"}\n\n" + ); + // A declared unfinished body keeps the producer connection open. + socket.write_all(format!("HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\n\r\n{body}",body.len()+100).as_bytes()).await.unwrap(); + let mut byte = [0]; + tokio::time::timeout(std::time::Duration::from_secs(5), socket.read(&mut byte)) + .await + .unwrap() + .unwrap() + }); + let response = reqwest::Client::builder() + .no_proxy() + .build() + .unwrap() + .get(url) + .send() + .await + .unwrap(); + let mut stream = pumas_tokens(response, "request".into(), "model".into(), "profile".into()); + assert_eq!(stream.next().await.unwrap().unwrap(), "partial"); + drop(stream); + assert_eq!(peer.await.unwrap(), 0); +} diff --git a/crates/server/src/ai_generation_runtime.rs b/crates/server/src/ai_generation_runtime.rs index 31157431..a0b98c3b 100644 --- a/crates/server/src/ai_generation_runtime.rs +++ b/crates/server/src/ai_generation_runtime.rs @@ -151,7 +151,7 @@ async fn finish_generation_stream( persist_successful_generation(state, project_path, node_uuid, full_text, inputs).await; } -async fn attach_rag_context( +pub(crate) async fn attach_rag_context( state: &AppState, config: &crate::state::AiConfig, project_path: &std::path::Path, @@ -159,20 +159,38 @@ async fn attach_rag_context( request: &mut GenerateRequest, ) { request.rag_context.clear(); - { - let store = state.vector_store.lock(); - if scope != store.scope() || store.is_empty() { - return; - } + if scope != state.vector_store.lock().scope() { + return; } - let embed_client = - EmbeddingClient::new(&config.base_url, crate::state::constants::EMBEDDING_MODEL); - match embed_client.embed(&request.target_node.content.notes).await { - Ok(query_embedding) => { - attach_rag_embedding(state, project_path, scope, &query_embedding, request) - } - Err(_) => tracing::warn!("Reference retrieval unavailable: query embedding failed"), + let has_references = state + .project + .lock() + .as_ref() + .is_some_and(|project| !project.references.is_empty()); + if !has_references { + return; } + let result = async { + if state.vector_store.lock().is_empty() {return Err("References are not indexed; check reference index status and Reindex".to_owned());} + let embed_client = EmbeddingClient::from_config(&config.embedding).await?; + let query_embedding = embed_client.embed(&request.target_node.content.notes).await?; + attach_rag_embedding(state, project_path, scope, &query_embedding, request); + if request.rag_context.is_empty() {return Err("Reference index is stale or incompatible with the current embedding model revision/dimensions; Reindex".to_owned());} + Ok(()) + }.await; + let error = result.err(); + if let Some(error) = &error { + tracing::warn!(%error, "Reference retrieval unavailable"); + } + let _ = state.events_tx.send(ServerEvent::ReferenceRetrieval { + node_id: request.target_node.id.0, + sources: request + .rag_context + .iter() + .map(|chunk| chunk.source.clone()) + .collect(), + error, + }); } /// Final publication boundary after external model I/O. The request carries the @@ -186,9 +204,7 @@ fn attach_rag_embedding( ) { request.rag_context.clear(); let config = state.ai_config.lock().clone(); - let current_client = - EmbeddingClient::new(&config.base_url, crate::state::constants::EMBEDDING_MODEL); - if query_embedding.identity != current_client.identity() { + if !query_embedding.matches_config(&config.embedding) { return; } let guard = state.project.lock(); @@ -198,7 +214,11 @@ fn attach_rag_embedding( if state.project_database.active_path().as_deref() != Some(project_path) { return; } - let store = state.vector_store.lock(); + let mut store = state.vector_store.lock(); + if scope != store.scope() { + return; + } + store.mark_incompatible(query_embedding); request.rag_context = store .search( scope, diff --git a/crates/server/src/ai_service.rs b/crates/server/src/ai_service.rs index 2ed4d49c..da4f0e5e 100644 --- a/crates/server/src/ai_service.rs +++ b/crates/server/src/ai_service.rs @@ -37,6 +37,8 @@ pub struct AiConfigUpdate { pub max_tokens: Option, pub base_url: Option, pub api_key: Option>, + pub pumas_profile: Option, + pub embedding: Option, } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] @@ -54,6 +56,18 @@ pub struct AiGenerateChildrenRequest { pub async fn get_ai_status(state: &AppState) -> AiStatus { let config = state.ai_config.lock().clone(); + if config.backend_type == BackendType::Pumas + && config.base_url.is_empty() + && let Some(error) = &state.pumas_startup_error + { + return AiStatus { + backend: BackendType::Pumas, + model: config.model, + connected: false, + message: None, + error: Some(format!("Pumas startup failed: {error}")), + }; + } let backend = Backend::from_config(&config); match backend.health_check().await { @@ -120,7 +134,22 @@ pub fn update_ai_config(state: &AppState, update: AiConfigUpdate) -> AiConfig { if let Some(api_key) = update.api_key { config.api_key = api_key.filter(|value| !value.is_empty()); } - config.clone() + if let Some(profile) = update.pumas_profile { + config.pumas_profile = profile; + } + let reindex = update + .embedding + .as_ref() + .is_some_and(|embedding| *embedding != config.embedding); + if let Some(embedding) = update.embedding { + config.embedding = embedding; + } + let result = config.clone(); + drop(config); + if reindex { + crate::reference_service::schedule_reindex(state); + } + result } pub(crate) async fn active_sqlite_project( @@ -355,6 +384,8 @@ mod tests { max_tokens: Some(1024), base_url: Some("https://example.test/v1".to_string()), api_key: Some(Some(String::new())), + pumas_profile: None, + embedding: None, }, ); diff --git a/crates/server/src/embeddings.rs b/crates/server/src/embeddings.rs index 11e41148..ba69db82 100644 --- a/crates/server/src/embeddings.rs +++ b/crates/server/src/embeddings.rs @@ -1,12 +1,38 @@ use reqwest::Client; use serde::{Deserialize, Serialize}; -/// Configured representation space. This does not attest immutable model weights; -/// Pumas package/revision identity must replace it before that stronger guarantee. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)] +#[serde(rename_all = "snake_case")] +pub enum EmbeddingProvider { + #[default] + Disabled, + Pumas, + OpenAiCompatible, +} + +/// Embeddings have their own endpoint, model, profile, revision and credentials. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)] +#[serde(default)] +pub struct EmbeddingConfig { + pub provider: EmbeddingProvider, + pub base_url: String, + pub model: String, + pub profile: String, + /// Required operator revision for a non-Pumas endpoint. Pumas resolves its + /// current model provenance/load revision directly from the producer. + pub revision: String, + pub api_key: Option, +} + +/// Exact selected representation and producer load revision. Dimensions are +/// validated on each vector and must also match at retrieval. #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) struct EmbeddingIdentity { + provider: EmbeddingProvider, endpoint: String, model: String, + profile: String, + revision: String, } /// A non-empty, finite, non-zero vector with its configured provider/model identity. @@ -17,6 +43,33 @@ pub(crate) struct Embedding { } impl Embedding { + pub(crate) fn model(&self) -> &str { + &self.identity.model + } + pub(crate) fn revision_label(&self) -> String { + match serde_json::from_str::(&self.identity.revision) { + Ok(value) => format!( + "{} / {}", + value["metadata"]["effective_metadata"]["upstream_revision"] + .as_str() + .unwrap_or("local package metadata"), + value["serving_cursor"] + .as_str() + .or_else(|| value["loaded_at"].as_str()) + .unwrap_or("unknown load") + ), + Err(_) => self.identity.revision.clone(), + } + } + pub(crate) fn matches_config(&self, config: &EmbeddingConfig) -> bool { + self.identity.provider == config.provider + && config.provider != EmbeddingProvider::Disabled + && self.identity.endpoint == config.base_url.trim_end_matches('/') + && self.identity.model == config.model + && self.identity.profile == config.profile + && (config.provider == EmbeddingProvider::Pumas + || self.identity.revision == config.revision) + } pub(crate) fn new(identity: EmbeddingIdentity, values: Vec) -> Result { if values.is_empty() || values.iter().any(|value| !value.is_finite()) { return Err("embedding must be non-empty and finite".into()); @@ -28,10 +81,12 @@ impl Embedding { } } -/// Transitional OpenAI-compatible adapter. Pumas runtime integration is separate. +/// Real Pumas typed operations or an independently configured HTTP provider. pub struct EmbeddingClient { client: Client, identity: EmbeddingIdentity, + config: EmbeddingConfig, + pumas: Option, } #[derive(Serialize)] @@ -53,30 +108,102 @@ struct EmbedResponseData { } impl EmbeddingClient { + #[cfg(test)] pub fn new(base_url: impl Into, model: impl Into) -> Self { Self { client: Client::new(), identity: EmbeddingIdentity { + provider: EmbeddingProvider::OpenAiCompatible, endpoint: base_url.into().trim_end_matches('/').to_string(), model: model.into(), + profile: String::new(), + revision: "fixture".into(), }, + config: EmbeddingConfig::default(), + pumas: None, + } + } + + pub(crate) async fn from_config(config: &EmbeddingConfig) -> Result { + if config.provider == EmbeddingProvider::Disabled { + return Err( + "Reference embeddings are disabled; select an embedding provider/model".into(), + ); } + if config.model.trim().is_empty() { + return Err("Choose an embedding model".into()); + } + let pumas = if config.provider == EmbeddingProvider::Pumas { + Some(crate::pumas_inference::PumasClient::connect(&config.base_url).await?) + } else { + None + }; + let revision = if let Some(client) = &pumas { + client + .embedding_revision(&config.model, &config.profile) + .await? + } else { + if config.revision.trim().is_empty() { + return Err("Configure an embedding model revision before indexing".into()); + } + config.revision.clone() + }; + Ok(Self { + client: Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .timeout(std::time::Duration::from_secs(180)) + .build() + .map_err(|e| e.to_string())?, + identity: EmbeddingIdentity { + provider: config.provider, + endpoint: config.base_url.trim_end_matches('/').to_owned(), + model: config.model.clone(), + profile: config.profile.clone(), + revision, + }, + config: config.clone(), + pumas, + }) } + #[cfg(test)] pub(crate) fn identity(&self) -> EmbeddingIdentity { self.identity.clone() } pub(crate) async fn embed(&self, text: &str) -> Result { + if let Some(client) = &self.pumas { + let before = client + .embedding_revision(&self.identity.model, &self.identity.profile) + .await?; + if before != self.identity.revision { + return Err("Pumas embedding model revision changed; reindex references".into()); + } + let values = client + .embed(&self.identity.model, &self.identity.profile, text) + .await?; + if client + .embedding_revision(&self.identity.model, &self.identity.profile) + .await? + != before + { + return Err( + "Pumas embedding model changed during inference; reindex references".into(), + ); + } + return Embedding::new(self.identity.clone(), values); + } let url = format!("{}/embeddings", self.identity.endpoint); let body = EmbedRequest { model: &self.identity.model, input: text, }; - let resp = self - .client - .post(&url) - .json(&body) + let request = self.client.post(&url).json(&body); + let request = match self.config.api_key.as_deref().filter(|key| !key.is_empty()) { + Some(key) => request.bearer_auth(key), + None => request, + }; + let resp = request .send() .await .map_err(|e| format!("embedding request failed: {e}"))?; diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index f0333881..81aed130 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -53,7 +53,7 @@ pub(crate) mod command_service_timeline_requests; pub mod context_influence_service; pub(crate) mod context_influence_store; pub(crate) mod context_stack_projection; -pub(crate) mod embeddings; +pub mod embeddings; pub(crate) mod export; pub mod export_service; pub mod graph_proposal_service; @@ -75,6 +75,8 @@ pub(crate) mod propagation_proposal_script_accept; pub(crate) mod propagation_proposal_store; pub(crate) mod propagation_proposal_target; pub(crate) mod propagation_proposal_update; +pub mod pumas_inference; +pub(crate) mod pumas_runtime; pub mod reference_service; pub(crate) mod revision_projection; pub(crate) mod script_block_create; diff --git a/crates/server/src/project_service.rs b/crates/server/src/project_service.rs index 84713d7b..662c6e5f 100644 --- a/crates/server/src/project_service.rs +++ b/crates/server/src/project_service.rs @@ -242,6 +242,8 @@ pub(crate) fn replace_active_project( *state.project_session_id.lock() = uuid::Uuid::new_v4(); state.project_database.set_active_path(path); *active = Some(project); + drop(active); + crate::reference_service::schedule_reopened_index(state); } pub async fn list_projects() -> serde_json::Value { diff --git a/crates/server/src/pumas_inference.rs b/crates/server/src/pumas_inference.rs new file mode 100644 index 00000000..fb7e2e43 --- /dev/null +++ b/crates/server/src/pumas_inference.rs @@ -0,0 +1,337 @@ +//! Consumer of Pumas ad31e391's serving RPC and typed model-operation v1. +use std::time::Duration; + +use pumas_library::discovery::{ + HTTP_DISCOVERY_PATH, HTTP_INSTANCE_GENERATION_HEADER, HTTP_SERVICE_GENERATION_HEADER, + HttpServiceDescription, LoopbackHttpEndpoint, +}; +pub use pumas_library::models::{ServeModelRequest, UnserveModelRequest, UnserveModelResponse}; +use pumas_library::models::{ServedModelLoadState, ServedModelStatus, ServingStatusResponse}; +use reqwest::{Client, RequestBuilder}; +use serde_json::{Value, json}; +use uuid::Uuid; + +#[derive(Clone)] +pub(crate) struct PumasClient { + client: Client, + pub(crate) description: HttpServiceDescription, +} + +#[derive(serde::Serialize)] +pub(crate) struct PumasOperation<'a> { + pub(crate) model: &'a str, + pub(crate) profile: &'a str, + pub(crate) capability: &'a str, + pub(crate) input: Value, + pub(crate) output: &'a str, + pub(crate) options: Value, + pub(crate) stream: bool, +} + +impl PumasClient { + pub(crate) async fn connect(endpoint: &str) -> Result { + let endpoint = LoopbackHttpEndpoint::parse(endpoint).map_err(|e| e.to_string())?; + let client = Client::builder() + .no_proxy() + .redirect(reqwest::redirect::Policy::none()) + .connect_timeout(Duration::from_secs(5)) + .timeout(Duration::from_secs(180)) + .build() + .map_err(|e| e.to_string())?; + let response = client + .get(format!("{}{HTTP_DISCOVERY_PATH}", endpoint.as_str())) + .send() + .await + .map_err(|e| format!("Pumas discovery: {e}"))?; + let description: HttpServiceDescription = + serde_json::from_value(read_json(response).await?) + .map_err(|e| format!("Pumas description: {e}"))?; + if description.endpoint != endpoint || description.advertisement_schema_version != 1 { + return Err("Pumas description endpoint/schema mismatch".into()); + } + description.admission_fence().map_err(|e| e.to_string())?; + if !description + .build_info + .compiled_features + .iter() + .any(|f| f == "pumas-rpc/inference-plugins") + { + return Err("Pumas was built without inference-plugins".into()); + } + Ok(Self { + client, + description, + }) + } + + pub(crate) fn for_description(description: HttpServiceDescription) -> Result { + description.admission_fence().map_err(|e| e.to_string())?; + let client = Client::builder() + .no_proxy() + .redirect(reqwest::redirect::Policy::none()) + .connect_timeout(Duration::from_secs(5)) + .timeout(Duration::from_secs(180)) + .build() + .map_err(|e| e.to_string())?; + Ok(Self { + client, + description, + }) + } + + fn fenced(&self, request: RequestBuilder) -> RequestBuilder { + request + .header( + HTTP_INSTANCE_GENERATION_HEADER, + &self.description.instance.generation, + ) + .header( + HTTP_SERVICE_GENERATION_HEADER, + &self.description.service_generation, + ) + } + + pub(crate) async fn rpc(&self, method: &str, params: Value) -> Result { + let id = Uuid::new_v4().to_string(); + let response = self + .fenced( + self.client + .post(format!("{}/rpc", self.description.endpoint.as_str())), + ) + .json(&json!({"jsonrpc":"2.0", "id":id, "method":method, "params":params})) + .send() + .await + .map_err(|e| format!("Pumas {method}: {e}"))?; + let value = read_json(response).await?; + if value["jsonrpc"] != "2.0" || value["id"] != id { + return Err("Pumas RPC response correlation mismatch".into()); + } + if let Some(error) = value.get("error") { + return Err(format!("Pumas {method}: {error}")); + } + let result = value + .get("result") + .cloned() + .ok_or("Pumas RPC missing result")?; + if result.get("success") == Some(&Value::Bool(false)) + || result.get("loaded") == Some(&Value::Bool(false)) + || result.get("load_error").is_some_and(|e| !e.is_null()) + { + return Err(format!( + "Pumas {method}: {}", + result + .get("load_error") + .or_else(|| result.get("error")) + .unwrap_or(&result) + )); + } + Ok(result) + } + + pub(crate) async fn selected( + &self, + model: &str, + profile: &str, + capability: &str, + ) -> Result<(ServedModelStatus, String), String> { + if model.trim().is_empty() + || model.eq_ignore_ascii_case("auto") + || profile.trim().is_empty() + { + return Err("Choose an explicit Pumas model and runtime profile".into()); + } + let status: ServingStatusResponse = + serde_json::from_value(self.rpc("get_serving_status", json!({})).await?) + .map_err(|e| format!("Pumas serving status: {e}"))?; + if status.snapshot.schema_version != 1 || status.snapshot.cursor.is_empty() { + return Err("Pumas serving snapshot schema/cursor mismatch".into()); + } + let served = status + .snapshot + .served_models + .into_iter() + .find(|s| { + s.model_id == model + && s.profile_id.as_str() == profile + && s.load_state == ServedModelLoadState::Loaded + }) + .ok_or("Selected Pumas model/profile is not loaded; use Load in Pumas")?; + let response = self + .fenced(self.client.get(format!( + "{}/v1/capabilities", + self.description.endpoint.as_str() + ))) + .query(&[("model", model), ("profile", profile)]) + .send() + .await + .map_err(|e| format!("Pumas capabilities: {e}"))?; + let value = read_json(response).await?; + if value["model"] != model + || value["profile"] != profile + || !value["supported_contract_versions"] + .as_array() + .is_some_and(|v| v.contains(&json!(1))) + { + return Err("Pumas capability response correlation/version mismatch".into()); + } + if !value["capabilities"].as_array().is_some_and(|caps| { + caps.iter() + .any(|c| c["capability"] == capability && c["availability"]["state"] == "available") + }) { + return Err(format!( + "Pumas {capability} unavailable for selected model/profile: {}", + value["capabilities"] + )); + } + Ok((served, status.snapshot.cursor)) + } + + /// Session revision includes actual Pumas provenance and the selected load. + /// It is deliberately not a claim that a mutable package is weight-attested. + pub(crate) async fn embedding_revision( + &self, + model: &str, + profile: &str, + ) -> Result { + let (status, serving_cursor) = self.selected(model, profile, "text_embedding").await?; + let metadata = self + .rpc("get_library_model_metadata", json!({"model_id":model})) + .await?; + if metadata["model_id"] != model { + return Err("Pumas metadata model mismatch".into()); + } + // The pinned adapters leave loaded_at unset. The producer's monotonic + // serving cursor fences even same-model unload/reload with unchanged + // metadata. It conservatively invalidates on unrelated serving changes. + Ok(json!({"service":self.description.service_generation, "instance":self.description.instance.generation, + "provider":status.provider, "profile":profile, "serving_cursor":serving_cursor, "loaded_at":status.loaded_at, + "metadata": metadata}).to_string()) + } + + pub(crate) async fn operation( + &self, + request: PumasOperation<'_>, + ) -> Result<(String, reqwest::Response), String> { + self.selected(request.model, request.profile, request.capability) + .await?; + let id = Uuid::new_v4().to_string(); + let mut body = serde_json::to_value(request).map_err(|e| e.to_string())?; + body["contract_version"] = json!(1); + body["request_id"] = json!(id); + let response = self + .fenced(self.client.post(format!( + "{}/v1/model-operations", + self.description.endpoint.as_str() + ))) + .json(&body) + .send() + .await + .map_err(|e| format!("Pumas operation: {e}"))?; + if !response.status().is_success() { + return Err(format!( + "Pumas operation: {}", + read_json(response).await.err().unwrap_or_default() + )); + } + Ok((id, response)) + } + + pub(crate) async fn embed( + &self, + model: &str, + profile: &str, + text: &str, + ) -> Result, String> { + let (id, response) = self + .operation(PumasOperation { + model, + profile, + capability: "text_embedding", + input: json!({"kind":"text", "text":text}), + output: "embeddings_float32", + options: json!({"kind":"embeddings"}), + stream: false, + }) + .await?; + let value = read_json(response).await?; + if value["contract_version"] != 1 + || value["request_id"] != id + || value["result"]["kind"] != "embeddings" + { + return Err("Pumas embedding response contract/correlation mismatch".into()); + } + let mut vectors: Vec> = serde_json::from_value(value["result"]["vectors"].clone()) + .map_err(|e| format!("Pumas embedding vectors: {e}"))?; + if vectors.len() != 1 { + return Err("Pumas embedding response must contain one vector".into()); + } + Ok(vectors.remove(0)) + } +} + +pub(crate) async fn read_json(mut response: reqwest::Response) -> Result { + let status = response.status(); + let mut body = Vec::new(); + while let Some(chunk) = response + .chunk() + .await + .map_err(|e| format!("Pumas response transport: {e}"))? + { + if body.len() + chunk.len() > 32 * 1024 * 1024 { + return Err("Pumas response exceeds 32 MiB".into()); + } + body.extend_from_slice(&chunk); + } + let value: Value = serde_json::from_slice(&body) + .map_err(|e| format!("Pumas HTTP {status}: invalid JSON ({e})"))?; + if !status.is_success() || value.get("error").is_some() { + return Err(format!( + "Pumas HTTP {status}: {}", + value.get("error").unwrap_or(&value) + )); + } + Ok(value) +} + +/// Explicit lifecycle controls: Pumas owns provider processes and native sessions. +/// No implicit unload or shutdown is sent to a borrowed Pumas service. +pub async fn load_model(endpoint: &str, request: ServeModelRequest) -> Result { + let client = PumasClient::connect(endpoint).await?; + client.rpc("serve_model", json!({"request":request})).await +} +pub async fn unload_model( + endpoint: &str, + request: UnserveModelRequest, +) -> Result { + let client = PumasClient::connect(endpoint).await?; + let result = client + .rpc("unserve_model", json!({"request":request})) + .await?; + let response: UnserveModelResponse = + serde_json::from_value(result).map_err(|e| format!("Pumas unload response: {e}"))?; + if !response.success || response.error.is_some() { + return Err(format!( + "Pumas unserve_model: {}", + response.error.as_deref().unwrap_or("unload failed") + )); + } + // `unloaded: false` with no error means already unserved, not an unload + // receipt. Preserve the typed distinction for the operator interface. + Ok(response) +} + +pub async fn catalog(endpoint: &str) -> Result { + let client = PumasClient::connect(endpoint).await?; + let models = client.rpc("get_models", json!({})).await?; + let models = models["models"].as_object().ok_or("Pumas catalog models must be an object")?.values().map(|entry| json!({ + "id":entry["id"], "official_name":entry["displayName"], "model_type":entry["modelType"] + })).collect::>(); + let profiles = client + .rpc("get_runtime_profiles_snapshot", json!({})) + .await?; + Ok(json!({"models":{"models":models}, "profiles":profiles})) +} + +#[cfg(test)] +#[path = "pumas_inference_tests.rs"] +pub(crate) mod tests; diff --git a/crates/server/src/pumas_inference_tests.rs b/crates/server/src/pumas_inference_tests.rs new file mode 100644 index 00000000..76d8dd13 --- /dev/null +++ b/crates/server/src/pumas_inference_tests.rs @@ -0,0 +1,715 @@ +//! Controlled producer fixtures. Vectors and screenplay text are fixture outputs, +//! never real-model evidence. These exercise the consumer's actual wire path. +use super::*; +use crate::{ + embeddings::{EmbeddingClient, EmbeddingConfig, EmbeddingProvider}, + state::{AiConfig, AppState, BackendType}, +}; +use parking_lot::Mutex; +use std::sync::{ + Arc, + atomic::{AtomicBool, AtomicUsize, Ordering}, +}; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::TcpListener, +}; + +pub(crate) struct Fixture { + pub(crate) url: String, + pub(crate) requests: Arc>>, + revision: Arc, + serving_cursor: Arc, + change_cursor_during_embedding: Arc, + dimensions: Arc, + unload_response: Arc>>, + fail_embedding: Arc, + unavailable: Arc, + loaded: Arc, + task: tokio::task::JoinHandle<()>, +} +impl Drop for Fixture { + fn drop(&mut self) { + self.task.abort(); + } +} +impl Fixture { + pub(crate) async fn start() -> Self { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let requests = Arc::new(Mutex::new(Vec::new())); + let revision = Arc::new(AtomicUsize::new(1)); + let serving_cursor = Arc::new(AtomicUsize::new(1)); + let change_cursor_during_embedding = Arc::new(AtomicBool::new(false)); + let dimensions = Arc::new(AtomicUsize::new(3)); + let unload_response = Arc::new(Mutex::new(None::)); + let fail_embedding = Arc::new(AtomicBool::new(false)); + let unavailable = Arc::new(AtomicBool::new(false)); + let loaded = Arc::new(AtomicBool::new(true)); + let (root_url, reqs, rev, cursor, change_cursor, dims, unload, fail, unavail, is_loaded) = ( + url.clone(), + requests.clone(), + revision.clone(), + serving_cursor.clone(), + change_cursor_during_embedding.clone(), + dimensions.clone(), + unload_response.clone(), + fail_embedding.clone(), + unavailable.clone(), + loaded.clone(), + ); + let task = tokio::spawn(async move { + loop { + let (mut socket, _) = listener.accept().await.unwrap(); + let (url, reqs, rev, cursor, change_cursor, dims, unload, fail, unavail, is_loaded) = ( + root_url.clone(), + reqs.clone(), + rev.clone(), + cursor.clone(), + change_cursor.clone(), + dims.clone(), + unload.clone(), + fail.clone(), + unavail.clone(), + is_loaded.clone(), + ); + tokio::spawn(async move { + let mut request = Vec::new(); + let (header_end, length) = loop { + let mut buffer = [0; 4096]; + let count = socket.read(&mut buffer).await.unwrap(); + if count == 0 { + return; + } + request.extend_from_slice(&buffer[..count]); + if let Some(end) = request.windows(4).position(|v| v == b"\r\n\r\n") { + let headers = String::from_utf8_lossy(&request[..end]).to_lowercase(); + let length = headers + .lines() + .find_map(|l| l.strip_prefix("content-length: ")) + .map(|v| v.parse::().unwrap()) + .unwrap_or(0); + break (end + 4, length); + } + }; + while request.len() < header_end + length { + let mut buffer = [0; 4096]; + let count = socket.read(&mut buffer).await.unwrap(); + if count == 0 { + return; + } + request.extend_from_slice(&buffer[..count]); + } + let headers = String::from_utf8_lossy(&request[..header_end]); + let path = headers + .lines() + .next() + .unwrap() + .split_whitespace() + .nth(1) + .unwrap() + .to_owned(); + let body: Value = if length == 0 { + json!({}) + } else { + serde_json::from_slice(&request[header_end..header_end + length]).unwrap() + }; + reqs.lock().push((path.clone(), body.clone())); + if path != HTTP_DISCOVERY_PATH { + assert!(headers.contains("pumas-instance-generation: instance-fixture")); + assert!(headers.contains("pumas-service-generation: service-fixture")); + } + let response = if path == HTTP_DISCOVERY_PATH { + description(&url) + } else if path.starts_with("/v1/capabilities?") { + let query = reqwest::Url::parse(&format!("{url}{path}")).unwrap(); + let pairs: std::collections::HashMap<_, _> = + query.query_pairs().into_owned().collect(); + json!({"supported_contract_versions":[1],"model":pairs["model"],"profile":pairs["profile"],"capabilities":[ + {"capability":"chat_generation","availability":{"state":if unavail.load(Ordering::SeqCst) {"unavailable"} else {"available"}}}, + {"capability":"text_embedding","availability":{"state":if unavail.load(Ordering::SeqCst) {"unavailable"} else {"available"}}} + ]}) + } else if path == "/rpc" { + let result = match body["method"].as_str().unwrap() { + "get_serving_status" => { + json!({"success":true,"snapshot":{"schema_version":1,"cursor":format!("serving:{}",cursor.load(Ordering::SeqCst)),"endpoint":{"endpoint_mode":"pumas_gateway","model_count":2},"last_errors":[],"served_models":if is_loaded.load(Ordering::SeqCst) {vec![served("embed"),served("chat")]} else {vec![]}}}) + } + "get_library_model_metadata" => { + json!({"success":true,"model_id":body["params"]["model_id"],"effective_metadata":{"upstream_revision":format!("weights-{}",rev.load(Ordering::SeqCst))}}) + } + "serve_model" if body["params"]["request"]["model_id"] == "missing" => { + json!({"success":true,"loaded":false,"load_error":{"code":"model_not_found","message":"fixture model missing"}}) + } + "serve_model" => { + is_loaded.store(true, Ordering::SeqCst); + cursor.fetch_add(1, Ordering::SeqCst); + json!({"success":true,"loaded":true}) + } + "unserve_model" if unload.lock().is_some() => { + unload.lock().as_ref().unwrap().clone() + } + "unserve_model" => { + is_loaded.store(false, Ordering::SeqCst); + cursor.fetch_add(1, Ordering::SeqCst); + json!({"success":true,"unloaded":true}) + } + "shutdown" => json!({"success":true}), + _ => panic!("Unexpected fixture RPC {}", body["method"]), + }; + json!({"jsonrpc":"2.0","id":body["id"],"result":result}) + } else if path == "/v1/model-operations" { + if body["capability"] == "text_embedding" { + if change_cursor.swap(false, Ordering::SeqCst) { + cursor.fetch_add(1, Ordering::SeqCst); + } + if fail.load(Ordering::SeqCst) { + json!({"contract_version":1,"request_id":body["request_id"],"error":{"code":"provider_failure","outcome":"not_admitted"}}) + } else { + let vector = (1..=dims.load(Ordering::SeqCst)) + .map(|v| v as f32) + .collect::>(); + json!({"contract_version":1,"request_id":body["request_id"],"result":{"kind":"embeddings","vectors":[vector]}}) + } + } else { + let events = [ + json!({"kind":"started","contract_version":1,"request_id":body["request_id"],"capability":"chat_generation","model":"chat","profile":"fixture"}), + json!({"kind":"delta","request_id":body["request_id"],"text":"Fixture screenplay"}), + json!({"kind":"completed","request_id":body["request_id"],"finish_reason":"stop"}), + ]; + let text = events + .iter() + .map(|e| format!("data: {e}\n\n")) + .collect::(); + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{text}", + text.len() + ); + let _ = socket.write_all(response.as_bytes()).await; + return; + } + } else { + panic!("Unexpected fixture path {path}") + }; + let text = response.to_string(); + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{text}", + text.len() + ); + let _ = socket.write_all(response.as_bytes()).await; + }); + } + }); + Self { + url, + requests, + revision, + serving_cursor, + change_cursor_during_embedding, + dimensions, + unload_response, + fail_embedding, + unavailable, + loaded, + task, + } + } + pub(crate) fn embedding_config(&self) -> EmbeddingConfig { + EmbeddingConfig { + provider: EmbeddingProvider::Pumas, + base_url: self.url.clone(), + model: "embed".into(), + profile: "fixture".into(), + ..Default::default() + } + } + + pub(crate) fn change_embedding_representation(&self, revision: usize, dimensions: usize) { + self.revision.store(revision, Ordering::SeqCst); + self.dimensions.store(dimensions, Ordering::SeqCst); + self.serving_cursor.fetch_add(1, Ordering::SeqCst); + } +} +pub(crate) fn description(url: &str) -> Value { + json!({"advertisement_schema_version":1,"service_generation":"service-fixture","endpoint":url, + "instance":{"discovery_schema_version":1,"registry_library_id":"fixture","library_root":"/tmp/fixture","generation":"instance-fixture","pumas_version":"0.8.0-rc.1","protocols":[],"capabilities":[],"model_ref_schema_version":1,"selector_schema_version":1}, + "build_info":{"build_info_schema_version":1,"component":"pumas-rpc","package_version":"0.8.0-rc.1","compiled_features":["pumas-rpc/inference-plugins"],"protocols":[],"schemas":[{"name":"pumas.http-admission-fence","version":1}]}}) +} +fn served(model: &str) -> Value { + // The pinned producer's real serving adapters omit the optional timestamp. + json!({"model_id":model,"provider":"onnx_runtime","profile_id":"fixture","load_state":"loaded","device_mode":"cpu","keep_loaded":true}) +} + +#[tokio::test] +async fn pumas_producer_without_load_timestamp_can_embed() { + let fixture = Fixture::start().await; + let client = EmbeddingClient::from_config(&fixture.embedding_config()) + .await + .expect("pinned producer does not provide loaded_at"); + assert_eq!( + client.embed("Reference").await.unwrap().values, + vec![1., 2., 3.] + ); +} + +#[tokio::test] +async fn pumas_serving_cursor_fences_same_model_reload_and_unrelated_serving_changes() { + use eidetic_core::{Template, ai::prompt::build_generate_request}; + let fixture = Fixture::start().await; + let state = AppState::new().await; + state.ai_config.lock().embedding = fixture.embedding_config(); + let project = Template::MultiCam.build_project("Cursor fixture"); + let path = std::env::temp_dir().join(format!("eidetic-cursor-{}.db", Uuid::new_v4())); + let mut request = build_generate_request(&project, project.timeline.nodes[0].id).unwrap(); + crate::project_service::replace_active_project(&state, project, path.clone()); + crate::reference_service::upload_reference( + &state, + crate::reference_service::UploadReferenceRequest { + name: "Reference".into(), + content: "Unique reference detail".into(), + doc_type: "StyleGuide".into(), + }, + ) + .unwrap(); + ready(&state).await; + let config = state.ai_config.lock().clone(); + let old_client = EmbeddingClient::from_config(&config.embedding) + .await + .unwrap(); + let producer = PumasClient::connect(&fixture.url).await.unwrap(); + // Same model, provider, profile, dimensions and package metadata; only the + // real producer's serving cursor distinguishes this unload/reload. + let metadata = producer + .rpc("get_library_model_metadata", json!({"model_id":"embed"})) + .await + .unwrap(); + producer + .rpc("unserve_model", json!({"request":{"model_id":"embed"}})) + .await + .unwrap(); + producer + .rpc("serve_model", json!({"request":{"model_id":"embed"}})) + .await + .unwrap(); + assert_eq!( + metadata, + producer + .rpc("get_library_model_metadata", json!({"model_id":"embed"})) + .await + .unwrap() + ); + assert!( + old_client + .embed("Query") + .await + .unwrap_err() + .contains("revision changed") + ); + let scope = state.vector_store.lock().scope(); + crate::ai_generation_runtime::attach_rag_context(&state, &config, &path, scope, &mut request) + .await; + assert!(request.rag_context.is_empty()); + assert_eq!( + crate::reference_service::index_status(&state) + .unwrap() + .documents[0] + .state, + "stale" + ); + crate::reference_service::schedule_reindex(&state); + ready(&state).await; + let scope = state.vector_store.lock().scope(); + crate::ai_generation_runtime::attach_rag_context(&state, &config, &path, scope, &mut request) + .await; + assert_eq!(request.rag_context.len(), 1); + // An unrelated serving mutation is intentionally conservative too. + producer + .rpc("serve_model", json!({"request":{"model_id":"chat"}})) + .await + .unwrap(); + crate::ai_generation_runtime::attach_rag_context(&state, &config, &path, scope, &mut request) + .await; + assert!(request.rag_context.is_empty()); + assert_eq!( + crate::reference_service::index_status(&state) + .unwrap() + .documents[0] + .state, + "stale" + ); + state.shutdown_tasks_async().await; + if path.exists() { + std::fs::remove_file(path).unwrap(); + } +} + +#[tokio::test] +async fn pumas_serving_change_during_embedding_fails_without_publishing_chunks() { + let fixture = Fixture::start().await; + fixture + .change_cursor_during_embedding + .store(true, Ordering::SeqCst); + let state = AppState::new().await; + state.ai_config.lock().embedding = fixture.embedding_config(); + *state.project.lock() = Some(eidetic_core::Template::MultiCam.build_project("Cursor change")); + crate::reference_service::upload_reference( + &state, + crate::reference_service::UploadReferenceRequest { + name: "Reference".into(), + content: "Reference content".into(), + doc_type: "StyleGuide".into(), + }, + ) + .unwrap(); + tokio::time::timeout(Duration::from_secs(10), async { + loop { + let status = crate::reference_service::index_status(&state).unwrap(); + if status.documents[0].state == "failed" { + assert!( + status.documents[0] + .error + .as_ref() + .unwrap() + .contains("changed during inference") + ); + assert_eq!(status.documents[0].indexed_chunks, 0); + break; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert!(state.vector_store.lock().is_empty()); + state.shutdown_tasks_async().await; +} + +#[tokio::test] +async fn pumas_real_wire_path_load_embed_revision_change_unload_and_errors() { + let fixture = Fixture::start().await; + let config = fixture.embedding_config(); + let client = EmbeddingClient::from_config(&config).await.unwrap(); + let first = client.embed("Reference text").await.unwrap(); + assert_eq!(first.values, vec![1., 2., 3.]); + fixture.revision.store(2, Ordering::SeqCst); + assert!( + client + .embed("Query") + .await + .unwrap_err() + .contains("revision changed") + ); + let replacement = EmbeddingClient::from_config(&config) + .await + .unwrap() + .embed("Query") + .await + .unwrap(); + assert_ne!(first.identity, replacement.identity); + fixture.fail_embedding.store(true, Ordering::SeqCst); + assert!( + EmbeddingClient::from_config(&config) + .await + .unwrap() + .embed("Query") + .await + .unwrap_err() + .contains("provider_failure") + ); + fixture.unavailable.store(true, Ordering::SeqCst); + assert!(EmbeddingClient::from_config(&config).await.is_err()); + fixture.unavailable.store(false, Ordering::SeqCst); + let unload: UnserveModelRequest = serde_json::from_value( + json!({"model_id":"embed","provider":"onnx_runtime","profile_id":"fixture"}), + ) + .unwrap(); + unload_model(&fixture.url, unload).await.unwrap(); + assert!(!fixture.loaded.load(Ordering::SeqCst)); + assert!(EmbeddingClient::from_config(&config).await.is_err()); + let request = |model| { + serde_json::from_value::(json!({"model_id":model,"config":{"provider":"onnx_runtime","profile_id":"fixture","device_mode":"cpu"}})).unwrap() + }; + assert!( + load_model(&fixture.url, request("missing")) + .await + .unwrap_err() + .contains("model_not_found") + ); + load_model(&fixture.url, request("embed")).await.unwrap(); + assert!(EmbeddingClient::from_config(&config).await.is_ok()); +} + +pub(crate) async fn ready(state: &AppState) { + tokio::time::timeout(Duration::from_secs(10), async { + loop { + let status = crate::reference_service::index_status(state).unwrap(); + if status.documents.iter().all(|d| d.state == "ready") { + return; + } + if status.documents.iter().any(|d| d.state == "failed") { + panic!("Fixture index failed: {:?}", status); + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); +} + +#[tokio::test] +async fn unload_rejects_typed_provider_failures_even_when_rpc_success_is_true() { + let fixture = Fixture::start().await; + for error in [ + "ONNX Runtime could not unload the selected model", + "Ollama could not unload the selected model", + ] { + *fixture.unload_response.lock() = Some(json!({ + "success": true, "error": error, "unloaded": false + })); + let request = serde_json::from_value(json!({ + "model_id": "embed", "provider": "onnx_runtime", "profile_id": "fixture" + })) + .unwrap(); + let failure = unload_model(&fixture.url, request).await.unwrap_err(); + assert!(failure.contains(error), "{failure}"); + assert!(fixture.loaded.load(Ordering::SeqCst)); + assert!( + EmbeddingClient::from_config(&fixture.embedding_config()) + .await + .is_ok() + ); + } +} + +#[tokio::test] +async fn unload_preserves_already_unserved_outcome_and_rejects_missing_receipt() { + let fixture = Fixture::start().await; + fixture.loaded.store(false, Ordering::SeqCst); + let request = || { + serde_json::from_value(json!({ + "model_id": "embed", "provider": "onnx_runtime", "profile_id": "fixture" + })) + .unwrap() + }; + *fixture.unload_response.lock() = Some(json!({"success": true, "unloaded": false})); + let response = unload_model(&fixture.url, request()).await.unwrap(); + assert!(!response.unloaded); + assert!(response.error.is_none()); + *fixture.unload_response.lock() = Some(json!({"success": true})); + assert!( + unload_model(&fixture.url, request()) + .await + .unwrap_err() + .contains("unloaded") + ); +} + +#[tokio::test] +async fn upload_reopen_change_model_retrieve_and_chat_generation_use_real_consumer_path() { + use eidetic_core::{Template, ai::prompt::build_generate_request}; + let fixture = Fixture::start().await; + let state = AppState::new().await; + state.ai_config.lock().embedding = fixture.embedding_config(); + let project = Template::MultiCam.build_project("Pumas fixture"); + let path = std::env::temp_dir().join(format!("eidetic-pumas-{}.db", Uuid::new_v4())); + crate::project_service::replace_active_project(&state, project, path.clone()); + let doc = crate::reference_service::upload_reference( + &state, + crate::reference_service::UploadReferenceRequest { + name: "Secret location".into(), + content: "The secret station is called Marmalade Harbor.".into(), + doc_type: "WorldBuilding".into(), + }, + ) + .unwrap(); + ready(&state).await; + let saved = state.project.lock().as_ref().unwrap().clone(); + crate::persistence::save_project(&saved, &path, None) + .await + .unwrap(); + let (reopened, _) = crate::persistence::load_project(&path).await.unwrap(); + let old_scope = state.vector_store.lock().scope(); + crate::project_service::replace_active_project(&state, reopened.clone(), path.clone()); + ready(&state).await; + let scope = state.vector_store.lock().scope(); + assert_ne!(old_scope, scope); + let mut request = build_generate_request(&reopened, reopened.timeline.nodes[0].id).unwrap(); + let config = state.ai_config.lock().clone(); + crate::ai_generation_runtime::attach_rag_context(&state, &config, &path, scope, &mut request) + .await; + assert_eq!(request.rag_context[0].content, doc.content); + let prompt = crate::prompt_format::build_chat_prompt(&request); + let config = AiConfig { + backend_type: BackendType::Pumas, + base_url: fixture.url.clone(), + model: "chat".into(), + pumas_profile: "fixture".into(), + ..Default::default() + }; + assert_eq!( + crate::ai_backends::Backend::from_config(&config) + .generate_full(&prompt, &config) + .await + .unwrap(), + "Fixture screenplay" + ); + let sent = fixture + .requests + .lock() + .iter() + .find(|(path, body)| { + path == "/v1/model-operations" && body["capability"] == "chat_generation" + }) + .unwrap() + .1 + .clone(); + assert!( + sent["input"]["messages"] + .to_string() + .contains("Marmalade Harbor") + ); + fixture.revision.store(2, Ordering::SeqCst); + let embedding_config = state.ai_config.lock().clone(); + crate::ai_generation_runtime::attach_rag_context( + &state, + &embedding_config, + &path, + scope, + &mut request, + ) + .await; + assert!(request.rag_context.is_empty()); + crate::reference_service::schedule_reindex(&state); + ready(&state).await; + crate::reference_service::delete_reference(&state, doc.id.0).unwrap(); + assert!(state.vector_store.lock().is_empty()); + state.shutdown_tasks_async().await; + // Borrowing a producer for inference never sends runtime shutdown. + assert!( + !fixture + .requests + .lock() + .iter() + .any(|(_, body)| body["method"] == "shutdown") + ); + std::fs::remove_file(path).unwrap(); +} + +#[tokio::test] +async fn embedding_failure_is_visible_and_does_not_publish_partial_document() { + let fixture = Fixture::start().await; + fixture.fail_embedding.store(true, Ordering::SeqCst); + let state = AppState::new().await; + state.ai_config.lock().embedding = fixture.embedding_config(); + *state.project.lock() = Some(eidetic_core::Template::MultiCam.build_project("Failure")); + crate::reference_service::upload_reference( + &state, + crate::reference_service::UploadReferenceRequest { + name: "Failed reference".into(), + content: "Test source".into(), + doc_type: "StyleGuide".into(), + }, + ) + .unwrap(); + tokio::time::timeout(Duration::from_secs(10), async { + loop { + let status = crate::reference_service::index_status(&state).unwrap(); + if status.documents[0].state == "failed" { + assert!( + status.documents[0] + .error + .as_ref() + .unwrap() + .contains("provider_failure") + ); + assert_eq!(status.documents[0].indexed_chunks, 0); + break; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert!(state.vector_store.lock().is_empty()); + state.shutdown_tasks_async().await; +} + +#[tokio::test] +async fn remote_or_unsupported_pumas_endpoint_is_refused() { + assert!(PumasClient::connect("https://example.com").await.is_err()); + assert!(PumasClient::connect("http://localhost:8080").await.is_err()); + assert!( + EmbeddingClient::from_config(&EmbeddingConfig::default()) + .await + .is_err() + ); +} + +#[tokio::test] +async fn pumas_embedding_selection_change_invalidates_pending_source_tickets() { + use crate::ai_service::{AiConfigUpdate, update_ai_config}; + let fixture = Fixture::start().await; + let state = AppState::new().await; + let mut project = eidetic_core::Template::MultiCam.build_project("Model change"); + project + .references + .push(eidetic_core::reference::ReferenceDocument::new( + "Ref", + "Source content", + eidetic_core::reference::ReferenceType::StyleGuide, + )); + *state.project.lock() = Some(project); + update_ai_config( + &state, + AiConfigUpdate { + embedding: Some(fixture.embedding_config()), + ..Default::default() + }, + ); + ready(&state).await; + let scope = state.vector_store.lock().scope(); + let mut replacement = fixture.embedding_config(); + replacement.model = "missing-model".into(); + update_ai_config( + &state, + AiConfigUpdate { + embedding: Some(replacement), + ..Default::default() + }, + ); + assert_ne!(scope, state.vector_store.lock().scope()); + assert!(state.vector_store.lock().is_empty()); + tokio::time::timeout(Duration::from_secs(10), async { + while crate::reference_service::index_status(&state) + .unwrap() + .documents[0] + .state + != "failed" + { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + update_ai_config( + &state, + AiConfigUpdate { + embedding: Some(fixture.embedding_config()), + ..Default::default() + }, + ); + ready(&state).await; + state.shutdown_tasks_async().await; +} + +#[tokio::test] +async fn selected_root_startup_failure_is_visible_without_provider_fallback() { + let mut state = AppState::new().await; + state.pumas_startup_error = Some("selected owner unavailable".into()); + state.ai_config.lock().backend_type = BackendType::Pumas; + state.ai_config.lock().base_url.clear(); + let status = crate::ai_service::get_ai_status(&state).await; + assert_eq!(status.backend, BackendType::Pumas); + assert!(!status.connected); + assert!(status.error.unwrap().contains("selected owner unavailable")); + state.shutdown_tasks_async().await; +} diff --git a/crates/server/src/pumas_runtime.rs b/crates/server/src/pumas_runtime.rs new file mode 100644 index 00000000..3cd6edb8 --- /dev/null +++ b/crates/server/src/pumas_runtime.rs @@ -0,0 +1,550 @@ +//! Optional selected-root control-plane bootstrap. Pumas owns inference children. +use crate::pumas_inference::PumasClient; +use pumas_library::discovery::{CompatibilityRequirements, HttpServiceDescription, LocalDiscovery}; +use serde::Deserialize; +use serde_json::json; +use std::{path::Path, process::Stdio, sync::Arc, time::Duration}; +use tokio::{ + io::{AsyncBufReadExt, BufReader}, + process::{Child, Command}, + sync::Mutex, + task::JoinHandle, +}; + +pub(crate) struct PumasRuntime { + pub(crate) description: HttpServiceDescription, + owner: Arc>, +} + +/// Child and output completion share the existing runtime lifecycle owner. +/// The composition root must still retain that owner through async cleanup. +#[derive(Default)] +struct RuntimeOwner { + child: Option, + stdout: Option>>, + service_owned: bool, + terminal_error: Option, +} + +#[derive(Deserialize)] +struct Bootstrap { + bootstrap_schema_version: u32, + ownership: String, + description: HttpServiceDescription, +} + +impl PumasRuntime { + pub(crate) async fn from_environment() -> Result>, String> { + let Some(root) = std::env::var_os("EIDETIC_PUMAS_ROOT") else { + return Ok(None); + }; + let binary = std::env::var_os("EIDETIC_PUMAS_RPC_BIN"); + Self::attach(Path::new(&root), binary.as_deref().map(Path::new)) + .await + .map(|runtime| Some(Arc::new(runtime))) + } + + pub(crate) async fn attach(root: &Path, binary: Option<&Path>) -> Result { + if let Some(binary) = binary { + if !cfg!(target_os = "linux") { + return Err("Pumas owned bootstrap is currently qualified on Linux only; run Pumas separately and borrow its endpoint".into()); + } + let root = root.to_path_buf(); + let binary = binary.to_path_buf(); + let (reply, result) = tokio::sync::oneshot::channel(); + // Once admitted, construction/drain outlives a cancelled caller. + tokio::spawn(async move { + let runtime = Self::start(&root, &binary).await; + if let Err(Ok(runtime)) = reply.send(runtime) { + if let Err(error) = runtime.shutdown().await { + tracing::error!(%error,"Cancelled Pumas startup cleanup failed"); + } + } + }); + return result.await.map_err(|_| "Pumas startup task stopped")?; + } + let root = root.canonicalize().map_err(|e| e.to_string())?; + let service = LocalDiscovery::open() + .map_err(|e| e.to_string())? + .borrow_http_service(&root, &CompatibilityRequirements::default()) + .await + .map_err(|e| e.to_string())?; + Ok(Self { + description: service.description().clone(), + owner: Arc::new(Mutex::new(RuntimeOwner::default())), + }) + } + + async fn start(root: &Path, binary: &Path) -> Result { + let mut child = Command::new(binary) + .arg("--attach-or-start-local-http") + .arg("--launcher-root") + .arg(root) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()) + .spawn() + .map_err(|e| format!("Pumas bootstrap: {e}"))?; + let stdout = child.stdout.take().ok_or("Pumas stdout unavailable")?; + let mut reader = BufReader::new(stdout); + let ack = tokio::time::timeout(Duration::from_secs(60), async { + let mut total = 0; + loop { + let mut line = String::new(); + let count = reader + .read_line(&mut line) + .await + .map_err(|e| e.to_string())?; + total += count; + if count == 0 { + return Err("Pumas exited before local access acknowledgement".to_owned()); + } + if total > 1024 * 1024 { + return Err("Pumas startup output exceeds limit".to_owned()); + } + if let Some(value) = line.strip_prefix("PUMAS_LOCAL_ACCESS=") { + return serde_json::from_str::(value.trim()) + .map_err(|e| e.to_string()); + } + } + }) + .await + .map_err(|_| "Pumas startup timed out".to_owned()) + .and_then(|r| r); + let ack = match ack { + Ok(ack) + if ack.bootstrap_schema_version == 1 + && matches!(ack.ownership.as_str(), "owned" | "borrowed") => + { + ack + } + result => { + terminate_startup(&mut child).await?; + return Err(result + .err() + .unwrap_or("Unsupported Pumas bootstrap acknowledgement".into())); + } + }; + // Retain completion and I/O errors with the child. Explicit shutdown + // joins the drain; cancelled construction/last Drop still need the + // composition-root owner described in the lifecycle acceptance report. + let runtime = Self { + description: ack.description, + owner: Arc::new(Mutex::new(RuntimeOwner { + child: Some(child), + stdout: Some(tokio::spawn(async move { + tokio::io::copy(&mut reader, &mut tokio::io::sink()).await + })), + service_owned: ack.ownership == "owned", + terminal_error: None, + })), + }; + if ack.ownership == "borrowed" { + let mut owner = runtime.owner.lock().await; + let child = owner.child.as_mut().expect("bootstrap child retained"); + let exit = match tokio::time::timeout(Duration::from_secs(10), child.wait()).await { + Ok(result) => result.map_err(|e| format!("Borrowed Pumas bootstrap wait: {e}"))?, + Err(_) => { + // This PID is only our spawned attach helper. The service + // identified by the borrowed acknowledgement is external; + // never send it shutdown/unload or signal its process. + child + .kill() + .await + .map_err(|e| format!("Borrowed Pumas helper termination: {e}"))?; + owner.child = None; + let output = finish_stdout(&mut owner).await; + return Err(match output { + Ok(()) => { + "Borrowed Pumas bootstrap did not exit; helper terminated and reaped" + .into() + } + Err(error) => { + format!("Borrowed Pumas bootstrap did not exit; helper reaped; {error}") + } + }); + } + }; + owner.child = None; + if !exit.success() { + owner.terminal_error = Some(format!("Borrowed Pumas bootstrap exited with {exit}")); + } + finish_stdout(&mut owner).await?; + } + let client = PumasClient::connect(runtime.description.endpoint.as_str()).await; + match client { + Ok(client) if client.description == runtime.description => Ok(runtime), + result => { + runtime.shutdown().await?; + Err(result + .err() + .unwrap_or("Pumas bootstrap generation changed".into())) + } + } + } + + pub(crate) async fn shutdown(&self) -> Result<(), String> { + shutdown_owner(&self.owner, &self.description).await + } +} + +async fn shutdown_owner( + owner: &Mutex, + description: &HttpServiceDescription, +) -> Result<(), String> { + let mut owner = owner.lock().await; + let service_owned = owner.service_owned; + if let Some(child) = owner.child.as_mut() { + let exit = if let Some(exit) = child.try_wait().map_err(|e| e.to_string())? { + // An exited child has no shutdown authority over a replacement. + exit + } else if service_owned { + let client = PumasClient::for_description(description.clone())?; + tokio::time::timeout(Duration::from_secs(30), client.rpc("shutdown", json!({}))) + .await + .map_err(|_| "Owned Pumas shutdown request timed out")??; + tokio::time::timeout(Duration::from_secs(30), child.wait()) + .await + .map_err(|_| "Owned Pumas drain timed out; child retained for retry")? + .map_err(|e| e.to_string())? + } else { + // Failed borrowed bootstrap: terminate/reap our helper only. + child + .kill() + .await + .map_err(|e| format!("Borrowed Pumas helper termination: {e}"))?; + child + .try_wait() + .map_err(|e| e.to_string())? + .ok_or("Borrowed Pumas helper exit missing")? + }; + owner.child = None; + if !exit.success() && service_owned { + owner.terminal_error = Some(format!("Owned Pumas exited with {exit}")); + } + } + finish_stdout(&mut owner).await +} + +async fn finish_stdout(owner: &mut RuntimeOwner) -> Result<(), String> { + if let Some(handle) = owner.stdout.as_mut() { + let result = tokio::time::timeout(Duration::from_secs(30), handle) + .await + .map_err(|_| "Pumas stdout drain timed out; completion retained for retry")?; + owner.stdout = None; + let failure = match result { + Ok(Ok(_)) => None, + Ok(Err(error)) => Some(format!("Pumas stdout drain failed: {error}")), + Err(error) => Some(format!("Pumas stdout drain task failed: {error}")), + }; + if let Some(failure) = failure { + owner.terminal_error = Some(match owner.terminal_error.take() { + Some(previous) => format!("{previous}; {failure}"), + None => failure, + }); + } + } + match &owner.terminal_error { + Some(error) => Err(error.clone()), + None => Ok(()), + } +} + +async fn terminate_startup(child: &mut Child) -> Result<(), String> { + if let Some(pid) = child.id() { + // The unreaped child handle retains identity; this is only our bootstrap + // subprocess. SIGTERM asks Pumas to drain, including before readiness. + #[cfg(unix)] + Command::new("kill") + .arg("-TERM") + .arg(pid.to_string()) + .status() + .await + .map_err(|e| e.to_string())?; + #[cfg(not(unix))] + { + let _ = pid; + child.start_kill().map_err(|e| e.to_string())?; + } + } + tokio::time::timeout(Duration::from_secs(30), child.wait()) + .await + .map_err(|_| "Pumas cancelled startup drain timed out")? + .map_err(|e| e.to_string())?; + Ok(()) +} + +impl Drop for PumasRuntime { + fn drop(&mut self) { + if self + .owner + .try_lock() + .is_ok_and(|owner| owner.child.is_none() && owner.stdout.is_none()) + { + return; + } + let owner = self.owner.clone(); + let description = self.description.clone(); + if let Ok(handle) = tokio::runtime::Handle::try_current() { + handle.spawn(async move { + if let Err(error) = shutdown_owner(&owner, &description).await { + tracing::error!(%error,"Owned Pumas drop cleanup failed"); + } + }); + } + } +} + +#[cfg(all(test, target_os = "linux"))] +mod tests { + use super::*; + use crate::pumas_inference::tests::{Fixture, description}; + use tokio::io::AsyncWriteExt; + + #[tokio::test] + async fn borrowed_shutdown_has_no_owner_effect_and_owned_shutdown_drains_only_its_child() { + let fixture = Fixture::start().await; + let description: HttpServiceDescription = + serde_json::from_value(description(&fixture.url)).unwrap(); + let borrowed = PumasRuntime { + description: description.clone(), + owner: Arc::new(Mutex::new(RuntimeOwner::default())), + }; + borrowed.shutdown().await.unwrap(); + assert!(fixture.requests.lock().is_empty()); + let mut child = Command::new("sh") + .args(["-c", "read line"]) + .stdin(Stdio::piped()) + .spawn() + .unwrap(); + let mut stdin = child.stdin.take().unwrap(); + let requests = fixture.requests.clone(); + let observer = tokio::spawn(async move { + loop { + if requests + .lock() + .iter() + .any(|(_, body)| body["method"] == "shutdown") + { + stdin.write_all(b"done\n").await.unwrap(); + return; + } + tokio::task::yield_now().await; + } + }); + let owned = PumasRuntime { + description, + owner: Arc::new(Mutex::new(RuntimeOwner { + child: Some(child), + service_owned: true, + ..RuntimeOwner::default() + })), + }; + owned.shutdown().await.unwrap(); + observer.await.unwrap(); + assert!(owned.owner.lock().await.child.is_none()); + owned.shutdown().await.unwrap(); + assert_eq!( + fixture + .requests + .lock() + .iter() + .filter(|(_, body)| body["method"] == "shutdown") + .count(), + 1 + ); + } + + #[tokio::test] + async fn cancelled_shutdown_retains_stdout_completion_for_the_next_waiter() { + let fixture = Fixture::start().await; + let (complete, receipt) = tokio::sync::oneshot::channel(); + let runtime = PumasRuntime { + description: serde_json::from_value(description(&fixture.url)).unwrap(), + owner: Arc::new(Mutex::new(RuntimeOwner { + stdout: Some(tokio::spawn(async move { + receipt.await.expect("output completion released"); + Ok(7) + })), + ..RuntimeOwner::default() + })), + }; + assert!( + tokio::time::timeout(Duration::from_millis(20), runtime.shutdown()) + .await + .is_err() + ); + assert!( + runtime.owner.lock().await.stdout.is_some(), + "cancelled waiter discarded stdout ownership" + ); + complete.send(()).unwrap(); + runtime.shutdown().await.unwrap(); + assert!(runtime.owner.lock().await.stdout.is_none()); + assert!(fixture.requests.lock().is_empty()); + } + + #[tokio::test] + async fn stdout_read_error_is_observed_and_remains_the_terminal_shutdown_result() { + struct FailedOutput; + impl tokio::io::AsyncRead for FailedOutput { + fn poll_read( + self: std::pin::Pin<&mut Self>, + _context: &mut std::task::Context<'_>, + _buffer: &mut tokio::io::ReadBuf<'_>, + ) -> std::task::Poll> { + std::task::Poll::Ready(Err(std::io::Error::other("controlled stdout read failure"))) + } + } + let fixture = Fixture::start().await; + let runtime = PumasRuntime { + description: serde_json::from_value(description(&fixture.url)).unwrap(), + owner: Arc::new(Mutex::new(RuntimeOwner { + stdout: Some(tokio::spawn(async { + tokio::io::copy(&mut FailedOutput, &mut tokio::io::sink()).await + })), + ..RuntimeOwner::default() + })), + }; + let error = runtime.shutdown().await.unwrap_err(); + assert!( + error.contains("Pumas stdout drain failed: controlled stdout read failure"), + "{error}" + ); + assert!(runtime.owner.lock().await.stdout.is_none()); + assert_eq!(runtime.shutdown().await.unwrap_err(), error); + assert!(fixture.requests.lock().is_empty()); + } + + #[tokio::test] + async fn stdout_task_panic_is_observed_and_not_reported_as_clean_shutdown() { + let fixture = Fixture::start().await; + let runtime = PumasRuntime { + description: serde_json::from_value(description(&fixture.url)).unwrap(), + owner: Arc::new(Mutex::new(RuntimeOwner { + stdout: Some(tokio::spawn(async { + panic!("controlled stdout task panic") + })), + ..RuntimeOwner::default() + })), + }; + let error = runtime.shutdown().await.unwrap_err(); + assert!( + error.contains("Pumas stdout drain task failed") + && error.contains("controlled stdout task panic"), + "{error}" + ); + assert_eq!(runtime.shutdown().await.unwrap_err(), error); + assert!(runtime.owner.lock().await.stdout.is_none()); + } + + #[tokio::test] + async fn borrowed_bootstrap_timeout_reaps_only_the_spawned_helper() { + use std::os::unix::fs::PermissionsExt; + let fixture = Fixture::start().await; + let root = std::env::temp_dir().join(format!( + "eidetic-borrowed-bootstrap-{}", + uuid::Uuid::new_v4() + )); + std::fs::create_dir_all(&root).unwrap(); + let binary = root.join("fixture-pumas"); + let ack = json!({"bootstrap_schema_version":1,"ownership":"borrowed","description":description(&fixture.url)}); + std::fs::write(&binary, format!("#!/bin/sh\nprintf '%s' \"$$\" > '{}/pid'\nprintf '%s\\n' 'PUMAS_LOCAL_ACCESS={ack}'\nexec sleep 300\n", root.display())).unwrap(); + std::fs::set_permissions(&binary, std::fs::Permissions::from_mode(0o700)).unwrap(); + let result = PumasRuntime::start(&root, &binary).await; + assert!( + matches!(&result, Err(error) if error.contains("Borrowed Pumas bootstrap did not exit")), + "unexpected bootstrap result" + ); + let pid = std::fs::read_to_string(root.join("pid")).unwrap(); + let live_path = format!("/proc/{}", pid.trim()); + let still_alive = Path::new(&live_path).exists(); + // Preserve the original failure without leaving the deliberately hung + // fixture behind when this regression runs against the old code. + if still_alive { + Command::new("kill") + .args(["-KILL", pid.trim()]) + .status() + .await + .unwrap(); + tokio::time::timeout(Duration::from_secs(5), async { + while Path::new(&live_path).exists() { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + } + assert!( + fixture.requests.lock().is_empty(), + "borrowed service received an RPC" + ); + PumasClient::connect(&fixture.url).await.unwrap(); + std::fs::remove_dir_all(root).unwrap(); + assert!( + !still_alive, + "timed-out borrowed bootstrap helper lost its exit observer" + ); + } + + #[tokio::test] + async fn cancelled_admitted_startup_still_observes_ack_and_drains_owned_child() { + use std::os::unix::fs::PermissionsExt; + let fixture = Fixture::start().await; + let root = + std::env::temp_dir().join(format!("eidetic-pumas-bootstrap-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&root).unwrap(); + let binary = root.join("fixture-pumas"); + let ack = json!({"bootstrap_schema_version":1,"ownership":"owned","description":description(&fixture.url)}); + // Delay acknowledgement to deterministically cancel the awaiting caller. + // The fixture process exits only after the exact fenced shutdown request. + std::fs::write(&binary,format!("#!/bin/sh\nprintf '%s' \"$$\" > '{}/started'\nsleep 0.1\nprintf '%s\\n' 'PUMAS_LOCAL_ACCESS={ack}'\nwhile [ ! -e '{}/exit' ]; do sleep 0.01; done\n",root.display(),root.display())).unwrap(); + std::fs::set_permissions(&binary, std::fs::Permissions::from_mode(0o700)).unwrap(); + let requests = fixture.requests.clone(); + let exit_root = root.clone(); + let observer = tokio::spawn(async move { + tokio::time::timeout(Duration::from_secs(10), async { + loop { + if requests + .lock() + .iter() + .any(|(_, body)| body["method"] == "shutdown") + { + std::fs::write(exit_root.join("exit"), "").unwrap(); + return; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + }); + let task_root = root.clone(); + let task = + tokio::spawn(async move { PumasRuntime::attach(&task_root, Some(&binary)).await }); + tokio::time::timeout(Duration::from_secs(10), async { + while !root.join("started").exists() { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + task.abort(); + assert!(matches!(task.await, Err(error) if error.is_cancelled())); + observer.await.unwrap(); + // Observe actual child reaping before removing the signal files or ending + // the test runtime; a shutdown request alone is not cessation evidence. + let pid = std::fs::read_to_string(root.join("started")) + .unwrap() + .parse::() + .unwrap(); + tokio::time::timeout(Duration::from_secs(10), async { + while Path::new(&format!("/proc/{pid}")).exists() { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + std::fs::remove_dir_all(root).unwrap(); + } +} diff --git a/crates/server/src/reference_retrieval_tests.rs b/crates/server/src/reference_retrieval_tests.rs index aaf70817..c7f60b67 100644 --- a/crates/server/src/reference_retrieval_tests.rs +++ b/crates/server/src/reference_retrieval_tests.rs @@ -1,7 +1,7 @@ use super::{attach_rag_context, attach_rag_embedding}; use crate::embeddings::{Embedding, EmbeddingClient}; use crate::project_service::replace_active_project; -use crate::state::{AppState, constants}; +use crate::state::AppState; use eidetic_core::ai::backend::{GenerateRequest, RagChunk}; use eidetic_core::ai::prompt::build_generate_request; use eidetic_core::reference::{ReferenceDocument, ReferenceType, chunk_document}; @@ -11,8 +11,7 @@ use std::sync::Arc; fn query(state: &AppState) -> Embedding { Embedding::new( - EmbeddingClient::new(&state.ai_config.lock().base_url, constants::EMBEDDING_MODEL) - .identity(), + EmbeddingClient::new(&state.ai_config.lock().embedding.base_url, "model").identity(), vec![1.0], ) .unwrap() @@ -44,6 +43,13 @@ fn request_with_old_context(project: &Project) -> GenerateRequest { async fn fixture() -> (AppState, Project, PathBuf) { let state = AppState::new().await; + state.ai_config.lock().embedding = crate::embeddings::EmbeddingConfig { + provider: crate::embeddings::EmbeddingProvider::OpenAiCompatible, + base_url: "http://localhost:18080/v1".into(), + model: "model".into(), + revision: "fixture".into(), + ..Default::default() + }; let source = ReferenceDocument::new("Notes", "Original", ReferenceType::StyleGuide); let mut project = Template::MultiCam.build_project("Retrieval"); project.references.push(source.clone()); @@ -98,7 +104,7 @@ async fn changed_config_path_or_missing_project_clears_preexisting_context() { let scope = state.vector_store.lock().scope(); let completed_query = query(&state); let config = state.ai_config.lock().clone(); - state.ai_config.lock().base_url = "http://localhost:18081/v1".into(); + state.ai_config.lock().embedding.base_url = "http://localhost:18081/v1".into(); let mut request = request_with_old_context(&project); attach_rag_embedding(&state, &path, scope, &completed_query, &mut request); assert!(request.rag_context.is_empty()); @@ -118,3 +124,82 @@ async fn changed_config_path_or_missing_project_clears_preexisting_context() { assert!(request.rag_context.is_empty()); state.shutdown_tasks(); } + +async fn late_query_preserves_new_index_status(reopen: bool) { + use crate::pumas_inference::tests::{Fixture, ready}; + let producer = Fixture::start().await; + let (state, project, path) = fixture().await; + state.ai_config.lock().embedding = producer.embedding_config(); + crate::reference_service::schedule_reindex(&state); + ready(&state).await; + let origin = state.vector_store.lock().scope(); + let config = state.ai_config.lock().embedding.clone(); + let delayed_query = EmbeddingClient::from_config(&config) + .await + .unwrap() + .embed("Original query") + .await + .unwrap(); + + // Pumas load revision can change while configuration stays exactly the same. + // Rebuild with a new revision AND dimensions before delivering the old query. + producer.change_embedding_representation(2, 4); + if reopen { + replace_active_project(&state, project.clone(), path.clone()); + } else { + crate::reference_service::schedule_reindex(&state); + } + ready(&state).await; + let current = state.vector_store.lock().scope(); + assert_ne!(origin, current); + let before = crate::reference_service::index_status(&state).unwrap(); + assert_eq!(before.documents[0].state, "ready"); + assert_eq!(before.documents[0].dimensions, Some(4)); + assert!( + before.documents[0] + .revision + .as_ref() + .unwrap() + .contains("weights-2") + ); + assert!(delayed_query.matches_config(&config)); + + let mut request = request_with_old_context(&project); + attach_rag_embedding(&state, &path, origin, &delayed_query, &mut request); + assert!(request.rag_context.is_empty()); + let after = crate::reference_service::index_status(&state).unwrap(); + assert_eq!( + serde_json::to_value(after).unwrap(), + serde_json::to_value(before).unwrap() + ); + + let fresh_query = EmbeddingClient::from_config(&config) + .await + .unwrap() + .embed("Current query") + .await + .unwrap(); + attach_rag_embedding(&state, &path, current, &fresh_query, &mut request); + assert_eq!( + request.rag_context[0].content, + project.references[0].content + ); + assert_eq!( + crate::reference_service::index_status(&state) + .unwrap() + .documents[0] + .state, + "ready" + ); + state.shutdown_tasks_async().await; +} + +#[tokio::test] +async fn late_query_cannot_mark_reindexed_reference_stale() { + late_query_preserves_new_index_status(false).await; +} + +#[tokio::test] +async fn late_query_cannot_mark_same_path_reopened_reference_stale() { + late_query_preserves_new_index_status(true).await; +} diff --git a/crates/server/src/reference_service.rs b/crates/server/src/reference_service.rs index 9bf1084e..c90b08b0 100644 --- a/crates/server/src/reference_service.rs +++ b/crates/server/src/reference_service.rs @@ -5,7 +5,9 @@ use serde::{Deserialize, Serialize}; use uuid::Uuid; use crate::backend_error::BackendError; -use crate::embeddings::{Embedding, EmbeddingClient}; +#[cfg(test)] +use crate::embeddings::Embedding; +use crate::embeddings::EmbeddingClient; use crate::state::AppState; use crate::validation; use crate::vector_store::IndexTicket; @@ -45,11 +47,6 @@ pub fn upload_reference( request.content, parse_reference_type(&request.doc_type), ); - let chunks = chunk_document( - &doc, - crate::state::constants::REFERENCE_CHUNK_SIZE, - crate::state::constants::REFERENCE_CHUNK_OVERLAP, - ); let response = doc.clone(); let ticket = { @@ -66,42 +63,137 @@ pub fn upload_reference( }; state.trigger_save(); - let state_clone = state.clone(); + schedule_document(state, response.clone(), ticket); + + Ok(response) +} + +/// Derived index state is session-bound and never persisted as canonical content. +#[derive(Debug, Clone, Serialize, Default)] +pub struct ReferenceIndexStatus { + pub documents: Vec, +} +#[derive(Debug, Clone, Serialize)] +pub struct ReferenceDocumentIndexStatus { + pub document_id: Uuid, + pub name: String, + pub state: String, + pub indexed_chunks: usize, + pub dimensions: Option, + pub model: Option, + pub revision: Option, + pub error: Option, +} + +pub fn index_status(state: &AppState) -> Result { + let project = state.project.lock(); + let project = project.as_ref().ok_or_else(BackendError::no_project)?; + let store = state.vector_store.lock(); + Ok(ReferenceIndexStatus { + documents: project + .references + .iter() + .map(|doc| store.document_status(doc)) + .collect(), + }) +} + +pub fn schedule_reindex(state: &AppState) { + rebuild_references(state, true); +} + +pub(crate) fn schedule_reopened_index(state: &AppState) { + rebuild_references(state, false); +} + +fn rebuild_references(state: &AppState, reset: bool) { + let documents = { + let project = state.project.lock(); + let Some(project) = project.as_ref() else { + return; + }; + let mut store = state.vector_store.lock(); + if reset { + store.reset(); + } + project + .references + .iter() + .map(|doc| (doc.clone(), store.begin_document(Arc::new(doc.clone())))) + .collect::>() + }; + for (doc, ticket) in documents { + schedule_document(state, doc, ticket); + } +} + +fn schedule_document(state: &AppState, doc: ReferenceDocument, ticket: IndexTicket) { + let state = state.clone(); + let config = state.ai_config.lock().embedding.clone(); state + .clone() .task_supervisor .spawn("reference-embedding", async move { - let config = state_clone.ai_config.lock().clone(); - let client = - EmbeddingClient::new(&config.base_url, crate::state::constants::EMBEDDING_MODEL); - - let (mut indexed, mut failed, mut discarded) = (0, 0, 0); - for chunk in chunks { - match client.embed(&chunk.content).await { - Ok(embedding) => { - let inserted = publish_embedding(&state_clone, &ticket, chunk, embedding); - if inserted { - indexed += 1; - } else { - discarded += 1; - } - } - Err(error) => { - failed += 1; - tracing::warn!("Failed to embed chunk: {error}"); - } + let result = index_document(&state, &doc, &ticket, &config).await; + if let Err(error) = result { + tracing::warn!(document=%doc.name, %error, "Reference indexing failed"); + let project = state.project.lock(); + if let Some(project) = project.as_ref() { + state.vector_store.lock().finish_document( + &ticket, + &project.references, + Some(error), + ); } } - tracing::info!( - indexed, - failed, - discarded, - "Reference embedding task finished" - ); }); +} - Ok(response) +async fn index_document( + state: &AppState, + doc: &ReferenceDocument, + ticket: &IndexTicket, + config: &crate::embeddings::EmbeddingConfig, +) -> Result<(), String> { + let client = EmbeddingClient::from_config(config).await?; + let chunks = chunk_document( + doc, + crate::state::constants::REFERENCE_CHUNK_SIZE, + crate::state::constants::REFERENCE_CHUNK_OVERLAP, + ); + let mut completed = Vec::new(); + let mut dimensions = None; + for chunk in chunks { + // Obsolete tasks stop before the next expensive model operation. + if state.ai_config.lock().embedding != *config + || !state.vector_store.lock().ticket_current(ticket) + { + return Ok(()); + } + let embedding = client.embed(&chunk.content).await?; + if dimensions.is_some_and(|value| value != embedding.values.len()) { + return Err("Embedding dimensions changed during reference indexing".into()); + } + dimensions = Some(embedding.values.len()); + completed.push((chunk, embedding)); + } + // Publish the entire document only after all real operations succeeded. + let project = state.project.lock(); + let Some(project) = project.as_ref() else { + return Ok(()); + }; + if state.ai_config.lock().embedding != *config { + return Ok(()); + } + let mut store = state.vector_store.lock(); + for (chunk, embedding) in completed { + store.insert(ticket, &project.references, chunk, embedding); + } + store.finish_document(ticket, &project.references, None); + Ok(()) } +#[cfg(test)] fn publish_embedding( state: &AppState, ticket: &IndexTicket, @@ -192,7 +284,9 @@ mod tests { #[tokio::test] async fn completed_embedding_cannot_resurrect_a_deleted_reference() { use super::{delete_reference, publish_embedding}; - use crate::embeddings::{Embedding, EmbeddingClient}; + #[cfg(test)] + use crate::embeddings::Embedding; + use crate::embeddings::EmbeddingClient; use eidetic_core::reference::{ReferenceDocument, chunk_document}; use std::sync::Arc; diff --git a/crates/server/src/state.rs b/crates/server/src/state.rs index 5fd13fa9..bdd2d7fe 100644 --- a/crates/server/src/state.rs +++ b/crates/server/src/state.rs @@ -29,8 +29,6 @@ pub mod constants { pub const REFERENCE_CHUNK_SIZE: usize = 500; /// Reference document chunk overlap in characters. pub const REFERENCE_CHUNK_OVERLAP: usize = 50; - /// Embedding model name. - pub const EMBEDDING_MODEL: &str = "nomic-embed-text"; /// Number of top RAG results to include. pub const RAG_TOP_K: usize = 3; } @@ -62,6 +60,11 @@ pub enum ServerEvent { node_id: uuid::Uuid, error: String, }, + ReferenceRetrieval { + node_id: uuid::Uuid, + sources: Vec, + error: Option, + }, BibleChanged, ScriptChanged, SemanticProposalsChanged, @@ -81,6 +84,7 @@ pub enum ServerEvent { #[serde(rename_all = "snake_case")] pub enum BackendType { LlamaCpp, + Pumas, OpenRouter, } @@ -93,6 +97,10 @@ pub struct AiConfig { pub max_tokens: usize, pub base_url: String, pub api_key: Option, + #[serde(default)] + pub pumas_profile: String, + #[serde(default)] + pub embedding: crate::embeddings::EmbeddingConfig, } impl Default for AiConfig { @@ -104,6 +112,8 @@ impl Default for AiConfig { max_tokens: constants::DEFAULT_MAX_TOKENS, base_url: constants::DEFAULT_LLAMACPP_URL.into(), api_key: None, + pumas_profile: String::new(), + embedding: crate::embeddings::EmbeddingConfig::default(), } } } @@ -137,6 +147,8 @@ pub struct AppState { save_tx: tokio::sync::mpsc::Sender<()>, /// Model library from Pumas for listing available local models. pub model_library: Option>, + pub(crate) pumas_runtime: Option>, + pub(crate) pumas_startup_error: Option, /// Backend-owned transient timeline selection projected to renderers and UI. pub selected_timeline_node_id: Arc>>, /// Backend-owned transient playhead position projected across timeline surfaces. @@ -179,6 +191,29 @@ impl AppState { // Initialize the Pumas model library (optional — best-effort). let model_library = Self::init_model_library().await; + let (pumas_runtime, pumas_startup_error) = + match crate::pumas_runtime::PumasRuntime::from_environment().await { + Ok(runtime) => (runtime, None), + Err(error) => { + tracing::error!(%error,"Pumas connection initialization failed"); + (None, Some(error)) + } + }; + let mut ai_config = AiConfig::default(); + if pumas_startup_error.is_some() { + // An explicitly selected Pumas root must fail visibly, rather than + // silently reverting to another local provider. + ai_config.backend_type = BackendType::Pumas; + ai_config.base_url.clear(); + ai_config.model.clear(); + } + if let Some(runtime) = &pumas_runtime { + ai_config.backend_type = BackendType::Pumas; + ai_config.base_url = runtime.description.endpoint.as_str().to_owned(); + ai_config.model.clear(); + ai_config.embedding.base_url = ai_config.base_url.clone(); + } + Self { project, project_session_gate, @@ -186,7 +221,7 @@ impl AppState { events_tx, doc_tx, doc_update_tx, - ai_config: Arc::new(Mutex::new(AiConfig::default())), + ai_config: Arc::new(Mutex::new(ai_config)), generating: Arc::new(Mutex::new(HashSet::new())), #[cfg(test)] project_path, @@ -194,6 +229,8 @@ impl AppState { vector_store: Arc::new(Mutex::new(VectorStore::new())), save_tx, model_library, + pumas_runtime, + pumas_startup_error, selected_timeline_node_id: Arc::new(Mutex::new(None)), timeline_playhead_ms: Arc::new(Mutex::new(0)), task_supervisor, @@ -220,6 +257,11 @@ impl AppState { pub async fn shutdown_tasks_async(&self) { self.task_supervisor.shutdown_all().await; + if let Some(runtime) = &self.pumas_runtime { + if let Err(error) = runtime.shutdown().await { + tracing::error!(%error,"Owned Pumas shutdown failed"); + } + } } /// Initialize the Pumas model library from env or sibling directory. diff --git a/crates/server/src/vector_store.rs b/crates/server/src/vector_store.rs index 28677bb8..2e909e95 100644 --- a/crates/server/src/vector_store.rs +++ b/crates/server/src/vector_store.rs @@ -18,6 +18,9 @@ pub(crate) struct IndexTicket { struct IndexedDocument { revision: Uuid, source: Arc, + error: Option, + complete: bool, + stale: bool, } /// Disposable retrieval index. Canonical references remain in the project. @@ -57,11 +60,95 @@ impl VectorStore { IndexedDocument { revision: ticket.revision, source, + error: None, + complete: false, + stale: false, }, ); ticket } + pub(crate) fn ticket_current(&self, ticket: &IndexTicket) -> bool { + ticket.scope == self.scope + && self + .documents + .get(&ticket.document_id) + .is_some_and(|doc| doc.revision == ticket.revision) + } + + pub(crate) fn finish_document( + &mut self, + ticket: &IndexTicket, + sources: &[ReferenceDocument], + error: Option, + ) { + if !self.ticket_current(ticket) { + return; + } + let Some(doc) = self.documents.get_mut(&ticket.document_id) else { + return; + }; + if !source_is_current(&doc.source, sources) { + return; + } + doc.complete = true; + doc.error = error; + if doc.error.is_some() { + self.entries + .retain(|_, (chunk, _)| chunk.document_id != ticket.document_id); + } + } + + pub(crate) fn mark_incompatible(&mut self, query: &Embedding) { + for (chunk, embedding) in self.entries.values() { + if (embedding.identity != query.identity + || embedding.values.len() != query.values.len()) + && let Some(doc) = self.documents.get_mut(&chunk.document_id) + { + doc.stale = true; + doc.error = Some( + "Embedding model revision or dimensions changed; Reindex references".into(), + ); + } + } + } + + pub(crate) fn document_status( + &self, + source: &ReferenceDocument, + ) -> crate::reference_service::ReferenceDocumentIndexStatus { + let document = self + .documents + .get(&source.id) + .filter(|doc| source_is_current(&doc.source, std::slice::from_ref(source))); + let entries: Vec<_> = self + .entries + .values() + .filter(|(chunk, _)| chunk.document_id == source.id) + .collect(); + crate::reference_service::ReferenceDocumentIndexStatus { + document_id: source.id.0, + name: source.name.clone(), + state: match document { + None => "needs_reindex", + Some(doc) if doc.stale => "stale", + Some(doc) if doc.error.is_some() => "failed", + Some(doc) if doc.complete => "ready", + _ => "indexing", + } + .into(), + indexed_chunks: entries.len(), + dimensions: entries.first().map(|(_, embedding)| embedding.values.len()), + model: entries + .first() + .map(|(_, embedding)| embedding.model().to_owned()), + revision: entries + .first() + .map(|(_, embedding)| embedding.revision_label()), + error: document.and_then(|doc| doc.error.clone()), + } + } + /// Caller holds the project guard through publication, closing the deletion race. pub(crate) fn insert( &mut self, diff --git a/docs/dependency-governance.md b/docs/dependency-governance.md index 2bdcfbb3..9ff5507c 100644 --- a/docs/dependency-governance.md +++ b/docs/dependency-governance.md @@ -50,7 +50,7 @@ EIDETIC_FETCH_PUMAS=1 scripts/prepare-pumas-dependency.sh Default source contract: - Repository: `https://github.com/MrScripty/Pumas-Library.git` -- Revision: `a94fd92021f27fdeedb6e2de6e01c41c250ef576` +- Revision: `ad31e391dcd94c204b3fcd748cc98d27b3281e65` - Required package: `pumas-library` - Required local path: `../../ai-systems/Pumas-Library/rust/crates/pumas-core` @@ -77,6 +77,30 @@ private mirror or an already provisioned checkout. accepting untrusted cookie metadata. See the exact finding and follow-up in [PR29's dependency audit](reports/pr29-dependency-audit.md). +The 2026-10-14 PR29 audit repair maps all 14 high and 3 critical baseline rows to +their actual consumers and upstream advisories. It uses same-family patches, +SvelteKit 2.70.3 for a reachable development-server header issue, and a coordinated +Vitest/coverage 4.1.11 update that removes vulnerable Tinypool 1.x. Node remains +24.12.0 and Vite remains on 6.x; no forced transitive major overrides are used. +The CI frontend job now runs the existing high/critical audit policy and the +static build in addition to lint, typechecking and tests. Runtime-capable Svelte +dependencies are reviewed even though this app's manifest labels them as dev +dependencies. Audit counts include inherited parent rows; `fixAvailable: false` +does not prove that upstream has no patched release. + +The current local-inference consumer uses Pumas 0.8.0-rc.1's HTTP advertisement / +admission-fence schema 1, typed model-operation contract 1 and serving RPC DTOs. +The `ad31e391` pin is intentional when integrating main `59793c19`: main's +`a94fd920` setup revision is Pumas 0.7.0 and does not expose the discovery module +or typed `/v1/model-operations` contract required by this consumer. The older +`8444b50d` documentation revision is also superseded. Keep the preparation script, +this source contract and the lockfile on the same tested producer revision; +native UI acceptance from PR30 does not qualify model inference. +Build `pumas-rpc` with its default `inference-plugins` feature; an inference-disabled +binary is explicitly refused. The path dependency and `Cargo.lock` package version +match the setup pin. See [local acceptance](pumas-local-inference.md) for runtime +setup; no application model is downloaded during build or test. + The 2026-10-14 authoring integration carries the exact reviewed frontend repair from PR29 commit `be03aa1f2b7e6e19da632fac3b2b7eb89ec12c73`, without importing its Pumas/model lifecycle changes. The security-relevant CI, package manifest, diff --git a/docs/pumas-local-inference.md b/docs/pumas-local-inference.md new file mode 100644 index 00000000..b52ead6f --- /dev/null +++ b/docs/pumas-local-inference.md @@ -0,0 +1,150 @@ +# Local inference through Pumas + +Eidetic pins Pumas `ad31e391dcd94c204b3fcd748cc98d27b3281e65` +(0.8.0-rc.1). Chat and reference embeddings are independent selections. Production +requests use real Pumas providers through its serving lifecycle and typed HTTP +model-operation contract; no fake embeddings, cloud substitution or model +download is used as a fallback. + +This exact pin is retained after integrating main `59793c19`; its older Pumas +setup pin predates discovery and typed model operations. See +[dependency governance](dependency-governance.md) for the compatibility decision. +PR30's native UI acceptance and green source CI do not satisfy the trained-model +acceptance below. PR29 remains draft until parent review and that runtime gate +are explicitly dispositioned. + +## Prepare + +1. Run `scripts/prepare-pumas-dependency.sh`. On a new machine use + `EIDETIC_FETCH_PUMAS=1`; an existing mismatched checkout is refused, never reset. + For active producer development set `EIDETIC_PUMAS_LIBRARY_REF` to the exact + tested checkout, update/review the lockfile and repeat the acceptance below. +2. Build the same Pumas checkout's `pumas-rpc` with default features: + `cargo build --manifest-path ../../ai-systems/Pumas-Library/rust/Cargo.toml -p pumas-rpc`. + `inference-plugins` is required. Supply the native ONNX Runtime library as + documented by Pumas (including `ORT_DYLIB_PATH` where required). Eidetic never + enables ONNX Runtime's download-binaries feature. +3. Use Pumas to install/configure the runtime profiles and the local models you + already have. Chat requires a chat-capable provider/model, typically llama.cpp + or Ollama. Embeddings require a text-embedding-capable model/provider, including + a complete real ONNX package/tokenizer/config supported by Pumas. The provider + rejects incompatible model/profile combinations. No gated model or credentials + are needed for the consumer tests. + +## Connect and select + +Run Pumas separately, for example: + +```bash +/path/to/pumas-rpc --launcher-root /absolute/pumas/root --port 8080 +``` + +Enter its numeric loopback root URL (`http://127.0.0.1:8080`, without `/v1`) in +Eidetic's AI panel. Select **Pumas**, list installed models/profiles, choose the +chat model and its profile, then **Load**. Choose **Pumas local inference** in +Reference embeddings, enter its independent endpoint, list and select the +embedding model and appropriate profile, and **Load**. Save configuration. +Capability checks require the exact loaded model/profile; merely listing a +model or reaching the HTTP service is insufficient. Load errors, unsupported +capabilities and inference-disabled builds are displayed. + +Alternatively set `EIDETIC_PUMAS_ROOT=/absolute/pumas/root` before launching +Eidetic to borrow an authenticated advertised local service. On Linux, also set +`EIDETIC_PUMAS_RPC_BIN=/absolute/path/to/pumas-rpc` to allow the selected-root +`--attach-or-start-local-http` bootstrap. This starts only the Pumas control plane; +model/runtime selection stays explicit. The bootstrap's ownership acknowledgement +and exact instance/service generations govern cleanup. Cancelled startup continues +its admitted construction; the controlled success fixture observes owned-service +drain. Complete startup error/panic observation after cancellation and last-Drop +cleanup still require the retained composition owner described in +[the lifecycle acceptance gate](reports/pr29-lifecycle-ownership.md). +On explicit application shutdown an owned service receives a generation-fenced +shutdown, and its child exit and stdout-drain result are observed. +A borrowed Pumas service receives no implicit unload/shutdown. Explicit bootstrap +failures appear in AI status and keep Pumas selected; they do not revert to a +direct llama.cpp endpoint. **Unload** is an +explicit operator action on the selected model/profile, including borrowed models. +Provider unload errors remain failures even if Pumas reports RPC `success: true`; +an error-free `unloaded: false` is displayed as already not served. +For non-Linux systems start Pumas separately and borrow its endpoint. + +Existing direct llama.cpp and OpenRouter chat remain supported. An independently +configured OpenAI-compatible embedding provider uses its own `/v1` base URL, +model, operator revision and bearer API key. Its revision is operator supplied, +not a producer weight attestation. Reference embeddings default to **Disabled**; +this cannot accidentally send reference text to a chat provider. + +Configuration remains application-session state. Reopening the AI panel restores +that state's selections; after a full restart select/save providers again. Saved +reference text is SQLite project content and is never discarded by index failure. + +## Local acceptance with real models + +1. Load both selected models. Confirm chat status is connected; choose a notes + node and generate to exercise the actual Pumas chat provider. Also try a + structured child-plan call; typed v1 has no JSON-mode option, so invalid JSON + becomes a visible provider error rather than a successful structured result. +2. Upload a short reference containing a unique detail (for example a location + name absent from the project). Wait for **ready**, nonzero chunk count and + the model/revision/dimension display. If it fails, inspect the displayed error + and Pumas's runtime diagnostics; uploading is not proof of embedding success. +3. Put related notes on a generation target and generate. Confirm the notification + names the reference source. Open **Last Generation Prompt** in the Notes panel and + confirm the retrieved detail appears in the exact prompt sent to Pumas. Inspect + the actual generated text independently; prompt inclusion demonstrates retrieval + contribution even if the model chooses not to repeat that detail. +4. Save and reopen the SQLite project. References remain listed and are automatically + reindexed with the current embedding selection. **Reindex references** also + explicitly retries saved text after startup/configuration or a failed load. +5. Change the embedding model/profile and save. The previous index is invalidated + before new publication. Reload the same Pumas model at a new revision: retrieval + refuses the old index and shows **stale**; reindex it before expecting retrieval. + Serving-state changes, including unrelated chat loads/unloads, conservatively + invalidate existing embeddings; load both models before indexing and Reindex + after any serving change. A changed vector dimension also excludes old + candidates. Delete a reference while indexing and reopen during indexing; + obsolete results cannot publish. +6. Unload embeddings and generate: reference retrieval must report unavailable; + ordinary chat generation may continue, and it must not claim references were + consumed. Unload chat and generate: generation must fail visibly. Try an + unsupported model/profile and an inference-disabled Pumas binary. +7. Close Eidetic while using a borrowed Pumas service: it and its loaded models + remain available. With Linux owning bootstrap, close Eidetic and confirm the + exact owned Pumas service and its managed providers/native sessions drain. + A borrowed/external provider's process ownership remains governed by Pumas. + +## Index identity and limits + +The disposable index binds exact document content/name/type/id and source ticket, +project epoch, embedding provider/endpoint/model/profile, Pumas instance/service +generations, the monotonic serving-state cursor, optional load timestamp and actual +library metadata (including upstream revision if available), plus matching vector +dimensions. All chunks of a +document publish together after successful embedding. Deleted sources, project +replacement and embedding-selection changes invalidate outstanding tickets. +Generation rechecks scope/config and current source at attachment. +The captured scope is checked before changing index compatibility/status, so a +late query from an earlier index cannot mark a rebuilt reference stale. + +Pumas typed v1 returns vectors but no immutable package-weight revision in each +operation response. Its serving adapters omit the optional load timestamp. +Eidetic observes provenance and the service-scoped serving cursor before and after +embedding. Reloads, including same-model reloads with unchanged metadata, advance +that cursor. Unrelated serving changes also require reindexing; this conservative +invalidation avoids inventing per-model load identity. It prevents mixing observed +reloads/revisions; it cannot prove that weights were not mutated out of band without +metadata or a serving-cursor change, including a provider/package ABA outside +Pumas's serving lifecycle. Do not edit loaded model files in place; reload and +reindex. A stronger producer-issued immutable revision +would require a coordinated Pumas contract change; none is introduced here. + +## Automated evidence + +`cargo test -p eidetic-server --lib pumas` exercises the actual consumer HTTP/RPC +and SSE paths with explicitly controlled producer fixtures. It checks lifecycle, +errors, revision changes, upload/reopen/reindex, retrieved prompt inclusion, +ownership cleanup and cancellation. Existing vector/index/session and nonlocal +provider regressions remain active. Fixture vectors and screenplay text are +synthetic **test data**, never production fallbacks or real-model evidence. +Real ONNX/llama.cpp model acceptance was not executed in the cloud environment, +which contains no chat or embedding weights/native model runtime fixture. diff --git a/docs/reports/pr29-lifecycle-ownership.md b/docs/reports/pr29-lifecycle-ownership.md new file mode 100644 index 00000000..c0b2dd8c --- /dev/null +++ b/docs/reports/pr29-lifecycle-ownership.md @@ -0,0 +1,103 @@ +# PR29 lifecycle observation repair and remaining decision + +## Scope + +This follow-up starts at `b4af7e94a0eb7896e2118fb9cff564f39f66ec03`. +It changes only the existing private Pumas runtime owner and its regressions; +the public AppState/desktop/model contracts, producer pin, model selection and +inference behavior are unchanged. It is a partial lifecycle repair, not product +acceptance. The PR discussion owns the published repair SHA and exact-head CI. + +## Repaired within the existing owner + +- The child and stdout-drain task are retained in one runtime owner. Explicit + shutdown waits for child exit and joins stdout; output I/O errors and task + panics become terminal errors that repeated shutdown cannot erase. +- Cancelling a shutdown waiter leaves the stdout completion handle in the + owner. A subsequent explicit shutdown can observe it. A stdout-drain timeout + also retains its handle instead of reporting clean completion. +- A bootstrap that acknowledges borrowed ownership but fails to exit within + the existing ten-second budget has only its spawned helper terminated and + reaped. No unload/shutdown RPC or signal is sent to the borrowed service. + The output observer is joined before reporting the normal timeout outcome. + +The borrowed-helper regression first failed against the original source with +the helper still present in `/proc`. The test cleaned up that deliberate hang. +After repair it requires helper disappearance, an empty service RPC log and +continued external-service connectivity. Controlled stdout regressions cover +I/O failure, task panic, sticky terminal errors and cancelled-waiter handoff. +These are process/I/O fixture claims, not trained-model or native UI evidence. + +## Required composition-owner decision + +The admitted startup task and last-Drop fallback still lack an external owner +that can observe their completion, errors and panics after their caller is +gone. This repair does not relabel either path as complete. Exceptional helper +termination/wait errors or stdout timeouts during failed startup can still +reach that detached fallback and remain part of this gate. + +An owner stored only inside the cancellable `attach`/`AppState::new` future +disappears when construction is cancelled. An owner stored only in the last +`PumasRuntime` disappears during Drop. A self-retaining task, logging, +`catch_unwind`, `kill_on_drop` or another detached observer cannot establish +external terminal observation. The existing generic backend supervisor also +aborts and drops its handles on last Drop, so merely registering there is not +a complete solution. + +The smallest sound next design is a Pumas lifecycle owner retained by the +application composition root **before subprocess admission**. It must register +startup before spawn, retain startup/output/cleanup outcomes after caller +cancellation, receive last-Drop cleanup requests without spawning unowned +work, close admission during application exit and be explicitly drained by +the root. Incomplete cleanup retains custody and reports failure. The generic +backend supervisor need not be redesigned. + +This requires an explicit decision about the AppState construction boundary +and the desktop/smoke callers that retain and finally drain that owner. The +alternative is to narrow the cancellation/last-Drop promise, which changes the +existing contract. Neither choice is silently implemented here. No global +owner, alternate runtime, cleanup thread or public API change is introduced. + +## Synchronous shutdown execution contract + +Source inspection uses the locked Tauri 2.11.2 implementation: +`async_runtime.rs` creates a Tokio runtime by default and `block_on` delegates +to it; `Builder::on_window_event` registers a synchronous window-event handler. +Eidetic calls `block_on(AppState::shutdown_tasks_async())` from that handler on +Destroyed, after stopping its event bridges and renderer owners. This keeps the +explicit drain receipt before the callback returns, and blocks further event +dispatch while waiting. The owned Pumas drain uses Tokio HTTP/process I/O and +does not intentionally dispatch UI work back to that callback. + +This source inspection does not demonstrate deadlock, establish every possible +runtime override/callback context, or constitute native execution acceptance. +Moving final drain to a root retained outside the event loop (for example an +explicit post-event-loop exit phase) belongs to the composition decision above. +No event-loop/exit API is changed by this partial repair. + +## Retained acceptance gates + +- Decide and implement the retained composition owner, then directly test + cancelled startup success/error/panic and last-Drop fallback completion and + failure, including incomplete cleanup custody. +- Qualify the accepted desktop exit/callback execution contract in a supported + native environment. There is no demonstrated deadlock claim to waive. +- Run actual trained chat/embedding and native lifecycle acceptance from + [the operator procedure](../pumas-local-inference.md#local-acceptance-with-real-models). + Model packages and native development libraries remain absent locally. No + models, credentials or denied-access retries are part of this follow-up. + +Keep PR29 draft until these gates are independently dispositioned. Source +tests and hosted CI prove their named source/fixture claims only. + +## Local verification + +- Six targeted Pumas runtime tests pass, including the original cancellation + success fixture and the four new regressions. +- Locked core/server all-target tests pass: 129 core and 642 server tests, + with all example targets; strict all-target/all-feature Clippy passes. +- Rust formatting, staged whitespace and decision traceability pass. +- The canonical `./launcher.sh --test` stops at missing `glib-2.0` development + metadata during native compilation. The original output is retained; no + assertion, discovery filter or threshold is weakened. Hosted exact-head + native-workspace CI remains a separate gate recorded on the PR. diff --git a/docs/reports/pr29-merge-readiness.md b/docs/reports/pr29-merge-readiness.md new file mode 100644 index 00000000..7f1872ed --- /dev/null +++ b/docs/reports/pr29-merge-readiness.md @@ -0,0 +1,118 @@ +# PR29 integration and remaining acceptance + +## Cleanup integration on 2026-10-21 + +The disposition pass integrates main `6bef18778ad3a02148188b0b044d7a065acd28f8` +into PR29 head `be03aa1f2b7e6e19da632fac3b2b7eb89ec12c73`, preserving both +histories. Six documentation conflicts retain both authoring and local-inference +contracts; product code merges mechanically. The Pumas pin remains `ad31e391`. +The frontend dependency repair and its CI gates are already present on main and +are not unique remaining PR29 work. + +Independent GPT-6.1 Sol High review found no new confirmed P1/P2 in the inspected +consumer paths and approved the six documentation resolutions. The integration +passes `git diff --cached --check`, launcher/setup shell syntax checks, five +ONNX guard unit tests, seven authoring fixture regressions, Rust formatting and +decision traceability. Pinned Node 24.12.0 frontend installation, lint, formatting, +type checks, all 115 files / 646 tests and static build pass. A fresh dependency +audit passes the existing high/critical gate, retaining only the three tracked +low package rows for the single cookie advisory. Independent review against +the exact `ad31e391` producer confirms the consumed discovery, DTO, typed +operation/SSE, metadata and serving-cursor source contracts align. Core/server +Rust checks are in progress. Exact-head hosted CI and final Rust results must +be recorded in the PR before claiming the integration validated. The 2026-10-14 +counts below are historical evidence, not results for this composed tree. + +PR29 remains draft. Actual trained chat/embedding inference and native lifecycle +acceptance below remain unexecuted; no model acquisition, credentials, scientific +campaign, release or PR merge is performed as part of this source integration. The deferred +borrowed-bootstrap observer issue remains recorded below. The PR discussion owns +the final integration SHA, fresh validation and exact unresolved gates. + +## Scope and source identity + +PR29's existing `codex/pumas-local-inference` branch starts at +`b09f9c288b8feeb5da3f1643b085b0338c106dd2`. This integration merges main +`59793c1950916a067f6429db59bbc639fe3c4acf`, preserving both histories and PR30's +native UI changes. It does not release or merge PR29. The parent owns that +decision after reviewing the source and the separate runtime acceptance gate. + +Pumas remains pinned to `ad31e391dcd94c204b3fcd748cc98d27b3281e65` +(`0.8.0-rc.1`). Main's older `a94fd920` setup pin is `0.7.0` and lacks the +discovery and typed-operation APIs used here. The setup script, governance +document and lockfile agree. No producer source is changed. See +[dependency governance](../dependency-governance.md). + +## Confirmed defect and repair + +The pinned producer's ONNX, llama.cpp and Ollama serving adapters omit the +optional `loaded_at` timestamp. Eidetic previously required it, so a successfully +loaded real embedding model could never be used. The old fixture invented a +timestamp and concealed this incompatibility. A producer-shaped regression +first failed with `Pumas embedding load is missing revision timestamp`. + +The consumer now binds embedding identity to the producer's monotonic serving +cursor, instance/service generations, selected provider/model/profile and actual +metadata. The optional timestamp is retained if supplied. Existing observations +before and after inference remain required. Same-model unload/reload with +unchanged metadata invalidates old vectors, and a serving change during inference +fails without publishing any document chunks. An unrelated serving mutation also +requires reindexing; this conservative behavior is explicit in the operator guide. +No fabricated model revision or fixture output enters production. + +## Verification on 2026-10-14 + +The executor has Rust 1.92.0 and Node 24.12.0. Tests use the selected exact Pumas +checkout. Application data/config/cache were isolated under a temporary directory +because the executor's host home is read-only. + +| Check | Result and boundary | +| --- | --- | +| `cargo test --locked -p eidetic-core -p eidetic-server --all-targets` | 123 core and 615 server tests pass; all example test targets pass. | +| `cargo test --locked -p eidetic-server --lib pumas` | 16 controlled consumer/process fixture tests pass. | +| `cargo clippy --locked -p eidetic-core -p eidetic-server --all-targets --all-features -- -D warnings` | Pass. | +| Frontend lint, format, typecheck and `npm run test` | Pass; 608 tests in 107 files. | +| Locked all-feature Cargo metadata and ONNX no-download guard | Pass; five guard tests pass. | +| Authoring fixture regressions | Seven tests pass. | +| `cargo fmt --all -- --check` and `git diff --check` | Pass. | +| Independent GPT-6.1 Sol High source review | No remaining confirmed P1/P2 in the candidate repair. | +| `./launcher.sh --test` | Environment-blocked at missing `glib-2.0`; GTK/WebKit development libraries are absent. Full native workspace checks must run in hosted Linux/Windows CI on the pushed head. | +| `scripts/check-dependencies.sh` | Baseline at `1148fb0e` failed with 42 reported rows (1 low, 24 moderate, 14 high, 3 critical). The [dependency follow-up](pr29-dependency-audit.md) maps and repairs the high/critical findings, adds CI audit/static-build gates, and records the remaining cookie issue. Automatic `fixAvailable` flags were not reliable evidence of upstream fix availability. Rust audit/deny/machete tools remain absent and reported as local skips. | + +The PR record owns the exact pushed head and its hosted CI run. Passing source +checks do not resolve trained-model acceptance below. The linked dependency +follow-up owns the later lockfile, verification and remaining audit disposition. + +## Trained-model acceptance remains blocked + +There are no local chat or embedding weights. Ordinary unauthenticated HEAD +requests for public trained chat/embedding artifacts at Hugging Face fail at the +executor proxy with `CONNECT tunnel failed: 403 Forbidden`. Access was not +retried through another host or bypass. No trained inference was executed. +Controlled vectors and screenplay text are synthetic fixture data only. +PR30's native UI captures do not qualify this new model/reference lifecycle. + +Minimal operator step: on a machine with accessible local trained chat and +embedding packages, build this exact Pumas pin with default inference plugins, +load both selected models/profiles, and execute +[the local real-model acceptance procedure](../pumas-local-inference.md#local-acceptance-with-real-models). +Record actual generation, reference indexing and exact prompt contribution, +reopen/reindex, same-model reload refusal, and owned/borrowed cleanup. Keep PR29 +draft until the parent reviews and dispositions the gates; no denial bypass, +credentials or release action is part of this repair. + +## Deferred robustness observation + +Independent review also noted that a nonconforming bootstrap that acknowledges +borrowed ownership but then hangs can outlive the ten-second bootstrap wait; that +failure path drops its child observer. The pinned producer normally exits +immediately after borrowed acknowledgement. This lower-priority subprocess +robustness issue is deferred to a separately scoped lifecycle repair. It does not +grant authority to unload or shut down the borrowed Pumas service. + +That paragraph records the historical observation. The subsequent +[partial lifecycle repair](pr29-lifecycle-ownership.md) terminates/reaps the +spawned borrowed helper on its normal timeout path and observes stdout completion, +without touching the borrowed service. Complete failed-startup and last-Drop +cleanup observation still requires the composition-owner decision in that report; +the earlier deferred classification is superseded only for the repaired path. diff --git a/scripts/prepare-pumas-dependency.sh b/scripts/prepare-pumas-dependency.sh index ce3912cc..b8e3d93a 100755 --- a/scripts/prepare-pumas-dependency.sh +++ b/scripts/prepare-pumas-dependency.sh @@ -6,7 +6,7 @@ PROJECT_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" DEFAULT_PUMAS_ROOT="$(cd "$PROJECT_ROOT/../.." && pwd)/ai-systems/Pumas-Library" PUMAS_ROOT="${EIDETIC_PUMAS_LIBRARY_ROOT:-$DEFAULT_PUMAS_ROOT}" PUMAS_REPOSITORY="${EIDETIC_PUMAS_LIBRARY_REPOSITORY:-https://github.com/MrScripty/Pumas-Library.git}" -PUMAS_REF="${EIDETIC_PUMAS_LIBRARY_REF:-a94fd92021f27fdeedb6e2de6e01c41c250ef576}" +PUMAS_REF="${EIDETIC_PUMAS_LIBRARY_REF:-ad31e391dcd94c204b3fcd748cc98d27b3281e65}" PUMAS_PACKAGE_PATH="$PUMAS_ROOT/rust/crates/pumas-core" log() { diff --git a/src-tauri/src/README.md b/src-tauri/src/README.md index 1c8d84a6..3fea5fe5 100644 --- a/src-tauri/src/README.md +++ b/src-tauri/src/README.md @@ -103,6 +103,22 @@ The command registry includes `projection_bible_recall` for explicit selected Bible entity inspection; screenplay replacement remains behind existing review and acceptance commands. +Pumas desktop commands retain backend ownership: `pumas_catalog` projects installed +model/profile choices, `pumas_load_model` / `pumas_unload_model` consume Pumas's +actual serving DTOs, and `ai_config_get` returns current independent chat and +embedding selections. `reference_index_status` and `reference_reindex` expose +source-bound derived indexing without changing canonical reference storage. +Owned Pumas bootstrap cleanup runs through `AppState::shutdown_tasks_async` on +window shutdown. Borrowed services receive no implicit unload/shutdown request. + +Window destruction observes backend/Pumas shutdown with `block_on` before exit +can tear down the async runtime. A spawned fire-and-forget drain is insufficient +for an owned Pumas child; the exact child exit receipt must remain awaited. + +`pumas_unload_model` returns the actual typed `UnserveModelResponse` after backend +failure checks; the frontend preserves `unloaded: false` as already unserved +rather than claiming the runtime performed an unload. + ## Authoring command registration The invoke registry admits `command_timeline_node_arcs` and diff --git a/src-tauri/src/ai_commands.rs b/src-tauri/src/ai_commands.rs index 8e534f56..db0ce77f 100644 --- a/src-tauri/src/ai_commands.rs +++ b/src-tauri/src/ai_commands.rs @@ -67,3 +67,35 @@ pub async fn ai_generate_batch( .await .map_err(CommandError::from) } + +#[tauri::command] +pub async fn pumas_load_model( + endpoint: String, + request: eidetic_server::pumas_inference::ServeModelRequest, +) -> Result { + eidetic_server::pumas_inference::load_model(&endpoint, request) + .await + .map_err(|e| CommandError::from(eidetic_server::backend_error::BackendError::internal(e))) +} + +#[tauri::command] +pub async fn pumas_unload_model( + endpoint: String, + request: eidetic_server::pumas_inference::UnserveModelRequest, +) -> Result { + eidetic_server::pumas_inference::unload_model(&endpoint, request) + .await + .map_err(|e| CommandError::from(eidetic_server::backend_error::BackendError::internal(e))) +} + +#[tauri::command] +pub async fn pumas_catalog(endpoint: String) -> Result { + eidetic_server::pumas_inference::catalog(&endpoint) + .await + .map_err(|e| CommandError::from(eidetic_server::backend_error::BackendError::internal(e))) +} + +#[tauri::command] +pub fn ai_config_get(app: tauri::AppHandle) -> AiConfig { + app.state::().ai_config.lock().clone() +} diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index efe57127..68da5c5b 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -80,10 +80,10 @@ pub fn run() { { let _ = timeline_owner.stop(); } - let task_supervisor = window.state::().task_supervisor.clone(); - tauri::async_runtime::spawn(async move { - task_supervisor.shutdown_all().await; - }); + let app_state = window.state::().inner().clone(); + // Observe the owned Pumas drain before desktop exit can tear + // down the runtime that carries its shutdown receipt. + tauri::async_runtime::block_on(app_state.shutdown_tasks_async()); } }) .invoke_handler(tauri::generate_handler![ @@ -117,6 +117,12 @@ pub fn run() { timeline_renderer_commands::timeline_renderer_status, timeline_renderer_commands::timeline_renderer_close, reference_commands::reference_list, + reference_commands::reference_index_status, + reference_commands::reference_reindex, + ai_commands::pumas_load_model, + ai_commands::pumas_catalog, + ai_commands::ai_config_get, + ai_commands::pumas_unload_model, reference_commands::reference_upload, reference_commands::reference_delete, commands::object_script_story::command_object_field, diff --git a/src-tauri/src/reference_commands.rs b/src-tauri/src/reference_commands.rs index e780fdba..3af11458 100644 --- a/src-tauri/src/reference_commands.rs +++ b/src-tauri/src/reference_commands.rs @@ -29,3 +29,19 @@ pub fn reference_delete( let state = app.state::(); reference_service::delete_reference(&state, id).map_err(CommandError::from) } + +#[tauri::command] +pub fn reference_index_status( + app: tauri::AppHandle, +) -> Result { + let state = app.state::(); + reference_service::index_status(&state).map_err(CommandError::from) +} + +#[tauri::command] +pub fn reference_reindex(app: tauri::AppHandle) -> Result<(), CommandError> { + let state = app.state::(); + reference_service::list_references(&state).map_err(CommandError::from)?; + reference_service::schedule_reindex(&state); + Ok(()) +} diff --git a/ui/src/lib/README.md b/ui/src/lib/README.md index dfb07734..edb9604d 100644 --- a/ui/src/lib/README.md +++ b/ui/src/lib/README.md @@ -193,6 +193,18 @@ boundary, preserving unrelated author drafts and backend source/version custody. Timeline title transport carries exact text and its owned field read to the canonical native rename command; it does not use generic sparse object fields. +Local inference contracts add `pumas` chat selection, independent +`EmbeddingConfig`, reference index status and Pumas catalog projections in +`aiTypes.ts`. `api.ts` uses the existing desktop invocation boundary for current +AI configuration, installed Pumas model/profile selection, explicit Load/Unload, +and saved-reference Reindex/status. Pumas serving requests preserve the selected +provider/profile/device scope; no provider process lifecycle lives in the renderer. +`serverEventTypes.ts` includes `reference_retrieval` source names or an error for +the actual generation attachment, separate from upload acknowledgement. +Pumas unload transport exposes the typed success/error/unloaded outcome so the +operator surface can distinguish an unload receipt from an already-unserved +model; provider errors remain rejected desktop commands. + ## Guarded arc assignment and sibling reorder `timelineCommandApi.ts` sends `SetTimelineNodeArcsCommand` and diff --git a/ui/src/lib/aiTypes.ts b/ui/src/lib/aiTypes.ts index 1317b2d5..2ee05038 100644 --- a/ui/src/lib/aiTypes.ts +++ b/ui/src/lib/aiTypes.ts @@ -1,4 +1,4 @@ -export type BackendType = 'llama_cpp' | 'open_router'; +export type BackendType = 'llama_cpp' | 'pumas' | 'open_router'; export interface AiConfig { backend_type: BackendType; @@ -7,6 +7,8 @@ export interface AiConfig { max_tokens: number; base_url: string; api_key: string | null; + pumas_profile?: string; + embedding?: EmbeddingConfig; } export interface AiStatus { @@ -30,3 +32,36 @@ export interface ModelListResponse { models: ModelEntry[]; total_count: number; } + +export interface EmbeddingConfig { + provider: 'disabled' | 'pumas' | 'open_ai_compatible'; + base_url: string; + model: string; + profile: string; + revision: string; + api_key: string | null; +} +export interface ReferenceIndexStatus { + documents: { + document_id: string; + name: string; + state: string; + indexed_chunks: number; + dimensions: number | null; + model: string | null; + revision: string | null; + error: string | null; + }[]; +} +export interface PumasCatalog { + models: { models: { id: string; official_name: string; model_type: string }[] }; + profiles: { + snapshot: { profiles: { profile_id: string; provider: string; device: { mode: string } }[] }; + }; +} + +export interface PumasUnserveResult { + success: boolean; + error?: string | null; + unloaded: boolean; +} diff --git a/ui/src/lib/api.ts b/ui/src/lib/api.ts index ffdcc6c9..f91dfc51 100644 --- a/ui/src/lib/api.ts +++ b/ui/src/lib/api.ts @@ -1,4 +1,11 @@ -import type { AiConfig, AiStatus, ModelListResponse } from './aiTypes.js'; +import type { + AiConfig, + AiStatus, + ModelListResponse, + PumasCatalog, + PumasUnserveResult, + ReferenceIndexStatus, +} from './aiTypes.js'; import type { ChildPlan } from './childPlanningTypes.js'; import { invokeDesktop } from './desktopTransport.js'; import type { Project, ReferenceDocument, ReferenceType } from './projectTypes.js'; @@ -136,3 +143,42 @@ export function loadProject(path: string): Promise { export function listProjects(): Promise<{ name: string; path: string; modified: string }[]> { return invokeDesktop<{ name: string; path: string; modified: string }[]>('project_list'); } + +export function getAiConfig(): Promise { + return invokeDesktop('ai_config_get'); +} +export function getReferenceIndexStatus(): Promise { + return invokeDesktop('reference_index_status'); +} +export function reindexReferences(): Promise { + return invokeDesktop('reference_reindex'); +} +export function getPumasCatalog(endpoint: string): Promise { + return invokeDesktop('pumas_catalog', { endpoint }); +} +export function loadPumasModel( + endpoint: string, + model: string, + profile: string, + provider: string, + deviceMode: string, +): Promise { + return invokeDesktop('pumas_load_model', { + endpoint, + request: { + model_id: model, + config: { profile_id: profile, provider, device_mode: deviceMode, keep_loaded: true }, + }, + }); +} +export function unloadPumasModel( + endpoint: string, + model: string, + profile: string, + provider: string, +): Promise { + return invokeDesktop('pumas_unload_model', { + endpoint, + request: { model_id: model, profile_id: profile, provider }, + }); +} diff --git a/ui/src/lib/components/editor/AiPromptPreview.ssr.test.ts b/ui/src/lib/components/editor/AiPromptPreview.ssr.test.ts new file mode 100644 index 00000000..704754ac --- /dev/null +++ b/ui/src/lib/components/editor/AiPromptPreview.ssr.test.ts @@ -0,0 +1,23 @@ +import { describe, expect, it } from 'vitest'; +import { render } from 'svelte/server'; +import AiPromptPreview from './AiPromptPreview.svelte'; + +describe('consumed generation prompt inspection', () => { + it('shows the actually retrieved reference text without a refresh that could replace the receipt', () => { + const { body } = render(AiPromptPreview, { + props: { + notes: 'Generate a scene at the secret station', + context: { + system: 'Reference material: The station is called Marmalade Harbor.', + user: 'Write the scene.', + }, + loading: false, + label: 'Last Generation Prompt', + refreshable: false, + }, + }); + expect(body).toContain('Last Generation Prompt'); + expect(body).toContain('Marmalade Harbor'); + expect(body).not.toContain(' {}, + label = 'Raw AI Prompt', + refreshable = true, }: { notes: string; context: { system: string; user: string } | null; loading: boolean; - onrefresh: () => void; + onrefresh?: () => void; + label?: string; + refreshable?: boolean; } = $props(); {#if notes.trim()}
- Raw AI Prompt - + {label} + {#if refreshable} + + {/if} {#if context}
diff --git a/ui/src/lib/components/editor/BeatEditor.svelte b/ui/src/lib/components/editor/BeatEditor.svelte index fc199a47..ebe8ec63 100644 --- a/ui/src/lib/components/editor/BeatEditor.svelte +++ b/ui/src/lib/components/editor/BeatEditor.svelte @@ -384,6 +384,9 @@ streamingTokenCount={editorState.streamingTokenCount} generationError={editorState.generationError} {nodeContext} + generationContext={editorState.lastGenerationNodeId === node.id + ? editorState.generationContext + : null} {contextLoading} editorNode={selectedProjection!.node!} onrefreshcontext={refreshContext} diff --git a/ui/src/lib/components/editor/BeatNotesPanel.svelte b/ui/src/lib/components/editor/BeatNotesPanel.svelte index 2ee20d1d..f07eea0c 100644 --- a/ui/src/lib/components/editor/BeatNotesPanel.svelte +++ b/ui/src/lib/components/editor/BeatNotesPanel.svelte @@ -10,6 +10,7 @@ streamingTokenCount, generationError, nodeContext, + generationContext = null, contextLoading, editorNode, onrefreshcontext, @@ -20,6 +21,7 @@ generationError: string | null; nodeContext: { system: string; user: string } | null; contextLoading: boolean; + generationContext?: { system: string; user: string } | null; editorNode: SelectedNodeEditorNode; onrefreshcontext: () => void; } = $props(); @@ -38,6 +40,15 @@
{generationError}
{/if} + {#if generationContext} + + {/if} + import PumasSelection from './PumasSelection.svelte'; + import EmbeddingConfigPanel from './EmbeddingConfigPanel.svelte'; import type { AiConfig, BackendType } from '$lib/aiTypes.js'; - import { updateAiConfig } from '$lib/api.js'; + import { getAiConfig, updateAiConfig } from '$lib/api.js'; import { aiStatusState, refreshAiStatus } from '$lib/stores/aiStatus.svelte.js'; const BACKEND_BASE_URLS: Record = { + pumas: 'http://127.0.0.1:8080', llama_cpp: 'http://127.0.0.1:18080/v1', open_router: 'https://openrouter.ai/api/v1', }; @@ -19,7 +22,7 @@ let saving = $state(false); let statusMessage = $state(''); - let isLocalBackend = $derived(config.backend_type === 'llama_cpp'); + let isLocalBackend = $derived(config.backend_type !== 'open_router'); async function checkStatus() { await refreshAiStatus(); @@ -33,8 +36,8 @@ await updateAiConfig(config); await checkStatus(); statusMessage = aiStatusState.status?.connected ? 'Connected' : 'Connection failed'; - } catch { - statusMessage = 'Failed to save config'; + } catch (error) { + statusMessage = String(error); } saving = false; @@ -56,8 +59,15 @@ return 'auto (detect available model)'; } - // Check status on mount. + // Restore active backend configuration when the panel is reopened. $effect(() => { + void getAiConfig() + .then((active) => { + config = active; + }) + .catch((error) => { + statusMessage = String(error); + }); void checkStatus(); }); @@ -83,19 +93,22 @@ - + {#if config.backend_type !== 'pumas'} + + {/if} {#if isLocalBackend} {/if} + {#if config.backend_type === 'pumas'} + + {/if} + + +