Integrate runtime-v2 across daemon, SDK, turn runner, and multi-lane IPC
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151# klbr runtime-v2 — Deep Architecture & Implementation Report
**Date:** September 6, 2026 **Status:** Runtime v2 foundation fully implemented, proven through adversarial tests, and committed. **Git commit:** `97d42c9` on `main` **Quality gates:** All Rust, Python, and TypeScript tests passing 100% green.
---
## 1. Architecture & Design Philosophy (Lucid Code)
Runtime v2 replaces the previous implicit agent loop and monolithic memory coupling with a principled **Rust mechanism / Python policy** architecture.
### Core Tenets1. **Typestate & Illegal State Elimination:** - State machines never rely on ambiguous boolean flags (e.g. `is_running`, `is_waiting`). - Every state is a closed algebraic enum where only valid transitions are syntactically and semantically legal. - Hook evaluations fail closed: timeouts, exceptions, or missing workers yield `Held` or `Decline`, never accidental approval.2. **Boundary Parsing & Canonical Ingress:** - Raw input from WebSockets, CLIs, or Discord is parsed at the outer edge into strongly-typed `SourceRef` records before touching persistence. - Deduplication keys are canonical and deterministic (`{source_kind}:{conversation_id}:{message_id}`).3. **Strict Lane Separation:** - **Scratchpad Lane:** Model thinking and raw token generation are strictly diagnostic reasoning records. - **Cell Output Lane:** Stdout/stderr and expression results from code execution are strictly diagnostic terminal logs. - **Speech Lane:** Conversational messages to participants occur **exclusively** via explicit `await local.send(...)`, validated by `delivery.review` and committed as durable events.4. **Terminal Yields vs Idle Spinning:** - `await runtime.wait(...)` is a hard host-level yield and state commit, suspending execution at a clean boundary. It never burns CPU or spins an `asyncio.sleep` polling loop.5. **Content-Addressed Behavior & Epoch Pinning:** - Releases are content-addressed SHA-256 snapshots of source files. - In-flight cells pin their active release epoch so hook activations never shift out from under an active execution.
---
## 2. What Has Been Implemented & How
### A. Core Runtime Substrate (`klbr-runtime`)- **Transport & Protocol (`protocol.rs`):** - Private Unix-domain stream sockets with randomized handshake startup tokens. - 4-byte big-endian framing with strict length bounds (1..1,048,576 bytes). - Strict identifier validation (1..128 ASCII alphanumeric/dot/hyphen/underscore). - Typed envelopes (`hello`, `execute`, `invoke`, `host_request`, `host_reply`, `completed`, `interrupt`, `shutdown`).- **Process Supervision (`worker.rs`):** - Supervised CPython processes with isolated environment clearing. - Two distinct roles: * `workbench`: Persistent globals across cells, top-level await, explicit effect RPCs (`local.send`, `runtime.wait`). * `hooks`: Stateless, isolated policy worker evaluating hooks with zero access to workbench heap or ambient side-effect scopes. - Continuous streaming into separate bounded 8 KiB diagnostic tails for native stdout/stderr. - Hard deadlines and non-blocking process killing.- **Persistence & Immutability (`store.rs`, `migrations/001_runtime.sql`):** - SQLite `STRICT` tables with PRAGMA `application_id = 1263288882` (`KLB2`). - Cooperative file locking (`.owner.lock`) preventing concurrent daemon processes from corrupting the DB. - Upstream WAL-reset verification guard (SQLite 3.51.3+, 3.50.7+, 3.44.6+). - Immutability triggers: `events` table rejects all `UPDATE` and `DELETE` operations at the database engine level. - Atomic transactions for input admission, local effect publication, and wait state updates.- **Session Supervision (`session.rs`):** - `SessionRuntime` manages active worker processes, request/reply routing, and policy release epochs. - Mid-flight revision pinning and epoch compare-and-swap (CAS).- **Hooks & Verification (`hooks.rs`, `behavior.rs`):** - Fail-closed execution for `attention.plan` and `delivery.review`. - Deterministic SHA-256 snapshot hashing for behavior releases.
### B. Session Registry & Ingress Coordinator (`klbr-daemon`)- **Session Registry (`session_registry.rs`):** - `SessionState` typestate: `Ready`, `Executing { epoch, execution_id }`, `Waiting { receipt, wake_at_ms, reason }`, `Draining`, `Stopped`. - Single-owner synchronization: concurrent `get_or_open` calls for the same session resolve to an identical `Arc<SessionEntry>`. - Operator control API: independent `stop`, `drain`, `activate`, and `status` endpoints that work without model or Python cooperation.- **Ingress Boundary & Admission (`ingress.rs`):** - `SourceRef` parser with automatic timestamping and UUID-based or explicit deduplication keys. - `admit_input`: verifies admissibility (`can_admit`), executes atomic SQLite `store.admit()` prior to acknowledgment, and spawns non-blocking attention policy evaluation. - Critical invariant: policy error, timeout, or panic **never drops, deletes, or marks consumed** any pending inbox item.- **Chaos & Adversarial Proofs (`tests/chaos_test.rs`):** - `test_burst_input_flooding_during_blocked_hooks`: 100 concurrent tasks flooded while hooks block and fail; 100% durable admission into SQLite inbox with status `pending`. - `test_unacknowledged_items_survive_restart`: Simulated sudden node crash/restart; verified all unacknowledged items survive in exact order with intact payloads. - `test_independent_operator_control_and_typestate`: Proved operator stop and drain halt workers immediately and reject subsequent admissions. - `test_concurrent_duplicate_admit_race`: 50 concurrent duplicate requests deduplicated cleanly into a single event and inbox row.
### C. Model Turn Loop & Provider Boundary (`klbr-core`)- **Turn State Machine Algebra (`runtime_turn/state.rs`):** - Sealed typestate phases: `Idle`, `Streaming`, `ExecutingTool`, `Yielded`, `Completed`, `Failed`, `Interrupted`. - Strict event typing: * `ScratchpadEvent::{ThoughtDelta, ProseDelta, ToolCallProposed}` * `CellOutputRecord` with `CellOutputStream::{Stdout, Stderr, Result}` * `SpeechReceipt` with `SpeechStatus::{Confirmed, Held}` * `YieldRecord` representing `runtime.wait`- **Integrator & Single-Tool Contract (`runtime_turn/integrator.rs`):** - Model has exactly one tool: `python(code: str)`. - Streaming parser: maps `LlmEvent::ThinkToken` and `Token` to scratchpad records. - Supervised workbench bridge: executes cells via `SessionRuntime`, maps outcomes, and polls Store for approved/held speech receipts. - Async stream turn runner: `run_turn_step_from_stream` and `run_turn_loop`.- **Adversary Verification (`runtime_turn/tests.rs`):** - 37 passing unit tests covering mid-stream truncation, malformed tool args, stdout isolation, terminal wait, and cancellation preemption races.
### D. Autonomous Behavior & Hook SDK (`python/klbr-runtime`, `behavior/`)- **Self-Modification SDK (`klbr/behavior.py`, `behavior_runner.py`):** - `Draft -> ValidatedCandidate -> PublishedRelease -> ActiveEpoch` typestate progression. - `Draft.create()`, `Draft.write_file()`, `Draft.validate()`. - `EphemeralRunner`: isolated dry-run harness evaluating synthetic `attention.plan` and `delivery.review` inputs in a temporary subprocess before CAS activation. - Atomic SQLite CAS switch (`BEGIN IMMEDIATE`, version check, `hooks.activated` event).- **Adversarial Test Suite (`test_behavior_adversary.py`):** - 90 passing tests covering corrupted manifests, duplicate keys, path traversal, symlink rejection, byte limits, post-publish tampering, hook infinite loops/hangs, and CAS contention.
### E. Client Multi-Lane IPC & Svelte UI (`klbr-ipc`, `klbr-web`)- **Multi-Lane IPC Protocol (`klbr-ipc/src/v2.rs`, `cursor.rs`):** - Discriminated union event schemas covering `scratchpad`, `cell_output`, `local_sent`, `input`, and `lifecycle`. - Sequence cursor algebra: tracks `last_seen_seq` for gap detection, deduplication, and out-of-order reassembly. - Idempotent reconnects: client sends last seen cursor; daemon replays missed durable events.- **Frontend Multi-Lane UI (`klbr-web/src/`):** - `ScratchpadAccordion.svelte`: collapsible drawer rendering streaming thoughts. - `CellTerminal.svelte`: styled terminal output rendering stdout, stderr, and expression results. - `SpeechBubble.svelte`: conversational speech bubbles for confirmed `local.send()` messages. - `LaneTimeline.svelte`: chronological interleaved event view. - `streaming-v2.ts`: resilient WebSocket client with automatic cursor tracking. - 18 rust tests in `klbr-ipc`, 19 bun tests in `klbr-web` (including 10,000-line burst stress tests), and clean production Vite build.
---
## 3. What Has NOT Been Done Yet
The following items are deliberate boundaries not yet implemented:
1. **Discord Gateway Transport (`klbr-discord`):** - The Twilight gateway bot currently runs on the legacy loop. It has not yet been wired to feed incoming Discord events into `klbr_daemon::ingress::admit_input`. - The Discord outbox worker (dispatching confirmed `local.sent` events to Discord channels via Twilight HTTP) has not been written.2. **Autonomous Background Scheduler (`wake_due`):** - `Store::wake_due()` is implemented and verified, but no daemon thread or timer worker automatically calls it on an interval to wake expired wait deadlines.3. **Lossless Context Management (LCM) & Artifact Spooling:** - Output beyond the preview bound (64 KiB) is currently truncated with a byte counter. It is not yet spooled to disk artifacts (`artifacts/`). - Chronological summary coverage and structured context injection are not yet wired into the v2 turn loop.4. **Tree-Sitter / Code-Intel Port:** - The rich code navigation tools (`find_definition`, `read_symbol`, `symbol_overview`) remain in legacy `klbr-core` tools. They have not yet been packaged as native Python skills inside the agent's workbench.5. **Production SQLite Migration from Legacy DB:** - The runtime store deliberately refuses to open the legacy memory database (`agent.db`). There is no automated migration script converting legacy turn/memory rows into v2 events.6. **Operating System Level Sandboxing:** - Subprocesses run under standard same-user Unix process boundaries with cleared environment variables. OS-level cgroups, namespaces, and RLIMITs are not implemented.
---
## 4. Verification & Quality Gates
All checks pass with zero warnings:
| Suite | Command | Result ||---|---|---|| Rust Workspace | `cargo test --workspace` | All unit, integration, and chaos tests pass || Daemon Chaos Tests | `cargo test -p klbr-daemon --test chaos_test` | 4/4 passed (bursts, crash/restart, races) || Core Turn Tests | `cargo test -p klbr-core --lib runtime_turn` | 37/37 passed || IPC Protocol Tests | `cargo test -p klbr-ipc` | 18/18 passed || Python Runtime & Behavior | `python3 tools/check_runtime.py --rust` | 90 python + 17 rust tests passed || Frontend Web Client | `cd klbr-web && bun test` | 19/19 passed (4,113 expect assertions) || Frontend Production Build | `cd klbr-web && bun run build` | Clean Vite build in ~2.6s || Beads Task Tracker | `bd status` | 123/123 issues closed |123 unmodified linesExpand all124125126127128129130131132123 unmodified linesExpand all{"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"}}{"id":"int-1cc31b5389e12bccb8d39ed0cd81a98e","kind":"field_change","created_at":"2026-09-06T14:31:11.533675Z","actor":"dawn","issue_id":"klbr-4fr","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-7522a87e70f6fbbc3a67cb49fdbb30fa","kind":"field_change","created_at":"2026-09-06T14:38:58.564197Z","actor":"dawn","issue_id":"klbr-h37","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-76681c810ef3778cea80ba29b5c876e6","kind":"field_change","created_at":"2026-09-06T14:38:59.945928Z","actor":"dawn","issue_id":"klbr-06i","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-af125e68fb1857ce6a635781b5b963b5","kind":"field_change","created_at":"2026-09-06T14:45:19.598535Z","actor":"dawn","issue_id":"klbr-2d0","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-234b77c3332e105337eb5a361c9564f3","kind":"field_change","created_at":"2026-09-06T14:55:35.56433Z","actor":"dawn","issue_id":"klbr-99b","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}268 unmodified linesExpand all269270271272273274275276277278452 unmodified linesExpand all7317327337347337357367377387397407412 unmodified lines7447457467477447487497507517527537547557567577587597602 unmodified lines76376476576675576776876977077177277342 unmodified lines816817818819820821822823824103 unmodified linesExpand all9289299309319329339349359361280 unmodified linesExpand all268 unmodified linesExpand all Usage(Usage), /// emitted once when the model decides to call one or more tools ToolCalls(Vec<ToolCall>), ToolCallStarted, Completed, Error(String),}
// ── client ────────────────────────────────────────────────────────────────────
452 unmodified linesExpand all
let res = match res { Ok(r) => r, Err(e) => { tracing::error!("failed to send stream request to {}: {:?}", endpoint, e); let err_msg = format!("failed to send stream request to {endpoint}: {e:?}"); tracing::error!("{}", err_msg); let _ = tok_tx.send(LlmEvent::Error(err_msg)).await; return Err(e.into()); } }; tracing::info!("stream response status: {}", res.status());2 unmodified lines .get("x-bridge-setup-ms") .and_then(|h| h.to_str().ok()) .and_then(|s| s.parse::<u64>().ok()) .unwrap_or(0); let res = Self::error_for_status_with_body(res).await?; let res = match Self::error_for_status_with_body(res).await { Ok(r) => r, Err(e) => { let err_msg = format!("stream error response from {endpoint}: {e:?}"); tracing::error!("{}", err_msg); let _ = tok_tx.send(LlmEvent::Error(err_msg)).await; return Err(e); } }; let mut res = res.bytes_stream();
let mut buf = String::new(); let mut partial_calls: HashMap<usize, PartialCall> = HashMap::new();2 unmodified lines while let Some(chunk) = res.next().await { let chunk = match chunk { Ok(c) => c, Err(e) => { tracing::error!("error reading stream chunk: {:?}", e); let err_msg = format!("error reading stream chunk: {e:?}"); tracing::error!("{}", err_msg); let _ = tok_tx.send(LlmEvent::Error(err_msg)).await; return Err(e.into()); } }; buf.push_str(std::str::from_utf8(&chunk).unwrap_or(""));42 unmodified lines u.generation_ms = 0; } u.bridge_processing_ms = bridge_setup_ms; let _ = tok_tx.send(LlmEvent::Usage(u)).await; let _ = tok_tx.send(LlmEvent::Completed).await;
return Ok(()); }
103 unmodified linesExpand all u.generation_ms = 0; } u.bridge_processing_ms = bridge_setup_ms; let _ = tok_tx.send(LlmEvent::Usage(u)).await; let _ = tok_tx.send(LlmEvent::Completed).await;
Ok(()) }
1280 unmodified linesExpand all1233456782 unmodified lines11121314151617181920211617222324252627282930313220333435363738394028414230434433454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672684826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934010 unmodified lines35135235335435535635735835936036136236336436536636736836937037137237337437537637737837938011011111238138238338438511411511611711811912012112212312412512612712812913013138638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944014244144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255314814915015115255455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558616016116258758858959015 unmodified lines606607608609610186187188189190611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653198199200201202203204205206207208209210654655656657658659660661218662663664665666221667668669670671224672673674675676677678229679680681682683684260 unmodified linesExpand alluse std::path::Path;use std::sync::Arc;use anyhow::Result;use std::time::Duration;use anyhow::{bail, Result};use futures::{SinkExt, StreamExt};use tokio::{ net::{TcpListener, TcpStream}, sync::{broadcast, mpsc, watch},2 unmodified lines
use klbr_core::{ interrupt::{ExternalEvent, Interrupt}, memory::MemoryStore, models::{LlmClient, Message as CoreMessage}, runtime_turn::{ integrator::run_turn_loop, state::{CellOutputStream, ScratchpadEvent, TurnEvent}, }, AgentEvent, MetricsSnapshot,};use klbr_ipc::{ClientMsg, ServerMsg};use klbr_runtime::protocol::SessionId;use klbr_ipc::{ v2::{InputPayload, Lane, LanePayload, LifecyclePayload, V2ClientMsg, V2EventRecord, V2ServerMsg}, ClientMsg, ServerMsg,};use klbr_runtime::{ behavior::Release, protocol::{OperationId, SessionId}, store::{Store, StoreAdmit},};
use crate::ingress::{admit_input, SourceRef};use crate::session_registry::SessionRegistry;use crate::session_registry::{SessionRegistry, SessionState};
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 {pub 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 { '_' }) .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 query_events_after( store: &Store, session_id: &SessionId, after_seq: i64,) -> Result<Vec<V2EventRecord>> { let events = store.history(session_id, after_seq, 1024).await?; let mut records = Vec::new(); for ev in events { let payload = if ev.kind == "input" { let text = ev .payload .get("content") .and_then(|c| c.as_str()) .unwrap_or("") .to_string(); let source = ev .payload .get("source_kind") .and_then(|s| s.as_str()) .unwrap_or("unknown") .to_string(); let author = ev .payload .get("author_name") .and_then(|a| a.as_str()) .map(|s| s.to_string()); LanePayload::Input(InputPayload { source, text, author, }) } else { LanePayload::Lifecycle(LifecyclePayload { kind: ev.kind, data: ev.payload, }) }; records.push(V2EventRecord::new( ev.id, session_id.to_string(), payload, ev.created_ms, )); } Ok(records)}
pub async fn trigger_session_turn( registry: Arc<SessionRegistry>, llm: LlmClient, session_id: SessionId, initial_release: Release, v2_event_tx: broadcast::Sender<V2ServerMsg>,) -> Result<()> { let entry = registry.get_or_open(&session_id, &initial_release).await?;
let state = entry.state().await; match state { SessionState::Draining | SessionState::Stopped => { bail!("session {session_id} is in state {state:?}, cannot run turn"); } SessionState::Executing { .. } => { return Ok(()); } SessionState::Waiting { .. } => { entry.transition_to_ready_with_store(registry.store()).await?; } SessionState::Ready => {} }
let (abort_handle, abort_reg) = futures::future::AbortHandle::new_pair(); entry.register_runner(abort_handle).await;
let entry_clone = entry.clone(); let reg_clone = registry.clone(); let s_id = session_id.clone(); let tx_clone = v2_event_tx.clone();
tokio::spawn(async move { let runner_future = async { loop { if !entry_clone.can_admit().await { break; }
let pending_ids = match reg_clone.store().pending(&s_id, 10).await { Ok(ids) if !ids.is_empty() => ids, _ => break, };
let op_id = OperationId::fresh(); let epoch = entry_clone.runtime.hook_epoch().await; if let Err(e) = entry_clone.transition_to_executing(epoch, op_id).await { tracing::warn!(session = %s_id, err = %e, "cannot transition to executing"); break; }
let mut messages = Vec::new(); for id in &pending_ids { if let Ok(Some(ev)) = reg_clone.store().get_event(&s_id, *id).await { let content = ev .payload .get("content") .and_then(|c| c.as_str()) .unwrap_or(""); messages.push(CoreMessage::user(content)); } }
let turn_stream = run_turn_loop( &llm, &entry_clone.runtime, messages, 10, Duration::from_secs(30), ); tokio::pin!(turn_stream);
let mut yielded = None; while let Some(turn_event) = turn_stream.next().await { match &turn_event { TurnEvent::Scratchpad(sp) => { let text = match sp { ScratchpadEvent::ThoughtDelta(d) | ScratchpadEvent::ProseDelta(d) => { d.clone() } ScratchpadEvent::ToolCallProposed { arguments, .. } => { arguments.clone() } _ => String::new(), }; let _ = tx_clone.send(V2ServerMsg::LiveChunk { session_id: s_id.to_string(), lane: Lane::Scratchpad, execution_id: None, stream: None, delta: text, }); } TurnEvent::CellOutput(co) => { let stream_str = match co.stream { CellOutputStream::Stdout => "stdout", CellOutputStream::Stderr => "stderr", CellOutputStream::Result => "result", }; let _ = tx_clone.send(V2ServerMsg::LiveChunk { session_id: s_id.to_string(), lane: Lane::Cell, execution_id: None, stream: Some(stream_str.to_string()), delta: co.text.clone(), }); } TurnEvent::Speech(sp) => { let _ = tx_clone.send(V2ServerMsg::LiveChunk { session_id: s_id.to_string(), lane: Lane::Speech, execution_id: None, stream: None, delta: sp.text.clone(), }); } TurnEvent::Yielded(y) => { yielded = Some(y.clone()); } _ => {} } }
let _ = reg_clone.store().acknowledge_inputs(&s_id, pending_ids).await;
if let Some(y) = yielded { let wake_at = y .seconds .map(|s| chrono::Utc::now().timestamp_millis() + (s * 1000.0) as i64); let _ = entry_clone .transition_to_waiting_with_store( reg_clone.store(), None, wake_at, Some(y.reason), ) .await; break; } else { let _ = entry_clone .transition_to_ready_with_store(reg_clone.store()) .await; } } };
let _ = futures::future::Abortable::new(runner_future, abort_reg).await; entry_clone.clear_runner().await; });
Ok(())}
pub async fn serve( interrupt_tx: mpsc::Sender<Interrupt>, output: broadcast::Sender<AgentEvent>, v2_event_tx: broadcast::Sender<V2ServerMsg>, snapshot: MetricsSnapshot, memory: MemoryStore, history_window: usize, default_soul: String, registry: Arc<SessionRegistry>, llm: Option<LlmClient>, initial_release: Option<Release>, v2_mode: bool, mut shutdown: watch::Receiver<bool>, shutdown: watch::Receiver<bool>,) -> Result<()> { let listener = TcpListener::bind(WS_BIND_ADDR).await?; tracing::info!(addr = WS_BIND_ADDR, "daemon listening"); serve_listener( listener, interrupt_tx, output, v2_event_tx, snapshot, memory, history_window, default_soul, registry, llm, initial_release, v2_mode, shutdown, ) .await}
pub async fn serve_listener( listener: TcpListener, interrupt_tx: mpsc::Sender<Interrupt>, output: broadcast::Sender<AgentEvent>, v2_event_tx: broadcast::Sender<V2ServerMsg>, snapshot: MetricsSnapshot, memory: MemoryStore, history_window: usize, default_soul: String, registry: Arc<SessionRegistry>, llm: Option<LlmClient>, initial_release: Option<Release>, v2_mode: bool, mut shutdown: watch::Receiver<bool>,) -> Result<()> { loop { let accepted = tokio::select! { _ = shutdown.changed() => break, accepted = listener.accept() => accepted, }; let (stream, _) = accepted?; let tx = interrupt_tx.clone(); let bcast = output.clone(); let v2_bcast = v2_event_tx.clone(); let snapshot = snapshot.clone(); let memory = memory.clone(); let default_soul = default_soul.clone(); let registry = registry.clone(); let shutdown = shutdown.clone(); let llm = llm.clone(); let initial_release = initial_release.clone();
tokio::spawn(async move { if let Err(e) = handle( stream, tx, bcast, v2_bcast, snapshot, memory, history_window, default_soul, registry, llm, initial_release, v2_mode, shutdown, ) .await {10 unmodified lines ws_tx.send(Message::Text(json)).await?; Ok(())}
async fn send_v2_msg(ws_tx: &mut WsSink, msg: &V2ServerMsg) -> Result<()> { let json = serde_json::to_string(msg)?; ws_tx.send(Message::Text(json)).await?; Ok(())}
async fn handle( sock: TcpStream, interrupt_tx: mpsc::Sender<Interrupt>, output: broadcast::Sender<AgentEvent>, v2_bcast: broadcast::Sender<V2ServerMsg>, snapshot: MetricsSnapshot, memory: MemoryStore, history_window: usize, default_soul: String, registry: Arc<SessionRegistry>, llm: Option<LlmClient>, initial_release: Option<Release>, v2_mode: bool, mut shutdown: watch::Receiver<bool>,) -> Result<()> { let mut rx = output.subscribe(); let mut v2_rx = v2_bcast.subscribe(); let ws = accept_async(sock).await?; let (mut ws_tx, mut ws_rx): (WsSink, WsSource) = ws.split();
// push current context window so tui starts with history let turns = memory.recent_turns(history_window).unwrap_or_default(); send_msg(&mut ws_tx, &ServerMsg::History { turns }).await?; let mut subscribed_session: Option<String> = None;
if !v2_mode { let turns = memory.recent_turns(history_window).unwrap_or_default(); send_msg(&mut ws_tx, &ServerMsg::History { turns }).await?; // send last known metrics so status bar is populated immediately let snap = snapshot.read().await.clone(); if let Some(metrics) = snap { send_msg( &mut ws_tx, &ServerMsg::Metrics { turn_count: metrics.turn_count, context_tokens: metrics.context_tokens, watermark: metrics.watermark, prompt_tokens: metrics.prompt_tokens, completion_tokens: metrics.completion_tokens, cached_prompt_tokens: metrics.cached_prompt_tokens, prompt_processing_ms: metrics.prompt_processing_ms, generation_ms: metrics.generation_ms, bridge_processing_ms: metrics.bridge_processing_ms, }, ) .await?;
let snap = snapshot.read().await.clone(); if let Some(metrics) = snap { send_msg( &mut ws_tx, &ServerMsg::Metrics { turn_count: metrics.turn_count, context_tokens: metrics.context_tokens, watermark: metrics.watermark, prompt_tokens: metrics.prompt_tokens, completion_tokens: metrics.completion_tokens, cached_prompt_tokens: metrics.cached_prompt_tokens, prompt_processing_ms: metrics.prompt_processing_ms, generation_ms: metrics.generation_ms, bridge_processing_ms: metrics.bridge_processing_ms, }, ) .await?; } }
loop { tokio::select! { _ = shutdown.changed() => break, v2_ev = v2_rx.recv() => { if let Ok(msg) = v2_ev { if let Some(sub_sess) = &subscribed_session { match &msg { V2ServerMsg::SyncStarted { session_id, .. } | V2ServerMsg::SyncCompleted { session_id, .. } | V2ServerMsg::LiveChunk { session_id, .. } | V2ServerMsg::Ack { session_id, .. } => { if session_id == sub_sess { let _ = send_v2_msg(&mut ws_tx, &msg).await; } } V2ServerMsg::Event { record } => { if &record.session_id == sub_sess { let _ = send_v2_msg(&mut ws_tx, &msg).await; } } _ => { let _ = send_v2_msg(&mut ws_tx, &msg).await; } } } else { let _ = send_v2_msg(&mut ws_tx, &msg).await; } } } ws_msg = ws_rx.next() => { let Some(ws_msg) = ws_msg else { break }; let ws_msg = ws_msg?; let Message::Text(text) = ws_msg else { continue };
let msg: ClientMsg = serde_json::from_str(&text)?; // 1. Check if V2ClientMsg if let Ok(v2_msg) = serde_json::from_str::<V2ClientMsg>(&text) { match v2_msg { V2ClientMsg::Subscribe { session_id, after_seq } => { subscribed_session = Some(session_id.clone()); if let Some(after) = after_seq { let s_id = parse_session_id(&session_id); let events = query_events_after(registry.store(), &s_id, after).await.unwrap_or_default(); let latest = events.last().map(|e| e.seq).unwrap_or(after); let _ = send_v2_msg(&mut ws_tx, &V2ServerMsg::SyncStarted { session_id: session_id.clone(), from_seq: after + 1, to_seq: latest, }).await; for record in events { let _ = send_v2_msg(&mut ws_tx, &V2ServerMsg::Event { record }).await; } let _ = send_v2_msg(&mut ws_tx, &V2ServerMsg::SyncCompleted { session_id: session_id.clone(), latest_seq: latest, }).await; } } V2ClientMsg::SendInput { session_id, source, text, client_request_id } => { let s_id = parse_session_id(&session_id); let source_ref = SourceRef::from_v2_input(&session_id, &source, &text, client_request_id.clone()); match admit_input(®istry, &s_id, source_ref).await { Ok(StoreAdmit::Inserted(seq)) => { let _ = send_v2_msg(&mut ws_tx, &V2ServerMsg::Ack { session_id: session_id.clone(), seq, client_request_id: client_request_id.clone(), duplicate: Some(false), }).await; if let (Some(l), Some(rel)) = (&llm, &initial_release) { let _ = trigger_session_turn( registry.clone(), l.clone(), s_id, rel.clone(), v2_bcast.clone(), ).await; } } Ok(StoreAdmit::Duplicate(seq)) => { let _ = send_v2_msg(&mut ws_tx, &V2ServerMsg::Ack { session_id: session_id.clone(), seq, client_request_id: client_request_id.clone(), duplicate: Some(true), }).await; } Ok(StoreAdmit::Conflict(msg)) => { let _ = send_v2_msg(&mut ws_tx, &V2ServerMsg::Error { code: "conflict".into(), message: msg, }).await; } Ok(StoreAdmit::Error(msg)) => { let _ = send_v2_msg(&mut ws_tx, &V2ServerMsg::Error { code: "admission_rejected".into(), message: msg, }).await; } Err(err) => { let _ = send_v2_msg(&mut ws_tx, &V2ServerMsg::Error { code: "admission_error".into(), message: format!("{err:#}"), }).await; } } } V2ClientMsg::Ack { session_id, seq } => { tracing::debug!(session = %session_id, seq, "client ack received"); } V2ClientMsg::Ping { timestamp } => { let _ = send_v2_msg(&mut ws_tx, &V2ServerMsg::Pong { timestamp }).await; } V2ClientMsg::Interrupt { session_id, .. } => { let s_id = parse_session_id(&session_id); if let Some(entry) = registry.get(&s_id).await { entry.abort_runner().await; } } V2ClientMsg::ResetKernel { session_id } => { let s_id = parse_session_id(&session_id); if let Some(entry) = registry.get(&s_id).await { entry.abort_runner().await; entry.runtime.shutdown().await; } } } continue; }
// 2. Check if legacy ClientMsg let msg: ClientMsg = match serde_json::from_str(&text) { Ok(m) => m, Err(e) => { let _ = send_v2_msg(&mut ws_tx, &V2ServerMsg::Error { code: "malformed_request".into(), message: format!("cannot parse message: {e}"), }).await; continue; } };
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(outcome) => { Ok(event_id) => { let _ = send_msg(&mut ws_tx, &ServerMsg::Admitted { session_id: session_id.to_string(), event_id, }).await; if let Some(seq) = outcome.seq() { let _ = send_msg(&mut ws_tx, &ServerMsg::Admitted { session_id: session_id.to_string(), event_id: seq, }).await; if !v2_mode { let int = Interrupt::message(source, content); let _ = interrupt_tx.send(int).await; } else if let (Some(l), Some(rel)) = (&llm, &initial_release) { let _ = trigger_session_turn( registry.clone(), l.clone(), session_id, rel.clone(), v2_bcast.clone(), ).await; } } else if let StoreAdmit::Conflict(msg) = outcome { let _ = send_msg(&mut ws_tx, &ServerMsg::Error { content: format!("conflict: {msg}"), }).await; } else if let StoreAdmit::Error(msg) = outcome { let _ = send_msg(&mut ws_tx, &ServerMsg::Error { content: format!("admission rejected: {msg}"), }).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 { source, conversation_id,15 unmodified lines timestamp, metadata.clone(), ); match admit_input(®istry, &session_id, source_ref).await { Ok(outcome) => { Ok(event_id) => { let _ = send_msg(&mut ws_tx, &ServerMsg::Admitted { session_id: session_id.to_string(), event_id, }).await; if let Some(seq) = outcome.seq() { let _ = send_msg(&mut ws_tx, &ServerMsg::Admitted { session_id: session_id.to_string(), event_id: seq, }).await; if !v2_mode { let _ = interrupt_tx .send(Interrupt::external_event(ExternalEvent { source, conversation_id, content, author_id, author_name, message_id, timestamp, metadata, })) .await; } else if let (Some(l), Some(rel)) = (&llm, &initial_release) { let _ = trigger_session_turn( registry.clone(), l.clone(), session_id, rel.clone(), v2_bcast.clone(), ).await; } } else if let StoreAdmit::Conflict(msg) = outcome { let _ = send_msg(&mut ws_tx, &ServerMsg::Error { content: format!("conflict: {msg}"), }).await; } else if let StoreAdmit::Error(msg) = outcome { let _ = send_msg(&mut ws_tx, &ServerMsg::Error { content: format!("admission rejected: {msg}"), }).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, content, author_id, author_name, message_id, timestamp, metadata, })) .await?; } ClientMsg::FetchHistory { before_id, limit } => { let limit = limit.min(HISTORY_PAGE); let turns = memory.turns_before(before_id, limit).unwrap_or_default(); send_msg(&mut ws_tx, &ServerMsg::History { turns }).await?; } ClientMsg::Compact => { if !v2_mode { interrupt_tx.send(Interrupt::Compact).await?; interrupt_tx.send(Interrupt::Compact).await?; } } ClientMsg::DebugReflectRequest => { if !v2_mode { interrupt_tx.send(Interrupt::DebugReflectRequest).await?; interrupt_tx.send(Interrupt::DebugReflectRequest).await?; } } ClientMsg::UpdateSoul { content } => { if !v2_mode { interrupt_tx.send(Interrupt::UpdateSoul { content }).await?; interrupt_tx.send(Interrupt::UpdateSoul { content }).await?; } } ClientMsg::Reset => { memory.reset()?; memory.ensure_soul_memory(&default_soul)?; if !v2_mode { interrupt_tx.send(Interrupt::Reset).await?; interrupt_tx.send(Interrupt::Reset).await?; } send_msg( &mut ws_tx, &ServerMsg::Status { content: "database reset".into(),260 unmodified linesExpand all1 unmodified line234567891063 unmodified lines747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211382 unmodified lines19619719819916720020120220320411 unmodified lines21621721821918722022118922222319119219319419519619719819920020120220320420520620720822422522621022722822921221321421521621721821922022122222322422523023123223323423523623723823924024124224324424524624724824925025122722822923023123223325225323425425523723825625725825926026158 unmodified lines32032132232330532432532632732832933016 unmodified lines34734834935033033133235135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539631 unmodified lines1 unmodified line
use anyhow::{bail, Context, Result};use chrono::Utc;use klbr_runtime::protocol::SessionId;pub use klbr_runtime::store::StoreAdmit;use serde::{Deserialize, Serialize};use uuid::Uuid;
use crate::session_registry::SessionRegistry;63 unmodified lines self.metadata = metadata; self }
pub fn from_v2_input( session_id: impl Into<String>, source: impl Into<String>, text: impl Into<String>, client_request_id: Option<String>, ) -> Self { let session_id = session_id.into(); let source = source.into(); let text = text.into(); let source_kind = if source.trim().is_empty() { "websocket".to_string() } else { source }; let (timestamp_ms, mid) = match client_request_id { Some(cid) => (0, cid), None => (Utc::now().timestamp_millis(), Uuid::new_v4().to_string()), }; let dedupe_key = format!("{source_kind}:{session_id}:{mid}"); Self { source_kind, conversation_id: session_id, author_id: None, author_name: None, message_id: Some(mid), timestamp_ms, dedupe_key, content: text, metadata: serde_json::Value::Null, } }
pub fn from_client_message( source: impl Into<String>, content: impl Into<String>, session_id: impl Into<String>,82 unmodified linespub async fn admit_input( registry: &SessionRegistry, session_id: &SessionId, source_ref: SourceRef,) -> Result<i64> {) -> Result<StoreAdmit> { // 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;11 unmodified lines .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! // b) Atomic admission: registry.store().admit_with_status() must land in SQLite before acknowledging or running policy! let payload = source_ref.to_payload(); let event_id = match registry let outcome = 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"); } } }; .admit_with_status(session_id, source_ref.dedupe_key.clone(), payload) .await?;
// c) Non-blocking policy: spawn a background task to evaluate attention policy on the session runtime. // c) Non-blocking policy: spawn a background task to evaluate attention policy on the session runtime only on newly inserted items. // 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 StoreAdmit::Inserted(event_id) = outcome { 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"); 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"); } } 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) // d) Return typed StoreAdmit outcome. Ok(outcome)}
#[cfg(test)]mod tests {58 unmodified lines
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?; let outcome = admit_input(®istry, &session_id, sref.clone()).await?; assert!(matches!(outcome, StoreAdmit::Inserted(_))); let event_id = outcome.seq().unwrap(); assert!(event_id > 0);
// Verify it landed atomically in SQLite inbox table with status pending let pending = store.pending(&session_id, 10).await?;16 unmodified lines 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 outcome1 = admit_input(®istry, &session_id, sref.clone()).await?; let outcome2 = admit_input(®istry, &session_id, sref).await?; assert!(matches!(outcome1, StoreAdmit::Inserted(_))); assert!(matches!(outcome2, StoreAdmit::Duplicate(_))); assert_eq!(outcome1.seq(), outcome2.seq(), "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_conflict_on_changed_payload() -> Result<()> { let temp = tempfile::tempdir()?; let store = klbr_runtime::store::Store::open(temp.path().join("conflict.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 sref1 = SourceRef::new("websocket", session_id.as_str(), "first text") .with_dedupe_key("ws:conv1:conflict-key"); let sref2 = SourceRef::new("websocket", session_id.as_str(), "second text") .with_dedupe_key("ws:conv1:conflict-key");
let outcome1 = admit_input(®istry, &session_id, sref1).await?; assert!(matches!(outcome1, StoreAdmit::Inserted(_)));
let outcome2 = admit_input(®istry, &session_id, sref2).await?; assert!(matches!(outcome2, StoreAdmit::Conflict(_)), "changed payload under same key must return Conflict");
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"))31 unmodified lines