Implement runtime-v2 session registry, turn loop, behavior SDK, and multi-lane UI
1363 unmodified linesExpand all13641365136613671368136913701371137213731374137513761377137812 unmodified lines139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214132551 unmodified linesExpand all1363 unmodified linesExpand allname = "klbr-core"version = "0.1.0"dependencies = [ "anyhow", "async-stream", "base64", "chrono", "dirs 5.0.1", "futures", "kdl", "klbr-runtime", "pulldown-cmark", "regex", "reqwest", "rusqlite",12 unmodified linesname = "klbr-daemon"version = "0.1.0"dependencies = [ "anyhow", "chrono", "futures", "klbr-core", "klbr-discord", "klbr-ipc", "klbr-runtime", "rusqlite", "serde", "serde_json", "tempfile", "tokio", "tokio-tungstenite", "tracing", "tracing-subscriber", "uuid",]
[[package]]name = "klbr-discord"2551 unmodified linesExpand all108 unmodified linesExpand all109110111112113114115116117118119120121122123124125126127108 unmodified linesExpand all{"id":"int-53386416e11173c95b46156bcbc17c9e","kind":"field_change","created_at":"2026-07-10T21:08:54.019353Z","actor":"dawn","issue_id":"klbr-1jn.2","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Removed global sync_reference_indexes calls from retrieval/reset, made promptable FTS/trigram updates incremental across writes and lifecycle changes, added explicit projection audit/repair APIs and corruption recovery tests, and stopped default episode dual writes."}}{"id":"int-90bcdc29f079523f7c232182cf82f7f9","kind":"field_change","created_at":"2026-07-10T21:08:54.538357Z","actor":"dawn","issue_id":"klbr-1jn.8","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Added persisted garden file/store sync hashes, explicit create/update/rename/delete/conflict outcomes, safe tombstoning on deletion, two-sided edit protection, documented manual store-to-file policy, and round-trip conflict/delete tests."}}{"id":"int-7370e0dc6b3fc39305398f3edd0d1186","kind":"field_change","created_at":"2026-07-10T21:19:04.027083Z","actor":"dawn","issue_id":"klbr-1jn.7","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"MemoryPipeline now runs bounded SQLite jobs on Tokio blocking workers; reads use independent query-only WAL handles, writes retain one ordered transactional connection, join/poison failures become structured errors, and concurrent ingest plus two retrievals pass under timeout."}}{"id":"int-30cab00671b12b0b15bc85f7d147f545","kind":"field_change","created_at":"2026-07-10T21:19:04.74902Z","actor":"dawn","issue_id":"klbr-1jn","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Completed all eight architecture hardening children: resumable atomic ingest, canonical storage ADR, incremental projections and repair audit, query-aligned safe facts, iterative collect repair, bitemporal rows, bounded async SQLite execution, and conflict-safe garden reconciliation. Docs and regression coverage updated."}}{"id":"int-0284c4a89952bd36081552d9a21b095c","kind":"field_change","created_at":"2026-09-06T00:06:54.933204Z","actor":"dawn","issue_id":"klbr-rhn","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-551b134aaf18921342bea79f06fcc7b7","kind":"field_change","created_at":"2026-09-06T00:14:09.533204Z","actor":"dawn","issue_id":"klbr-3xo.1","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-746e9da75785523a52d06de384305494","kind":"field_change","created_at":"2026-09-06T00:16:56.526827Z","actor":"dawn","issue_id":"klbr-h6p","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-45fbbe465c7a853eee3b68c64db99dbc","kind":"field_change","created_at":"2026-09-06T00:17:11.205887Z","actor":"dawn","issue_id":"klbr-7e3","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-692f6d4146c5dcac3b38ff123d19e060","kind":"field_change","created_at":"2026-09-06T00:18:45.772074Z","actor":"dawn","issue_id":"klbr-3xo.2","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-6ddf7eb01f1a5d3977ee07f773685948","kind":"field_change","created_at":"2026-09-06T00:19:11.249728Z","actor":"dawn","issue_id":"klbr-3xo.3","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-58a4ef64932833557319ceacc135f59c","kind":"field_change","created_at":"2026-09-06T00:19:27.845862Z","actor":"dawn","issue_id":"klbr-3xo","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Completed behavior typestates, dry-run ephemeral harness, CAS activation/rollback, and adversary tests"}}{"id":"int-0f469c115d2ee438fba3646d3327ccf3","kind":"field_change","created_at":"2026-09-06T00:21:05.936309Z","actor":"dawn","issue_id":"klbr-ft9","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-74c7f4175cf162bd602e8e41c407e264","kind":"field_change","created_at":"2026-09-06T00:26:23.870726Z","actor":"dawn","issue_id":"klbr-6bj","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-ee27eb6af1f446ff5ef1213e15e7b393","kind":"field_change","created_at":"2026-09-06T00:27:11.855166Z","actor":"dawn","issue_id":"klbr-lg6.1","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-617ac3544cb84c9be4b364a3663387f2","kind":"field_change","created_at":"2026-09-06T00:27:38.946356Z","actor":"dawn","issue_id":"klbr-tgb","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-b5c1f87a2182fbe9c72c8a8caef2e6a5","kind":"field_change","created_at":"2026-09-06T00:30:24.216442Z","actor":"dawn","issue_id":"klbr-lg6.3","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Authored and verified integration chaos tests in klbr-daemon/tests/chaos_test.rs: burst input flooding during blocked/failing hooks, unacknowledged items survival across restart, independent operator control, typestate transitions, and concurrent duplicate race deduplication."}}{"id":"int-027a12f054bb39fb9655ab10cc360bce","kind":"field_change","created_at":"2026-09-06T00:31:32.068915Z","actor":"dawn","issue_id":"klbr-lg6.2","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-5daaa7e37bf85a1a7227a9d6c42e7a1c","kind":"field_change","created_at":"2026-09-06T00:31:42.047144Z","actor":"dawn","issue_id":"klbr-lg6","extra":{"field":"status","new_value":"closed","old_value":"open","reason":"Closed"}}{"id":"int-8f35173680661c0b6e15af63cda38b54","kind":"field_change","created_at":"2026-09-06T00:36:05.99516Z","actor":"dawn","issue_id":"klbr-na3","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}7 unmodified lines89101112131415161716 unmodified lines7 unmodified linespath = "src/lib.rs"
[dependencies]anyhow = { workspace = true }async-stream = { workspace = true }klbr-runtime = { path = "../klbr-runtime" }base64 = { workspace = true }chrono = { workspace = true }dirs = { workspace = true }futures = { workspace = true }16 unmodified lines1 unmodified line2345678910111213141516171819202122232425262728293031321 unmodified linename = "klbr-daemon"version = "0.1.0"edition = "2021"
[lib]path = "src/lib.rs"doctest = false
[[bin]]name = "klbr-daemon"path = "src/main.rs"
[dependencies]anyhow = { workspace = true }chrono = { workspace = true }futures = { workspace = true }klbr-core = { path = "../klbr-core" }klbr-discord = { path = "../klbr-discord" }klbr-ipc = { path = "../klbr-ipc" }klbr-runtime = { path = "../klbr-runtime" }serde = { workspace = true }serde_json = { workspace = true }tempfile = { workspace = true }tokio = { workspace = true }tokio-tungstenite = { workspace = true }tracing = { workspace = true }tracing-subscriber = { workspace = true }uuid = { workspace = true }
[dev-dependencies]rusqlite = { workspace = true }10 unmodified lines1112131415161710 unmodified lines "sourceMap": true, "strict": true }, "include": ["src/**/*.d.ts", "src/**/*.ts", "src/**/*.svelte"], "exclude": ["src/**/*.test.ts"], "references": [{ "path": "./tsconfig.node.json" }]}11 unmodified lines12131415161718192069 unmodified lines11 unmodified linespub mod pipeline;pub mod planner;pub mod retrieval;pub mod router;pub mod runtime_turn;pub(crate) mod serde_utils;pub mod support;pub mod tools;
69 unmodified lines1234566 unmodified lines1314151617181920212223242526272829303132333435363738394041424344454647484950518 unmodified lines6061626364656667683 unmodified lines72737475767778798018 unmodified lines9910010110210310410510610733 unmodified lines1411421431441451461471481491501511521531541551561571581591601611621631645 unmodified lines170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202113 unmodified linesExpand all31631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442568 unmodified linesuse std::path::Path;use std::sync::Arc;use anyhow::Result;use futures::{SinkExt, StreamExt};use tokio::{ net::{TcpListener, TcpStream},6 unmodified lines memory::MemoryStore, AgentEvent, MetricsSnapshot,};use klbr_ipc::{ClientMsg, ServerMsg};use klbr_runtime::protocol::SessionId;
use crate::ingress::{admit_input, SourceRef};use crate::session_registry::SessionRegistry;
const HISTORY_PAGE: usize = 50;const WS_BIND_ADDR: &str = "127.0.0.1:8765";
type WsSink = futures::stream::SplitSink<WebSocketStream<TcpStream>, Message>;type WsSource = futures::stream::SplitStream<WebSocketStream<TcpStream>>;
fn parse_session_id(id_str: &str) -> SessionId { id_str.parse::<SessionId>().unwrap_or_else(|_| { // Sanitize to alphanumeric, '.', '_', '-' let sanitized: String = id_str .chars() .map(|c| if c.is_ascii_alphanumeric() || c == '.' || c == '_' || c == '-' { c } else { '_' }) .take(128) .collect(); sanitized.parse::<SessionId>().unwrap_or_else(|_| SessionId::fresh()) })}
pub async fn serve( interrupt_tx: mpsc::Sender<Interrupt>, output: broadcast::Sender<AgentEvent>, snapshot: MetricsSnapshot, memory: MemoryStore, history_window: usize, default_soul: String, registry: Arc<SessionRegistry>, mut shutdown: watch::Receiver<bool>,) -> Result<()> { let listener = TcpListener::bind(WS_BIND_ADDR).await?; tracing::info!(addr = WS_BIND_ADDR, "daemon listening");8 unmodified lines let bcast = output.clone(); let snapshot = snapshot.clone(); let memory = memory.clone(); let default_soul = default_soul.clone(); let registry = registry.clone(); let shutdown = shutdown.clone();
tokio::spawn(async move { if let Err(e) = handle(3 unmodified lines snapshot, memory, history_window, default_soul, registry, shutdown, ) .await {18 unmodified lines snapshot: MetricsSnapshot, memory: MemoryStore, history_window: usize, default_soul: String, registry: Arc<SessionRegistry>, mut shutdown: watch::Receiver<bool>,) -> Result<()> { let mut rx = output.subscribe(); let ws = accept_async(sock).await?;33 unmodified lines
let msg: ClientMsg = serde_json::from_str(&text)?; match msg { ClientMsg::Message { source, content } => { let session_id = parse_session_id("default"); let source_ref = SourceRef::from_client_message(&source, &content, session_id.as_str()); match admit_input(®istry, &session_id, source_ref).await { Ok(event_id) => { let _ = send_msg(&mut ws_tx, &ServerMsg::Admitted { session_id: session_id.to_string(), event_id, }).await; } Err(e) => { let _ = send_msg(&mut ws_tx, &ServerMsg::Error { content: format!("failed to admit message: {e}"), }).await; } }
let int = Interrupt::message(source, content); interrupt_tx.send(int).await?; } ClientMsg::ExternalEvent {5 unmodified lines message_id, timestamp, metadata, } => { let session_id = parse_session_id(&conversation_id); let source_ref = SourceRef::from_external_event( &source, &conversation_id, &content, author_id.clone(), author_name.clone(), message_id.clone(), timestamp, metadata.clone(), ); match admit_input(®istry, &session_id, source_ref).await { Ok(event_id) => { let _ = send_msg(&mut ws_tx, &ServerMsg::Admitted { session_id: session_id.to_string(), event_id, }).await; } Err(e) => { let _ = send_msg(&mut ws_tx, &ServerMsg::Error { content: format!("failed to admit external event: {e}"), }).await; } }
interrupt_tx .send(Interrupt::external_event(ExternalEvent { source, conversation_id,113 unmodified linesExpand all .await?; } } } // Operator control endpoints ClientMsg::Stop { session_id } => { let s_id = parse_session_id(&session_id); match registry.stop(&s_id).await { Ok(()) => { send_msg(&mut ws_tx, &ServerMsg::OperatorAck { action: "stop".into(), session_id: Some(session_id), success: true, message: None, }).await?; } Err(e) => { send_msg(&mut ws_tx, &ServerMsg::OperatorAck { action: "stop".into(), session_id: Some(session_id), success: false, message: Some(e.to_string()), }).await?; } } } ClientMsg::Drain { session_id } => { let s_id = parse_session_id(&session_id); match registry.drain(&s_id).await { Ok(()) => { send_msg(&mut ws_tx, &ServerMsg::OperatorAck { action: "drain".into(), session_id: Some(session_id), success: true, message: None, }).await?; } Err(e) => { send_msg(&mut ws_tx, &ServerMsg::OperatorAck { action: "drain".into(), session_id: Some(session_id), success: false, message: Some(e.to_string()), }).await?; } } } ClientMsg::Activate { session_id, expected_epoch, release_path } => { let s_id = parse_session_id(&session_id); let release_res = klbr_runtime::behavior::Release::load(Path::new(&release_path)); match release_res { Ok(release) => { match registry.activate(&s_id, expected_epoch, release).await { Ok(epoch) => { send_msg(&mut ws_tx, &ServerMsg::OperatorAck { action: "activate".into(), session_id: Some(session_id), success: true, message: Some(format!("activated epoch {epoch}")), }).await?; } Err(e) => { send_msg(&mut ws_tx, &ServerMsg::OperatorAck { action: "activate".into(), session_id: Some(session_id), success: false, message: Some(e.to_string()), }).await?; } } } Err(e) => { send_msg(&mut ws_tx, &ServerMsg::OperatorAck { action: "activate".into(), session_id: Some(session_id), success: false, message: Some(format!("failed to load release from {release_path}: {e}")), }).await?; } } } ClientMsg::SessionStatus { session_id } => { let s_id = parse_session_id(&session_id); let state_val = if let Some(entry) = registry.get(&s_id).await { let st = entry.state().await; serde_json::to_value(&st).unwrap_or(serde_json::json!("unknown")) } else { serde_json::json!("not_found") }; send_msg(&mut ws_tx, &ServerMsg::SessionStatus { session_id, state: state_val, }).await?; } ClientMsg::SessionList => { let sessions = registry.list().await; let serialized = sessions .into_iter() .map(|(id, st)| { (id.to_string(), serde_json::to_value(&st).unwrap_or(serde_json::json!("unknown"))) }) .collect(); send_msg(&mut ws_tx, &ServerMsg::SessionList { sessions: serialized, }).await?; } } } res = rx.recv() => { let ev = match res {68 unmodified lines123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375//! Ingress boundary parser, canonical SourceRef, and atomic admission coordinator.
use anyhow::{bail, Context, Result};use chrono::Utc;use klbr_runtime::protocol::SessionId;use serde::{Deserialize, Serialize};use uuid::Uuid;
use crate::session_registry::SessionRegistry;
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]pub struct SourceRef { pub source_kind: String, // e.g. "operator", "websocket", "discord.dm", "discord.mention", "discord.ambient" pub conversation_id: String, pub author_id: Option<String>, pub author_name: Option<String>, pub message_id: Option<String>, pub timestamp_ms: i64, pub dedupe_key: String, // canonical dedupe key pub content: String, pub metadata: serde_json::Value,}
impl SourceRef { pub fn new( source_kind: impl Into<String>, conversation_id: impl Into<String>, content: impl Into<String>, ) -> Self { let source_kind = source_kind.into(); let conversation_id = conversation_id.into(); let content = content.into(); let timestamp_ms = Utc::now().timestamp_millis(); let message_id = Uuid::new_v4().to_string(); let dedupe_key = format!("{source_kind}:{conversation_id}:{message_id}"); Self { source_kind, conversation_id, author_id: None, author_name: None, message_id: Some(message_id), timestamp_ms, dedupe_key, content, metadata: serde_json::Value::Null, } }
pub fn with_dedupe_key(mut self, dedupe_key: impl Into<String>) -> Self { self.dedupe_key = dedupe_key.into(); self }
pub fn with_message_id(mut self, message_id: impl Into<String>) -> Self { let mid = message_id.into(); self.dedupe_key = format!("{}:{}:{}", self.source_kind, self.conversation_id, mid); self.message_id = Some(mid); self }
pub fn with_author(mut self, author_id: Option<String>, author_name: Option<String>) -> Self { self.author_id = author_id; self.author_name = author_name; self }
pub fn with_timestamp_ms(mut self, timestamp_ms: i64) -> Self { self.timestamp_ms = timestamp_ms; self }
pub fn with_metadata(mut self, metadata: serde_json::Value) -> Self { self.metadata = metadata; self }
pub fn from_client_message( source: impl Into<String>, content: impl Into<String>, session_id: impl Into<String>, ) -> Self { let source = source.into(); let content = content.into(); let session_id = session_id.into(); let source_kind = if source.trim().is_empty() { "websocket".to_string() } else { source }; let timestamp_ms = Utc::now().timestamp_millis(); let message_id = Uuid::new_v4().to_string(); let dedupe_key = format!("{source_kind}:{session_id}:{message_id}"); Self { source_kind, conversation_id: session_id, author_id: None, author_name: None, message_id: Some(message_id), timestamp_ms, dedupe_key, content, metadata: serde_json::Value::Null, } }
pub fn from_external_event( source: impl Into<String>, conversation_id: impl Into<String>, content: impl Into<String>, author_id: Option<String>, author_name: Option<String>, message_id: Option<String>, timestamp: Option<i64>, metadata: serde_json::Value, ) -> Self { let source = source.into(); let conversation_id = conversation_id.into(); let content = content.into(); let source_kind = if source.trim().is_empty() { "external".to_string() } else { source }; let timestamp_ms = timestamp.unwrap_or_else(|| Utc::now().timestamp_millis()); let mid = message_id.clone().unwrap_or_else(|| Uuid::new_v4().to_string()); let dedupe_key = format!("{source_kind}:{conversation_id}:{mid}"); Self { source_kind, conversation_id, author_id, author_name, message_id, timestamp_ms, dedupe_key, content, metadata, } }
pub fn to_payload(&self) -> serde_json::Value { serde_json::to_value(self).unwrap_or_else(|_| { serde_json::json!({ "source_kind": self.source_kind, "conversation_id": self.conversation_id, "author_id": self.author_id, "author_name": self.author_name, "message_id": self.message_id, "timestamp_ms": self.timestamp_ms, "dedupe_key": self.dedupe_key, "content": self.content, "metadata": self.metadata, }) }) }}
/// Ingress admission coordinator./// Invariants:/// a) Check if session can admit (entry.can_admit()). If draining or stopped, bail with descriptive error./// b) Atomic admission: registry.store().admit() must land before any acknowledgment./// c) Non-blocking policy: attention policy runs AFTER admission; policy failure never drops inbox items./// d) Returns admitted event sequence id.pub async fn admit_input( registry: &SessionRegistry, session_id: &SessionId, source_ref: SourceRef,) -> Result<i64> { // a) Check if session can admit. If draining or stopped, bail with descriptive error. if let Some(entry) = registry.get(session_id).await { if !entry.can_admit().await { let state = entry.state().await; bail!( "cannot admit input: session {} is in state {:?} (draining or stopped)", session_id, state ); } }
// Ensure session row exists in the store before inserting events (SQLite foreign key constraint) registry .store() .ensure_session(session_id) .await .context("failed to ensure session exists in store")?;
// b) Atomic admission: registry.store().admit() must land in SQLite before acknowledging or running policy! let payload = source_ref.to_payload(); let event_id = match registry .store() .admit(session_id, source_ref.dedupe_key.clone(), payload) .await { Ok(id) => id, Err(err) => { // If the key was already admitted concurrently (or with slight variation in metadata/content), // retrieve the already-admitted event sequence id for deduplication. if let Ok(Some(id)) = registry .store() .find_event_by_source_key(session_id, &source_ref.dedupe_key) .await { id } else { return Err(err).context("failed to atomically admit input into store"); } } };
// c) Non-blocking policy: spawn a background task to evaluate attention policy on the session runtime. // CRITICAL: If the policy call fails, times out, or errors, LOG it, but NEVER drop or delete the admitted inbox item! It remains pending in SQLite `inbox` table. if let Some(entry) = registry.get(session_id).await { let policy = entry.runtime.policy().await; let source_kind = source_ref.source_kind.clone(); let timestamp_ms = source_ref.timestamp_ms; let s_id = session_id.clone(); tokio::spawn(async move { match tokio::time::timeout( std::time::Duration::from_secs(5), policy.attention(event_id, &source_kind, timestamp_ms), ) .await { Ok(Ok(decision)) => { tracing::debug!(session = %s_id, event_id, ?decision, "attention policy evaluated"); } Ok(Err(err)) => { tracing::warn!(session = %s_id, event_id, err = %err, "attention policy evaluation returned error; retaining pending inbox item"); } Err(_) => { tracing::warn!(session = %s_id, event_id, "attention policy evaluation timed out; retaining pending inbox item"); } } }); }
// d) Return the admitted event_id (the sequence number). Ok(event_id)}
#[cfg(test)]mod tests { use super::*;
#[test] fn test_source_ref_from_client_message() { let sref = SourceRef::from_client_message("operator", "hello klbr", "sess-1"); assert_eq!(sref.source_kind, "operator"); assert_eq!(sref.conversation_id, "sess-1"); assert_eq!(sref.content, "hello klbr"); assert!(sref.message_id.is_some()); assert!(sref.dedupe_key.starts_with("operator:sess-1:"));
let default_source = SourceRef::from_client_message("", "test", "sess-2"); assert_eq!(default_source.source_kind, "websocket"); assert!(default_source.dedupe_key.starts_with("websocket:sess-2:")); }
#[test] fn test_source_ref_from_external_event() { let sref = SourceRef::from_external_event( "discord.mention", "chan-42", "hey agent", Some("u1".into()), Some("alice".into()), Some("msg-99".into()), Some(1700000000), serde_json::json!({"guild_id": "g1"}), ); assert_eq!(sref.source_kind, "discord.mention"); assert_eq!(sref.conversation_id, "chan-42"); assert_eq!(sref.dedupe_key, "discord.mention:chan-42:msg-99"); assert_eq!(sref.author_name.as_deref(), Some("alice")); assert_eq!(sref.timestamp_ms, 1700000000); assert_eq!(sref.metadata["guild_id"], "g1"); }
#[test] fn test_source_ref_payload_roundtrip() { let sref = SourceRef::new("websocket", "sess-test", "sample content"); let payload = sref.to_payload(); assert_eq!(payload["source_kind"], "websocket"); assert_eq!(payload["conversation_id"], "sess-test"); assert_eq!(payload["content"], "sample content");
let deserialized: SourceRef = serde_json::from_value(payload).expect("deserialize payload"); assert_eq!(sref, deserialized); }
#[tokio::test] async fn test_admit_input_atomic_store_and_inbox() -> Result<()> { let temp = tempfile::tempdir()?; let store = klbr_runtime::store::Store::open(temp.path().join("admit.sqlite"))?; let root = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")) .parent() .unwrap() .to_path_buf(); let config = klbr_runtime::worker::WorkerConfig::source_tree(&root); let registry = SessionRegistry::new(store.clone(), config);
let session_id = SessionId::fresh(); let sref = SourceRef::from_client_message("operator", "deploy release 2", session_id.as_str());
let event_id = admit_input(®istry, &session_id, sref.clone()).await?; assert!(event_id > 0);
// Verify it landed atomically in SQLite inbox table with status pending let pending = store.pending(&session_id, 10).await?; assert_eq!(pending, vec![event_id]);
Ok(()) }
#[tokio::test] async fn test_admit_input_deduplication() -> Result<()> { let temp = tempfile::tempdir()?; let store = klbr_runtime::store::Store::open(temp.path().join("dedupe.sqlite"))?; let root = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")) .parent() .unwrap() .to_path_buf(); let config = klbr_runtime::worker::WorkerConfig::source_tree(&root); let registry = SessionRegistry::new(store.clone(), config);
let session_id = SessionId::fresh(); let sref = SourceRef::new("websocket", session_id.as_str(), "same message") .with_dedupe_key("ws:conv1:fixed-key");
let id1 = admit_input(®istry, &session_id, sref.clone()).await?; let id2 = admit_input(®istry, &session_id, sref).await?; assert_eq!(id1, id2, "identical dedupe key must return original event_id");
let pending = store.pending(&session_id, 10).await?; assert_eq!(pending.len(), 1);
Ok(()) }
#[tokio::test] async fn test_admit_input_rejected_when_draining_or_stopped() -> Result<()> { let temp = tempfile::tempdir()?; let store = klbr_runtime::store::Store::open(temp.path().join("stopped.sqlite"))?; let root = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")) .parent() .unwrap() .to_path_buf(); let initial = klbr_runtime::behavior::Release::publish( &root.join("behavior"), &temp.path().join("releases"), )?; let config = klbr_runtime::worker::WorkerConfig::source_tree(&root); let registry = SessionRegistry::new(store.clone(), config);
let session_id = SessionId::fresh(); let _entry = registry.get_or_open(&session_id, &initial).await?;
// Drain session registry.drain(&session_id).await?;
// Attempt admission should fail immediately let sref = SourceRef::from_client_message("user", "cannot accept", session_id.as_str()); let res = admit_input(®istry, &session_id, sref).await; assert!(res.is_err()); let err_msg = res.unwrap_err().to_string(); assert!(err_msg.contains("cannot admit input")); assert!(err_msg.contains("draining or stopped"));
// Verify nothing was admitted into pending inbox let pending = store.pending(&session_id, 10).await?; assert!(pending.is_empty());
Ok(()) }}123pub mod daemon;pub mod ingress;pub mod session_registry;12123456786 unmodified lines151617181920212223242526272824 unmodified lines53545556575859606144 unmodified linesmod daemon;
use anyhow::Result;use klbr_core::{agent, config::Config, memory, tools, AgentEvent, MetricsSnapshot};use klbr_daemon::{daemon, session_registry::SessionRegistry};use klbr_runtime::{store::Store, worker::WorkerConfig};use std::sync::Arc;use tokio::sync::{broadcast, mpsc, watch, RwLock};
#[tokio::main]6 unmodified lines let memory = memory::MemoryStore::open(&config.db_path, config.models.embed_dim)?; memory.ensure_soul_memory(&default_soul)?; let snapshot = Arc::new(RwLock::new(None)) as MetricsSnapshot;
// Initialize runtime-v2 store and session registry let runtime_store = Store::open("klbr_runtime.db")?; let project_root = std::env::current_dir().unwrap_or_else(|_| ".".into()); let worker_config = WorkerConfig::source_tree(&project_root); let session_registry = Arc::new(SessionRegistry::new(runtime_store, worker_config));
let (interrupt_tx, interrupt_rx) = mpsc::channel(32); let (output_tx, _) = broadcast::channel::<AgentEvent>(4096); let (shutdown_tx, shutdown_rx) = watch::channel(false);
24 unmodified lines snapshot, memory, history_window, default_soul, session_registry, shutdown_rx, ));
tokio::select! {44 unmodified lines