diff --git a/desktop/electron/main.cjs b/desktop/electron/main.cjs index 84aad747..2e9cf861 100644 --- a/desktop/electron/main.cjs +++ b/desktop/electron/main.cjs @@ -93,9 +93,28 @@ ipcMain.handle("nav:session-stacks", async (_event, requestedSessionId) => { method: "session.stacks", params: { sessionId: targetSessionId }, }); - return response.result.stacks; + return response.result; }); +ipcMain.handle( + "nav:session-stack-availability", + async (_event, requestedSessionId) => { + if (!backendUrl) { + throw new Error("chat session is not ready"); + } + const targetSessionId = requestedSessionId || sessionId; + if (!targetSessionId) { + throw new Error("chat session is not ready"); + } + const response = await sendRpc({ + backendUrl, + method: "session.stackAvailability", + params: { sessionId: targetSessionId }, + }); + return response.result; + }, +); + ipcMain.handle("nav:switch-session", async (_event, id) => { if (!backendUrl) { throw new Error("chat session is not ready"); diff --git a/desktop/electron/preload.cjs b/desktop/electron/preload.cjs index 9513fe0c..87a6a021 100644 --- a/desktop/electron/preload.cjs +++ b/desktop/electron/preload.cjs @@ -81,6 +81,13 @@ contextBridge.exposeInMainWorld("nav", { sessionId == null ? undefined : normalizeSessionId(sessionId), ); }, + // Advisory scan for whether stack records are currently retained. + sessionStackAvailability(sessionId) { + return ipcRenderer.invoke( + "nav:session-stack-availability", + sessionId == null ? undefined : normalizeSessionId(sessionId), + ); + }, // Switch the active conversation to an existing session. switchSession(sessionId) { return ipcRenderer.invoke( diff --git a/desktop/electron/renderer/src/App.jsx b/desktop/electron/renderer/src/App.jsx index 4efae5ba..724602a4 100644 --- a/desktop/electron/renderer/src/App.jsx +++ b/desktop/electron/renderer/src/App.jsx @@ -13,6 +13,7 @@ export default function App() { const [modelInfo, setModelInfo] = useState(null); const [stopPending, setStopPending] = useState(false); const [activeView, setActiveView] = useState("chat"); + const [stackAvailable, setStackAvailable] = useState(false); const [stackRefreshKey, setStackRefreshKey] = useState(0); const [newSessionMode, setNewSessionMode] = useState("local"); @@ -24,6 +25,7 @@ export default function App() { const stopSentForActiveRunRef = useRef(false); const stopRequestedRef = useRef(false); const nextMessageIdRef = useRef(0); + const stackAvailabilityRequestRef = useRef(0); const setConnectedState = useCallback((isConnected) => { connectedRef.current = isConnected; @@ -69,6 +71,10 @@ export default function App() { window.setTimeout(refreshStacks, 120); }, [refreshStacks]); + const markStacksUnavailable = useCallback(() => { + setStackAvailable(false); + }, []); + const appendMessage = useCallback( (role, text) => { setMessages((current) => [ @@ -146,6 +152,42 @@ export default function App() { } }, []); + const refreshStackAvailability = useCallback( + async (sessionId, options = {}) => { + if (!window.nav) { + return; + } + const targetSessionId = sessionId ?? activeSessionIdRef.current; + if (!targetSessionId) { + setStackAvailable(false); + return; + } + const requestId = stackAvailabilityRequestRef.current + 1; + stackAvailabilityRequestRef.current = requestId; + if (options.reset) { + setStackAvailable(false); + } + try { + const availability = + await window.nav.sessionStackAvailability(targetSessionId); + if ( + requestId === stackAvailabilityRequestRef.current && + targetSessionId === activeSessionIdRef.current + ) { + setStackAvailable(availability?.available === true); + } + } catch { + if ( + requestId === stackAvailabilityRequestRef.current && + targetSessionId === activeSessionIdRef.current + ) { + setStackAvailable(false); + } + } + }, + [], + ); + const renderBackendStatus = useCallback( (status) => { switch (status.state) { @@ -215,10 +257,12 @@ export default function App() { setRunningState(false); refreshSessions(); refreshModelInfo(status.sessionId); + refreshStackAvailability(status.sessionId, { reset: true }); }, [ refreshModelInfo, refreshSessions, + refreshStackAvailability, renderBackendStatus, setActiveSession, setConnectedState, @@ -277,6 +321,7 @@ export default function App() { setStopRequested(false); refreshSessions(); refreshModelInfo(); + refreshStackAvailability(); refreshStacksAfterTerminalEvent(); break; case "run.failed": @@ -286,6 +331,7 @@ export default function App() { setStopRequested(false); refreshSessions(); refreshModelInfo(); + refreshStackAvailability(); refreshStacksAfterTerminalEvent(); break; default: @@ -296,6 +342,7 @@ export default function App() { appendMessage, refreshModelInfo, refreshSessions, + refreshStackAvailability, refreshStacks, refreshStacksAfterTerminalEvent, setRunningState, @@ -310,10 +357,17 @@ export default function App() { const activateCreatedSession = useCallback( async (sessionId) => { setActiveSession(sessionId); + setStackAvailable(false); await refreshSessions(); refreshModelInfo(sessionId); + refreshStackAvailability(sessionId, { reset: true }); }, - [refreshModelInfo, refreshSessions, setActiveSession], + [ + refreshModelInfo, + refreshSessions, + refreshStackAvailability, + setActiveSession, + ], ); const selectSession = useCallback( @@ -328,16 +382,26 @@ export default function App() { const previousSessionId = activeSessionIdRef.current; clearTranscript(); + setActiveView("chat"); + setStackAvailable(false); setActiveSession(sessionId); try { await window.nav.switchSession(sessionId); refreshModelInfo(sessionId); + refreshStackAvailability(sessionId, { reset: true }); } catch (error) { setActiveSession(previousSessionId); + refreshStackAvailability(previousSessionId, { reset: true }); appendMessage("error", `Could not open session: ${error.message}`); } }, - [appendMessage, clearTranscript, refreshModelInfo, setActiveSession], + [ + appendMessage, + clearTranscript, + refreshModelInfo, + refreshStackAvailability, + setActiveSession, + ], ); const startNewChatInProject = useCallback( @@ -427,11 +491,13 @@ export default function App() { activeView={activeView} connected={connected} sessionId={activeSessionId} + showStacks={stackAvailable} onSelectView={setActiveView} /> {activeView === "stacks" ? ( ) : ( @@ -454,7 +520,13 @@ export default function App() { ); } -function SessionToolbar({ activeView, connected, sessionId, onSelectView }) { +function SessionToolbar({ + activeView, + connected, + sessionId, + showStacks, + onSelectView, +}) { return (
@@ -472,15 +544,17 @@ function SessionToolbar({ activeView, connected, sessionId, onSelectView }) { > Chat - + {showStacks ? ( + + ) : null}
); diff --git a/desktop/electron/renderer/src/components/StacksPage.jsx b/desktop/electron/renderer/src/components/StacksPage.jsx index 0e10abcf..c392bea8 100644 --- a/desktop/electron/renderer/src/components/StacksPage.jsx +++ b/desktop/electron/renderer/src/components/StacksPage.jsx @@ -1,24 +1,32 @@ import { useEffect, useMemo, useState } from "react"; -export default function StacksPage({ sessionId }) { +export default function StacksPage({ onUnavailable, sessionId }) { const [stacks, setStacks] = useState([]); const [loading, setLoading] = useState(false); const [error, setError] = useState(null); + const [unavailableReason, setUnavailableReason] = useState(null); useEffect(() => { let alive = true; if (!sessionId || !window.nav) { setStacks([]); + setUnavailableReason(null); return undefined; } setLoading(true); setError(null); + setUnavailableReason(null); window.nav .sessionStacks(sessionId) - .then((nextStacks) => { + .then((result) => { if (alive) { - setStacks(Array.isArray(nextStacks) ? nextStacks : []); + const nextStacks = Array.isArray(result?.stacks) ? result.stacks : []; + setStacks(nextStacks); + setUnavailableReason(result?.unavailableReason ?? null); + if (nextStacks.length === 0 && result?.unavailableReason) { + onUnavailable?.(result.unavailableReason); + } } }) .catch((fetchError) => { @@ -35,7 +43,7 @@ export default function StacksPage({ sessionId }) { return () => { alive = false; }; - }, [sessionId]); + }, [onUnavailable, sessionId]); const orderedStacks = useMemo( () => [...stacks].sort((left, right) => left.sequence - right.sequence), @@ -74,7 +82,7 @@ export default function StacksPage({ sessionId }) { {orderedStacks.length === 0 ? ( - + ) : (
    {orderedStacks.map((stack) => ( @@ -154,6 +162,19 @@ function EmptyStacks({ text }) { return
    {text}
    ; } +function emptyStackText(reason) { + switch (reason) { + case "trimmed_or_missing": + return "Stack records for this session were no longer available. The stack log is capped at 800MB, so older records may have been trimmed."; + case "stack_store_unavailable": + return "Stack storage is unavailable for this backend run."; + case "stack_store_error": + return "Stack records could not be read from the local stack log."; + default: + return "No model calls captured for this live session yet"; + } +} + function defaultOpen(layer) { return [ "system_prompt", diff --git a/desktop/electron/renderer/window.d.ts b/desktop/electron/renderer/window.d.ts index 5d88df47..188010d9 100644 --- a/desktop/electron/renderer/window.d.ts +++ b/desktop/electron/renderer/window.d.ts @@ -44,8 +44,8 @@ declare global { thinking?: string | null; tokenUsage?: { used: number; contextWindow: number } | null; }>; - sessionStacks(sessionId?: string): Promise< - Array<{ + sessionStacks(sessionId?: string): Promise<{ + stacks: Array<{ id: string; runId: string; sequence: number; @@ -61,8 +61,12 @@ declare global { text?: string; json?: unknown; }>; - }> - >; + }>; + unavailableReason?: string; + }>; + sessionStackAvailability(sessionId?: string): Promise<{ + available: boolean; + }>; switchSession(sessionId: string): Promise; newSession( workspaceRoot?: string | null, diff --git a/research/stack-persistence.md b/research/stack-persistence.md new file mode 100644 index 00000000..a1d2ffef --- /dev/null +++ b/research/stack-persistence.md @@ -0,0 +1,49 @@ +# Stack Persistence + +The stack view should distinguish between data that can be reconstructed from +stored turns and data that must be captured at the live model-call boundary. + +## Replay Boundary + +These stack details cannot be derived exactly from a DB replay today: + +- Raw provider response JSON. Stored assistant and tool turns keep the normalized + content, reasoning text, and tool calls/results, but not the original response + body. +- Provider request/response metadata: request id, response id, HTTP status, + provider model id, provider error payloads, and similar transport details. +- Exact historical provider request payload. Replay can rebuild an approximate + request, but system prompt assembly, context-file contents, tool schemas, + config, model settings, and adapter behavior may have changed. +- Exact system prompt and project-context snapshot loaded for that call. +- Exact tool definitions advertised to the provider for that call. +- Per-model-call timing. The DB stores run timing, not individual model-call + timing inside multi-call runs. +- Per-model-call token usage, source, and confidence. Session token columns are + aggregate counters. +- Non-obvious finish reason. Tool-call finishes are inferable from tool calls, + but plain replies do not preserve stop versus length versus provider-specific + finish reasons. +- Cancel-after-response hidden output. A model response can exist in the stack + even when cancellation prevents the assistant turn from being emitted. +- Queued steering that is dropped on cancel before it is drained and persisted. +- Stack-specific call id, status detail, and error strings. + +## Raw Provider Payloads + +Normalization is intentionally lossy. The OpenAI-compatible parser keeps only +the fields required by `ModelResponse`: + +- assistant content +- assistant reasoning content when exposed as `reasoning_content` +- tool call id, function name, and function arguments +- normalized finish reason +- selected token usage counters + +Everything else in the raw request/response is lost unless the `ProviderCallTrace` +is captured separately: provider object ids, created timestamps, choices beyond +the first, role/name annotations outside the normalized message shape, logprobs, +refusal or annotation payloads, service tier, system fingerprint, arbitrary +provider extensions, full usage breakdowns not mapped by `parse_token_usage`, +HTTP status, request id, response id, raw error bodies, and exact provider +request JSON. diff --git a/src/bin/nav-local-backend.rs b/src/bin/nav-local-backend.rs index 39e50247..0d218a80 100644 --- a/src/bin/nav-local-backend.rs +++ b/src/bin/nav-local-backend.rs @@ -5,7 +5,7 @@ use std::process::ExitCode; use std::sync::Arc; use std::time::{Instant, SystemTime, UNIX_EPOCH}; -use nav::{ModelChoice, SessionStore, Storage}; +use nav::{ModelChoice, SessionStore, StackStore, Storage}; use serde_json::{Value, json}; const STARTUP_TRACE_PREFIX: &str = "nav startup trace "; @@ -114,6 +114,45 @@ fn run() -> io::Result<()> { tracing::error!(%error, "storage unavailable, sessions will not persist"); } } + + let stacks_override = std::env::var("NAV_STACKS_PATH") + .ok() + .filter(|path| !path.is_empty()); + let stacks_location = stacks_override.as_deref().unwrap_or("~/.nav/stacks.jsonl"); + let stacks_source = if stacks_override.is_some() { + "override" + } else { + "default" + }; + let stacks_started = Instant::now(); + let opened_stacks = match &stacks_override { + Some(path) => StackStore::open(Path::new(path), nav::DEFAULT_STACKS_MAX_BYTES), + None => StackStore::open_default(), + }; + let stacks_duration_ms = elapsed_ms(stacks_started); + match opened_stacks { + Ok(stack_store) => { + trace.event( + "backend.stack_store.opened", + json!({ + "duration_ms": stacks_duration_ms, + "location": stacks_source, + }), + ); + tracing::info!(location = %stacks_location, "persisting model-call stacks"); + store = store.with_stack_store(Arc::new(stack_store)); + } + Err(error) => { + trace.event( + "backend.stack_store.failed", + json!({ + "duration_ms": stacks_duration_ms, + "location": stacks_source, + }), + ); + tracing::error!(%error, "stack store unavailable, stacks will not persist"); + } + } let store = Arc::new(store); trace.event("backend.ready", json!({})); diff --git a/src/lib.rs b/src/lib.rs index 8cddd53b..a73448f5 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -27,6 +27,7 @@ mod context; pub mod logging; mod model; mod session; +mod stack_store; mod stacks; mod storage; mod system_prompt; @@ -41,6 +42,9 @@ pub use model::{ ModelResponse, OpenAiConfig, OpenAiModel, Role, TokenBudgetInfo, ToolCall, ToolDef, }; pub use session::{Event, SendError, SessionStore, Subscription}; +pub use stack_store::{ + DEFAULT_STACKS_MAX_BYTES, StackAvailability, StackQueryResult, StackStore, StackStoreError, +}; pub use stacks::{ModelCallStack, StackEntry, StackLayer}; pub use storage::{SessionSummary, Storage, StorageError}; pub use system_prompt::{ @@ -190,7 +194,20 @@ fn handle_rpc(stream: &mut TcpStream, store: &Arc, body: &str) -> .and_then(Value::as_str); match session_id { Some(session_id) => match store.stacks(session_id) { - Some(stacks) => write_rpc_result(stream, &id, json!({ "stacks": stacks })), + Some(result) => write_rpc_result(stream, &id, json!(result)), + None => write_rpc_error(stream, &id, "unknown session"), + }, + None => write_rpc_error(stream, &id, "missing parameter: sessionId"), + } + } + Some("session.stackAvailability") => { + let session_id = request + .get("params") + .and_then(|p| p.get("sessionId")) + .and_then(Value::as_str); + match session_id { + Some(session_id) => match store.stack_availability(session_id) { + Some(availability) => write_rpc_result(stream, &id, json!(availability)), None => write_rpc_error(stream, &id, "unknown session"), }, None => write_rpc_error(stream, &id, "missing parameter: sessionId"), diff --git a/src/session.rs b/src/session.rs index 56464179..a579dbd3 100644 --- a/src/session.rs +++ b/src/session.rs @@ -18,6 +18,7 @@ use uuid::Uuid; use crate::agent::{Agent, AgentRunError, AgentRunSink, RunStop, TurnContinuation}; use crate::context::{ContextAssembler, TurnHistory}; use crate::model::{ChatMessage, ChatModel, ModelInfo, Role, ToolCall}; +use crate::stack_store::{StackAvailability, StackQueryResult, StackStore, StackStoreError}; use crate::stacks::ModelCallStack; use crate::storage::{ SessionSummary, Storage, StorageError, project_root_to_string, workspace_root_to_string, @@ -83,7 +84,7 @@ struct Session { turns: TurnHistory, events: Vec, subscribers: Vec>, - stacks: Vec, + next_stack_sequence: u64, /// The in-flight run, set while a run is executing. `None` when idle. active_run: Option, /// Latest model-call context usage for this session. @@ -110,7 +111,7 @@ impl Session { turns: TurnHistory::new(), events: Vec::new(), subscribers: Vec::new(), - stacks: Vec::new(), + next_stack_sequence: 0, active_run: None, token_usage: None, } @@ -172,6 +173,7 @@ pub struct SessionStore { agent: Agent, context_assembler: ContextAssembler, storage: Option>, + stack_store: Option>, /// Identifier of the active model, tagged onto persisted assistant turns. model_id: Option, /// Renderer-facing model metadata shown in the app's composer. @@ -185,6 +187,7 @@ impl SessionStore { agent: Agent::new(model), context_assembler: ContextAssembler::new(), storage: None, + stack_store: None, model_id: None, model_info: ModelInfo { label: "unknown model".to_owned(), @@ -201,6 +204,12 @@ impl SessionStore { self } + /// Attach bounded JSONL storage for debug model-call stack snapshots. + pub fn with_stack_store(mut self, stack_store: Arc) -> Self { + self.stack_store = Some(stack_store); + self + } + /// Record which model produced assistant replies (persisted on each turn). pub fn with_model_id(mut self, model_id: Option) -> Self { self.model_id = model_id; @@ -620,13 +629,47 @@ impl SessionStore { .map(|session| session.events.clone()) } - /// Snapshot of the model-call stacks captured for a live session. - pub fn stacks(&self, session_id: &str) -> Option> { - self.sessions - .lock() - .unwrap() - .get(session_id) - .map(|session| recent_model_call_stacks(&session.stacks)) + /// Whether the stack log currently contains any record for this session. + /// + /// This is advisory. Records can still be trimmed between this check and the + /// later `stacks` call; callers should handle an empty stack result. + pub fn stack_availability(&self, session_id: &str) -> Option { + if !self.sessions.lock().unwrap().contains_key(session_id) { + return None; + } + match &self.stack_store { + Some(stack_store) => match stack_store.availability(session_id) { + Ok(availability) => Some(availability), + Err(error) => { + tracing::error!(%session_id, %error, "failed to scan stack availability"); + Some(StackAvailability { available: false }) + } + }, + None => Some(StackAvailability { available: false }), + } + } + + /// Snapshot of the model-call stacks currently retained for a session. + pub fn stacks(&self, session_id: &str) -> Option { + if !self.sessions.lock().unwrap().contains_key(session_id) { + return None; + } + match &self.stack_store { + Some(stack_store) => match stack_store.stacks(session_id, MAX_STACKS_PER_SESSION) { + Ok(result) => Some(result), + Err(error) => { + tracing::error!(%session_id, %error, "failed to load stacks"); + Some(StackQueryResult { + stacks: Vec::new(), + unavailable_reason: Some("stack_store_error".to_owned()), + }) + } + }, + None => Some(StackQueryResult { + stacks: Vec::new(), + unavailable_reason: Some("stack_store_unavailable".to_owned()), + }), + } } fn with_default_workspace(&self, sessions: Vec) -> Vec { @@ -856,18 +899,17 @@ impl AgentRunSink for SessionRunSink<'_> { } fn model_call_stack(&mut self, mut stack: ModelCallStack) -> Result<(), Self::Error> { - self.store.with_session(self.session_id, |session| { - let next_sequence = session - .stacks - .last() - .map_or(0, |last| last.sequence.saturating_add(1)); - if session.stacks.len() >= MAX_STACKS_PER_SESSION { - let remove_count = session.stacks.len() + 1 - MAX_STACKS_PER_SESSION; - session.stacks.drain(0..remove_count); - } - stack.sequence = next_sequence; - session.stacks.push(stack); - }) + let sequence = self.store.with_session(self.session_id, |session| { + let sequence = session.next_stack_sequence; + session.next_stack_sequence = session.next_stack_sequence.saturating_add(1); + sequence + })?; + stack.sequence = sequence; + + if let Some(stack_store) = &self.store.stack_store { + log_stack_store("append_stack", stack_store.append(self.session_id, &stack)); + } + Ok(()) } } @@ -909,11 +951,6 @@ fn drain_steer_locked(session: &mut Session, run_id: &str) -> Vec { texts } -fn recent_model_call_stacks(stacks: &[ModelCallStack]) -> Vec { - let start = stacks.len().saturating_sub(MAX_STACKS_PER_SESSION); - stacks[start..].to_vec() -} - fn new_id() -> String { Uuid::now_v7().to_string() } @@ -926,34 +963,8 @@ fn log_storage(operation: &str, result: Result<(), crate::storage::StorageError> } } -#[cfg(test)] -mod tests { - use super::*; - use crate::stacks::StackLayer; - - fn stack(sequence: u64) -> ModelCallStack { - ModelCallStack { - id: format!("call-{sequence}"), - run_id: "run".to_owned(), - sequence, - status: "completed".to_owned(), - started_at_ms: sequence, - duration_ms: 1.0, - layers: Vec::::new(), - } - } - - #[test] - fn recent_model_call_stacks_returns_only_the_retained_tail() { - let stacks: Vec<_> = (0..MAX_STACKS_PER_SESSION as u64 + 3).map(stack).collect(); - - let recent = recent_model_call_stacks(&stacks); - - assert_eq!(recent.len(), MAX_STACKS_PER_SESSION); - assert_eq!(recent.first().map(|stack| stack.sequence), Some(3)); - assert_eq!( - recent.last().map(|stack| stack.sequence), - Some(MAX_STACKS_PER_SESSION as u64 + 2) - ); +fn log_stack_store(operation: &str, result: Result<(), StackStoreError>) { + if let Err(error) = result { + tracing::error!(%operation, %error, "failed to persist stack"); } } diff --git a/src/stack_store.rs b/src/stack_store.rs new file mode 100644 index 00000000..ac27a5a5 --- /dev/null +++ b/src/stack_store.rs @@ -0,0 +1,378 @@ +//! Append-only JSONL storage for model-call stacks. +//! +//! Chat/session storage remains in SQLite, but stack snapshots are large debug +//! payloads. Keeping them in a bounded JSONL file lets old records age out +//! independently from the durable conversation history. + +use std::fmt; +use std::fs::{self, File, OpenOptions}; +use std::io::{BufRead, BufReader, Write}; +use std::path::{Path, PathBuf}; +use std::sync::Mutex; +use std::time::{SystemTime, UNIX_EPOCH}; + +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +use crate::stacks::ModelCallStack; + +pub const DEFAULT_STACKS_MAX_BYTES: u64 = 800 * 1024 * 1024; +const STACK_RECORD_SCHEMA_VERSION: u32 = 1; + +#[derive(Debug)] +pub struct StackStoreError(String); + +impl fmt::Display for StackStoreError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "stack store error: {}", self.0) + } +} + +impl std::error::Error for StackStoreError {} + +impl From for StackStoreError { + fn from(error: std::io::Error) -> Self { + Self(error.to_string()) + } +} + +impl From for StackStoreError { + fn from(error: serde_json::Error) -> Self { + Self(error.to_string()) + } +} + +#[derive(Clone, Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct StackAvailability { + pub available: bool, +} + +#[derive(Clone, Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct StackQueryResult { + pub stacks: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + pub unavailable_reason: Option, +} + +#[derive(Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +struct StackRecord { + schema_version: u32, + session_id: String, + run_id: String, + written_at_ms: u64, + stack: ModelCallStack, +} + +pub struct StackStore { + path: PathBuf, + max_bytes: u64, + writer: Mutex<()>, +} + +impl StackStore { + pub fn open(path: &Path, max_bytes: u64) -> Result { + if max_bytes == 0 { + return Err(StackStoreError( + "max stack log size must be greater than zero".to_owned(), + )); + } + if let Some(parent) = path.parent() { + fs::create_dir_all(parent).map_err(|error| { + StackStoreError(format!("cannot create {}: {error}", parent.display())) + })?; + } + let store = Self { + path: path.to_path_buf(), + max_bytes, + writer: Mutex::new(()), + }; + store.compact_existing_file()?; + Ok(store) + } + + pub fn open_default() -> Result { + let home = std::env::var("HOME") + .or_else(|_| std::env::var("USERPROFILE")) + .map_err(|_| StackStoreError("cannot determine home directory".to_owned()))?; + Self::open( + &PathBuf::from(home).join(".nav").join("stacks.jsonl"), + DEFAULT_STACKS_MAX_BYTES, + ) + } + + pub fn append(&self, session_id: &str, stack: &ModelCallStack) -> Result<(), StackStoreError> { + let record = StackRecord { + schema_version: STACK_RECORD_SCHEMA_VERSION, + session_id: session_id.to_owned(), + run_id: stack.run_id.clone(), + written_at_ms: now_ms(), + stack: stack.clone(), + }; + let mut line = serde_json::to_vec(&record)?; + line.push(b'\n'); + + if line.len() as u64 > self.max_bytes { + return Err(StackStoreError(format!( + "stack record is {} bytes, exceeding max {} bytes", + line.len(), + self.max_bytes + ))); + } + + let _guard = self.writer.lock().unwrap(); + if let Some(parent) = self.path.parent() { + fs::create_dir_all(parent)?; + } + let current_len = fs::metadata(&self.path).map(|meta| meta.len()).unwrap_or(0); + if current_len + line.len() as u64 > self.max_bytes { + self.compact_for_append(&line)?; + return Ok(()); + } + + let mut file = OpenOptions::new() + .create(true) + .append(true) + .open(&self.path)?; + file.write_all(&line)?; + Ok(()) + } + + pub fn availability(&self, session_id: &str) -> Result { + Ok(StackAvailability { + available: self.has_session_record(session_id)?, + }) + } + + pub fn stacks( + &self, + session_id: &str, + limit: usize, + ) -> Result { + if limit == 0 { + return Ok(StackQueryResult { + stacks: Vec::new(), + unavailable_reason: None, + }); + } + + let mut stacks = Vec::new(); + let Ok(file) = File::open(&self.path) else { + return Ok(StackQueryResult { + stacks, + unavailable_reason: Some("trimmed_or_missing".to_owned()), + }); + }; + + for line in BufReader::new(file).lines() { + let line = line?; + let Ok(record) = serde_json::from_str::(&line) else { + continue; + }; + if record.schema_version != STACK_RECORD_SCHEMA_VERSION + || record.session_id != session_id + { + continue; + } + stacks.push(record.stack); + if stacks.len() > limit { + stacks.remove(0); + } + } + + if stacks.is_empty() { + return Ok(StackQueryResult { + stacks, + unavailable_reason: Some("trimmed_or_missing".to_owned()), + }); + } + + Ok(StackQueryResult { + stacks, + unavailable_reason: None, + }) + } + + fn has_session_record(&self, session_id: &str) -> Result { + let Ok(file) = File::open(&self.path) else { + return Ok(false); + }; + + for line in BufReader::new(file).lines() { + let line = line?; + let Ok(record) = serde_json::from_str::(&line) else { + continue; + }; + if record.schema_version == STACK_RECORD_SCHEMA_VERSION + && record.session_id == session_id + { + return Ok(true); + } + } + Ok(false) + } + + fn compact_for_append(&self, line: &[u8]) -> Result<(), StackStoreError> { + let selected = self.select_newest_records(line.len() as u64)?; + self.rewrite_with_records(selected, Some(line)) + } + + fn compact_existing_file(&self) -> Result<(), StackStoreError> { + let current_len = fs::metadata(&self.path).map(|meta| meta.len()).unwrap_or(0); + if current_len <= self.max_bytes { + return Ok(()); + } + let selected = self.select_newest_records(0)?; + self.rewrite_with_records(selected, None) + } + + fn select_newest_records(&self, reserved_bytes: u64) -> Result>, StackStoreError> { + let mut selected = Vec::new(); + let mut selected_len = reserved_bytes; + let bytes = fs::read(&self.path).unwrap_or_default(); + for chunk in bytes.split(|byte| *byte == b'\n').rev() { + if chunk.is_empty() || serde_json::from_slice::(chunk).is_err() { + continue; + } + let candidate_len = chunk.len() as u64 + 1; + if selected_len + candidate_len > self.max_bytes { + continue; + } + selected.push(chunk.to_vec()); + selected_len += candidate_len; + } + selected.reverse(); + Ok(selected) + } + + fn rewrite_with_records( + &self, + records: Vec>, + appended_line: Option<&[u8]>, + ) -> Result<(), StackStoreError> { + let temp_path = temp_path_next_to(&self.path); + { + let mut temp = File::create(&temp_path)?; + for chunk in records { + temp.write_all(&chunk)?; + temp.write_all(b"\n")?; + } + if let Some(line) = appended_line { + temp.write_all(line)?; + } + temp.flush()?; + temp.sync_all()?; + } + replace_file(&temp_path, &self.path)?; + Ok(()) + } +} + +fn temp_path_next_to(path: &Path) -> PathBuf { + // Keep the temp file beside the target so successful renames stay on the + // same filesystem. Some platforms still fail when replacing an existing + // target, so `replace_file` has a verified copy fallback. + path.with_extension(format!("jsonl.tmp-{}", Uuid::now_v7())) +} + +fn replace_file(temp_path: &Path, target_path: &Path) -> Result<(), StackStoreError> { + match fs::rename(temp_path, target_path) { + Ok(()) => Ok(()), + Err(rename_error) => { + tracing::warn!( + temp_path = %temp_path.display(), + target_path = %target_path.display(), + %rename_error, + "stack log rename failed; falling back to copy replacement" + ); + replace_file_by_copy(temp_path, target_path, &rename_error) + } + } +} + +fn replace_file_by_copy( + temp_path: &Path, + target_path: &Path, + rename_error: &std::io::Error, +) -> Result<(), StackStoreError> { + let bytes = fs::read(temp_path).map_err(|error| { + StackStoreError(format!( + "rename failed ({rename_error}); fallback could not read {}: {error}", + temp_path.display() + )) + })?; + + { + let mut target = OpenOptions::new() + .create(true) + .write(true) + .truncate(true) + .open(target_path) + .map_err(|error| { + StackStoreError(format!( + "rename failed ({rename_error}); fallback could not open {}: {error}", + target_path.display() + )) + })?; + target.write_all(&bytes).map_err(|error| { + StackStoreError(format!( + "rename failed ({rename_error}); fallback could not write {}: {error}", + target_path.display() + )) + })?; + target.sync_all().map_err(|error| { + StackStoreError(format!( + "rename failed ({rename_error}); fallback could not sync {}: {error}", + target_path.display() + )) + })?; + } + + let written_len = fs::metadata(target_path) + .map_err(|error| { + StackStoreError(format!( + "rename failed ({rename_error}); fallback could not stat {}: {error}", + target_path.display() + )) + })? + .len(); + if written_len != bytes.len() as u64 { + return Err(StackStoreError(format!( + "rename failed ({rename_error}); fallback wrote {} bytes, expected {} bytes", + written_len, + bytes.len() + ))); + } + + sync_parent_dir(target_path, rename_error); + fs::remove_file(temp_path).map_err(|error| { + StackStoreError(format!( + "rename failed ({rename_error}); fallback wrote {}, but could not remove {}: {error}", + target_path.display(), + temp_path.display() + )) + }) +} + +fn sync_parent_dir(path: &Path, rename_error: &std::io::Error) { + let Some(parent) = path.parent() else { + return; + }; + if let Err(error) = File::open(parent).and_then(|directory| directory.sync_all()) { + tracing::debug!( + parent = %parent.display(), + %rename_error, + %error, + "stack log fallback could not fsync parent directory" + ); + } +} + +fn now_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as u64 +} diff --git a/src/stacks.rs b/src/stacks.rs index 8cb3fb2f..2ba9f849 100644 --- a/src/stacks.rs +++ b/src/stacks.rs @@ -4,7 +4,7 @@ //! layered instead of being only a raw JSON blob: each layer names what was //! available, how it was assembled, and what state moved forward. -use serde::Serialize; +use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use crate::model::{ @@ -13,7 +13,7 @@ use crate::model::{ use crate::system_prompt::ContextFile; use crate::tokens::{TokenCountConfidence, TokenCountSource, TokenUsage}; -#[derive(Clone, Debug, Serialize)] +#[derive(Clone, Debug, Deserialize, Serialize)] #[serde(rename_all = "camelCase")] pub struct ModelCallStack { pub id: String, @@ -25,7 +25,7 @@ pub struct ModelCallStack { pub layers: Vec, } -#[derive(Clone, Debug, Serialize)] +#[derive(Clone, Debug, Deserialize, Serialize)] #[serde(rename_all = "camelCase")] pub struct StackLayer { pub kind: String, @@ -39,7 +39,7 @@ pub struct StackLayer { pub json: Option, } -#[derive(Clone, Debug, Serialize)] +#[derive(Clone, Debug, Deserialize, Serialize)] #[serde(rename_all = "camelCase")] pub struct StackEntry { pub label: String, diff --git a/tests/local_backend.rs b/tests/local_backend.rs index 0f1a117f..d0a96ce1 100644 --- a/tests/local_backend.rs +++ b/tests/local_backend.rs @@ -7,7 +7,7 @@ use std::thread; use std::time::{Duration, Instant}; use std::{env, fs}; -use nav::{MockModel, ModelInfo, SessionStore, Storage}; +use nav::{MockModel, ModelInfo, SessionStore, StackStore, Storage}; use serde_json::{Value, json}; /// An in-process backend bound to an ephemeral loopback port, driven over raw @@ -15,6 +15,7 @@ use serde_json::{Value, json}; /// mock model. struct TestBackend { address: String, + _stack_dir: TempDir, } struct TempDir { @@ -37,10 +38,22 @@ impl Drop for TempDir { impl TestBackend { fn start() -> Self { - Self::start_with(SessionStore::new(Arc::new(MockModel::new()))) + let stack_dir = TempDir::new("stack_store"); + let stack_store = Arc::new( + StackStore::open(&stack_dir.path.join("stacks.jsonl"), 1024 * 1024) + .expect("open stack store"), + ); + Self::start_with_temp( + SessionStore::new(Arc::new(MockModel::new())).with_stack_store(stack_store), + stack_dir, + ) } fn start_with(store: SessionStore) -> Self { + Self::start_with_temp(store, TempDir::new("backend")) + } + + fn start_with_temp(store: SessionStore, stack_dir: TempDir) -> Self { let store = Arc::new(store); let listener = TcpListener::bind("127.0.0.1:0").expect("bind loopback"); let address = listener.local_addr().expect("read local addr").to_string(); @@ -49,7 +62,10 @@ impl TestBackend { let _ = nav::serve(listener, store); }); - Self { address } + Self { + address, + _stack_dir: stack_dir, + } } fn connect(&self) -> TcpStream { @@ -514,6 +530,41 @@ fn session_stacks_rpc_returns_captured_model_calls() { .any(|layer| layer["kind"] == "normalized_response"), "normalized response layer should be present: {response}" ); + + let availability = json!({ + "jsonrpc": "2.0", + "id": "stack-availability", + "method": "session.stackAvailability", + "params": { "sessionId": session_id }, + }); + let response = backend.rpc(&availability.to_string()); + assert_eq!( + response["result"]["available"], true, + "stack availability should reflect the retained JSONL record: {response}" + ); +} + +#[test] +fn session_stacks_rpc_reports_when_stack_records_are_unavailable() { + let backend = TestBackend::start_with(SessionStore::new(Arc::new(MockModel::new()))); + let session_id = backend.create_session(); + + let request = json!({ + "jsonrpc": "2.0", + "id": "stacks", + "method": "session.stacks", + "params": { "sessionId": session_id }, + }); + let response = backend.rpc(&request.to_string()); + + assert_eq!( + response["result"]["stacks"].as_array().map(Vec::len), + Some(0) + ); + assert_eq!( + response["result"]["unavailableReason"], + "stack_store_unavailable" + ); } #[test] diff --git a/tests/session.rs b/tests/session.rs index 202d2b05..ce15d891 100644 --- a/tests/session.rs +++ b/tests/session.rs @@ -3,7 +3,7 @@ use std::sync::{Arc, Mutex}; use nav::{ ChatMessage, ChatModel, Event, MockModel, ModelContext, ModelError, ModelResponse, - SessionStore, Storage, TokenUsage, ToolDef, + SessionStore, StackStore, Storage, TokenUsage, ToolDef, }; #[test] @@ -140,12 +140,23 @@ fn provider_token_usage_is_recorded_for_the_session() { #[test] fn model_call_stacks_capture_context_response_and_carried_state() { - let store = SessionStore::new(Arc::new(RecordingModel::new())); + let path = + std::env::temp_dir().join(format!("nav_session_stacks_{}.jsonl", uuid::Uuid::now_v7())); + let stack_store = Arc::new(StackStore::open(&path, 1024 * 1024).expect("open stack store")); + let store = SessionStore::new(Arc::new(RecordingModel::new())).with_stack_store(stack_store); let session_id = store.create_session(); store.send_message(&session_id, "show the stack").unwrap(); - let stacks = store.stacks(&session_id).expect("the session exists"); + assert!( + store + .stack_availability(&session_id) + .expect("the session exists") + .available + ); + let result = store.stacks(&session_id).expect("the session exists"); + assert_eq!(result.unavailable_reason, None); + let stacks = result.stacks; assert_eq!(stacks.len(), 1); let stack = &stacks[0]; assert_eq!(stack.sequence, 0); @@ -176,6 +187,8 @@ fn model_call_stacks_capture_context_response_and_carried_state() { "carried-forward layer should include user plus assistant state: {:?}", layer("carried_forward") ); + + let _ = std::fs::remove_file(path); } #[test] diff --git a/tests/stack_store.rs b/tests/stack_store.rs new file mode 100644 index 00000000..d06f4f3b --- /dev/null +++ b/tests/stack_store.rs @@ -0,0 +1,130 @@ +use std::fs; + +use nav::{ModelCallStack, StackLayer, StackStore}; + +fn stack(id: &str, sequence: u64) -> ModelCallStack { + ModelCallStack { + id: id.to_owned(), + run_id: format!("run-{id}"), + sequence, + status: "completed".to_owned(), + started_at_ms: sequence, + duration_ms: 1.0, + layers: vec![StackLayer { + kind: "metadata".to_owned(), + title: "Metadata".to_owned(), + status: "available".to_owned(), + summary: format!("stack {id}"), + entries: Vec::new(), + text: None, + json: None, + }], + } +} + +#[test] +fn stack_store_appends_and_reads_session_records() { + let path = std::env::temp_dir().join(format!("nav_stack_store_{}.jsonl", uuid::Uuid::now_v7())); + let store = StackStore::open(&path, 1024 * 1024).expect("open stack store"); + + store.append("session-a", &stack("a1", 0)).unwrap(); + store.append("session-b", &stack("b1", 0)).unwrap(); + store.append("session-a", &stack("a2", 1)).unwrap(); + + assert!(store.availability("session-a").unwrap().available); + assert!(!store.availability("session-c").unwrap().available); + + let result = store.stacks("session-a", 256).unwrap(); + assert_eq!(result.unavailable_reason, None); + assert_eq!( + result + .stacks + .iter() + .map(|stack| stack.id.as_str()) + .collect::>(), + ["a1", "a2"] + ); + assert_eq!( + result + .stacks + .iter() + .map(|stack| stack.sequence) + .collect::>(), + [0, 1] + ); + + let _ = fs::remove_file(path); +} + +#[test] +fn stack_store_returns_empty_success_when_limit_is_zero() { + let path = std::env::temp_dir().join(format!( + "nav_stack_store_zero_limit_{}.jsonl", + uuid::Uuid::now_v7() + )); + let store = StackStore::open(&path, 1024 * 1024).expect("open stack store"); + + store.append("session", &stack("call-1", 0)).unwrap(); + + let result = store.stacks("session", 0).unwrap(); + assert!(result.stacks.is_empty()); + assert_eq!(result.unavailable_reason, None); + + let _ = fs::remove_file(path); +} + +#[test] +fn stack_store_compacts_to_newest_valid_records_under_the_cap() { + let path = std::env::temp_dir().join(format!( + "nav_stack_store_compact_{}.jsonl", + uuid::Uuid::now_v7() + )); + let store = StackStore::open(&path, 900).expect("open stack store"); + + for index in 0..12 { + store + .append("session", &stack(&format!("call-{index}"), index)) + .unwrap(); + } + + let size = fs::metadata(&path).unwrap().len(); + assert!(size <= 900, "stack log should stay capped, got {size}"); + + let result = store.stacks("session", 256).unwrap(); + assert!(result.stacks.len() < 12, "old records should be trimmed"); + assert_eq!( + result.stacks.last().map(|stack| stack.id.as_str()), + Some("call-11") + ); + + let _ = fs::remove_file(path); +} + +#[test] +fn stack_store_compacts_an_existing_file_on_open() { + let path = std::env::temp_dir().join(format!( + "nav_stack_store_open_compact_{}.jsonl", + uuid::Uuid::now_v7() + )); + let seed = StackStore::open(&path, 10 * 1024).expect("open seed stack store"); + for index in 0..12 { + seed.append("session", &stack(&format!("call-{index}"), index)) + .unwrap(); + } + drop(seed); + + let store = StackStore::open(&path, 900).expect("reopen with smaller cap"); + + let size = fs::metadata(&path).unwrap().len(); + assert!( + size <= 900, + "existing stack log should be capped, got {size}" + ); + let result = store.stacks("session", 256).unwrap(); + assert_eq!( + result.stacks.last().map(|stack| stack.id.as_str()), + Some("call-11") + ); + + let _ = fs::remove_file(path); +}