diff --git a/migrations/20260726180000_add_compaction_summary/down.sql b/migrations/20260726180000_add_compaction_summary/down.sql new file mode 100644 index 0000000..0b3f035 --- /dev/null +++ b/migrations/20260726180000_add_compaction_summary/down.sql @@ -0,0 +1 @@ +ALTER TABLE sessions DROP COLUMN compacted_summary; diff --git a/migrations/20260726180000_add_compaction_summary/up.sql b/migrations/20260726180000_add_compaction_summary/up.sql new file mode 100644 index 0000000..56fe3b5 --- /dev/null +++ b/migrations/20260726180000_add_compaction_summary/up.sql @@ -0,0 +1 @@ +ALTER TABLE sessions ADD COLUMN compacted_summary TEXT; diff --git a/src/app.rs b/src/app.rs index d4880fd..d100468 100644 --- a/src/app.rs +++ b/src/app.rs @@ -94,6 +94,8 @@ pub(crate) struct App { #[cfg(target_os = "macos")] active_generation: Option, #[cfg(target_os = "macos")] + active_compaction: Option, + #[cfg(target_os = "macos")] agent_tools: Option<(i32, Arc>)>, #[cfg(target_os = "macos")] active_tools: Option, @@ -104,6 +106,9 @@ pub(crate) struct App { #[cfg(target_os = "macos")] _endpoint: Option, error: Option, + pub(super) activity: Option, + #[cfg(target_os = "macos")] + skip_compaction_once: bool, } #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] @@ -295,6 +300,8 @@ impl App { #[cfg(target_os = "macos")] active_generation: None, #[cfg(target_os = "macos")] + active_compaction: None, + #[cfg(target_os = "macos")] agent_tools: None, #[cfg(target_os = "macos")] active_tools: None, @@ -314,6 +321,9 @@ impl App { None } }, + activity: None, + #[cfg(target_os = "macos")] + skip_compaction_once: false, } } Err(error) => Self::failed( @@ -384,6 +394,8 @@ impl App { #[cfg(target_os = "macos")] active_generation: None, #[cfg(target_os = "macos")] + active_compaction: None, + #[cfg(target_os = "macos")] agent_tools: None, #[cfg(target_os = "macos")] active_tools: None, @@ -397,6 +409,9 @@ impl App { Some(service_error) => format!("{error} {service_error}"), None => error, }), + activity: None, + #[cfg(target_os = "macos")] + skip_compaction_once: false, } } diff --git a/src/app/generation.rs b/src/app/generation.rs index e6b2f82..7c4245f 100644 --- a/src/app/generation.rs +++ b/src/app/generation.rs @@ -18,6 +18,18 @@ pub(super) struct TitleRequest { content: String, } +#[cfg(target_os = "macos")] +pub(super) enum PendingContinuation { + User(String), + Tool(String), +} + +#[cfg(target_os = "macos")] +pub(super) struct CompactionRequest { + active: ActiveGeneration, + pending: PendingContinuation, +} + #[derive(Clone, Debug)] pub(crate) struct ChatMessage { pub(super) id: i32, @@ -79,6 +91,18 @@ impl App { if prompt.is_empty() { return; } + #[cfg(target_os = "macos")] + if !std::mem::take(&mut self.skip_compaction_once) + && crate::compaction::should_compact(self.context_used, self.context_limit) + { + if let Err(error) = self.start_compaction( + PendingContinuation::User(prompt), + "soft limit before user turn", + ) { + self.error = Some(error); + } + return; + } let model = self.config.model; let effective = crate::settings::effective_settings( model, @@ -95,6 +119,10 @@ impl App { }; effective.turn.system_prompt = crate::agent::system_prompt(model, &effective.turn.system_prompt); + effective.turn.system_prompt = crate::compaction::summary_system_prompt( + &effective.turn.system_prompt, + self.compaction_summary(), + ); let assistant_reasoning = effective.turn.reasoning_mode != ReasoningMode::Direct; #[cfg(target_os = "macos")] let mut messages = self @@ -191,6 +219,10 @@ impl App { } pub(super) fn poll_generation(&mut self) -> bool { + #[cfg(target_os = "macos")] + if self.active_compaction.is_some() { + return self.poll_compaction(); + } #[cfg(target_os = "macos")] if let Some(active) = &self.active_tools { match crate::agent::try_tool_result(active) { @@ -229,6 +261,11 @@ impl App { loop { match active.events.try_recv() { Ok(GenerationEvent::Loading) => {} + Ok(GenerationEvent::Compacted(_)) => { + self.generating = false; + self.error = + Some("The model runtime returned an unexpected compaction event.".into()); + } Ok(GenerationEvent::Chunk { reasoning, content }) => { if let Some(message) = self.conversation.last_mut() && !message.user @@ -363,6 +400,18 @@ impl App { let session_id = self .selected_session .ok_or_else(|| "The active session is unavailable.".to_owned())?; + if !std::mem::take(&mut self.skip_compaction_once) + && crate::compaction::tool_result_needs_compaction( + self.context_used, + self.context_limit, + result, + ) + { + return self.start_compaction( + PendingContinuation::Tool(result.to_owned()), + "context pressure before tool continuation", + ); + } let model = self.config.model; let mut effective = crate::settings::effective_settings( model, @@ -372,6 +421,10 @@ impl App { )?; effective.turn.system_prompt = crate::agent::system_prompt(model, &effective.turn.system_prompt); + effective.turn.system_prompt = crate::compaction::summary_system_prompt( + &effective.turn.system_prompt, + self.compaction_summary(), + ); let assistant_reasoning = effective.turn.reasoning_mode != ReasoningMode::Direct; let saved = self .database @@ -412,6 +465,175 @@ impl App { Ok(()) } + #[cfg(target_os = "macos")] + fn start_compaction( + &mut self, + pending: PendingContinuation, + reason: &str, + ) -> Result<(), String> { + let model = self.config.model; + let mut effective = crate::settings::effective_settings( + model, + &self.config.generation, + &self.config.runtime, + &models_path(), + )?; + effective.turn.system_prompt = + crate::agent::system_prompt(model, &effective.turn.system_prompt); + effective.turn.system_prompt = crate::compaction::summary_system_prompt( + &effective.turn.system_prompt, + self.compaction_summary(), + ); + let messages = self + .conversation + .iter() + .map(|message| ChatTurn { + user: message.user, + tool: message.tool, + skip_previous_eos: false, + reasoning: message.reasoning.clone(), + reasoning_complete: message.reasoning_complete, + content: message.content.clone(), + }) + .collect(); + let idle_timeout = Duration::from_secs(self.config.idle_timeout_minutes.max(1) as u64 * 60); + let active = self + .generation_service + .as_ref() + .ok_or_else(|| "The model runtime is unavailable.".to_owned())? + .compact( + effective.engine, + effective.turn, + messages, + reason, + idle_timeout, + )?; + self.active_compaction = Some(CompactionRequest { active, pending }); + self.generating = true; + self.activity = Some("Compacting durable task state…".into()); + self.tokens_per_second = None; + Ok(()) + } + + #[cfg(target_os = "macos")] + fn poll_compaction(&mut self) -> bool { + let Some(request) = &mut self.active_compaction else { + return false; + }; + loop { + match request.active.events.try_recv() { + Ok(GenerationEvent::Loading) => self.activity = Some("Loading model…".into()), + Ok(GenerationEvent::Context { + used, + limit, + tokens_per_second, + }) => { + self.context_used = used; + self.context_limit = limit; + self.tokens_per_second = tokens_per_second; + } + Ok(GenerationEvent::Compacted(result)) => { + let request = self.active_compaction.take().unwrap(); + match result { + Ok(compacted) => { + if let Err(error) = self.apply_compaction(&compacted) { + self.generating = false; + self.activity = None; + self.error = Some(error); + return false; + } + self.context_used = compacted.context_tokens; + self.generating = false; + self.activity = None; + match request.pending { + PendingContinuation::User(prompt) => { + self.composer = prompt; + self.skip_compaction_once = true; + self.start_generation(); + } + PendingContinuation::Tool(result) => { + let result = crate::compaction::bounded_tool_result( + self.context_used, + self.context_limit, + result, + ); + self.skip_compaction_once = true; + if let Err(error) = self.continue_after_tool_result(&result) { + self.generating = false; + self.error = Some(error); + } + } + } + return true; + } + Err(error) => { + self.generating = false; + self.activity = None; + self.error = Some(error); + return false; + } + } + } + Ok(GenerationEvent::Finished(_)) | Ok(GenerationEvent::Chunk { .. }) => {} + Err(TryRecvError::Empty) => return false, + Err(TryRecvError::Disconnected) => { + self.active_compaction = None; + self.generating = false; + self.activity = None; + self.error = Some("The model runtime stopped unexpectedly.".into()); + return false; + } + } + } + } + + #[cfg(target_os = "macos")] + fn apply_compaction( + &mut self, + compacted: &crate::engine::CompactionOutput, + ) -> Result<(), String> { + let session_id = self + .selected_session + .ok_or_else(|| "The active session is unavailable.".to_owned())?; + let tail = compacted + .tail + .iter() + .map(|message| crate::database::MessageDraft { + user: message.user, + tool: message.tool, + reasoning: message.reasoning.clone(), + reasoning_complete: message.reasoning_complete, + content: message.content.clone(), + }) + .collect::>(); + let messages = self + .database + .as_mut() + .ok_or_else(|| "The project database is unavailable.".to_owned())? + .replace_with_compacted_transcript(session_id, &compacted.summary, &tail) + .map_err(|error| format!("Could not save compacted conversation: {error}"))?; + self.conversation = messages.into_iter().map(ChatMessage::from).collect(); + if let Some(session) = self + .projects + .iter_mut() + .flat_map(|project| &mut project.sessions) + .find(|session| session.id == session_id) + { + session.compacted_summary = Some(compacted.summary.clone()); + } + let _ = fs::remove_file(session_checkpoint_path(session_id)); + Ok(()) + } + + pub(super) fn compaction_summary(&self) -> Option<&str> { + let session_id = self.selected_session?; + self.projects + .iter() + .flat_map(|project| &project.sessions) + .find(|session| session.id == session_id) + .and_then(|session| session.compacted_summary.as_deref()) + } + /// Asks the model for a session title without opening a session for it: the /// turn runs against the shared transient KV cache and only its text is kept. /// diff --git a/src/app/view/chat.rs b/src/app/view/chat.rs index d2d83b0..abf7fba 100644 --- a/src/app/view/chat.rs +++ b/src/app/view/chat.rs @@ -56,6 +56,20 @@ impl App { ); } else { let markdown_style = markdown::Style::from_palette(app_theme().palette()); + if let Some(summary) = self.compaction_summary() { + messages = messages.push( + container( + column![ + text("Compacted task state").size(11), + text(summary).size(13).color(muted_text()), + ] + .spacing(5), + ) + .padding(14) + .width(Length::Fill) + .style(preference_group_style), + ); + } for (index, message) in self.conversation.iter().enumerate() { let label = if message.user { "You" @@ -113,7 +127,9 @@ impl App { ); } } else if active && message.reasoning.is_none() { - body = body.push(text("Loading model…").size(14)); + body = body.push( + text(self.activity.as_deref().unwrap_or("Loading model…")).size(14), + ); } if !message.user && !message.tool { for summary in diff --git a/src/compaction.rs b/src/compaction.rs new file mode 100644 index 0000000..3713bbb --- /dev/null +++ b/src/compaction.rs @@ -0,0 +1,141 @@ +use crate::engine::ChatTurn; + +pub(crate) const SUMMARY_MAX_TOKENS: i32 = 4096; +pub(crate) const TOOL_RESULT_RESERVE_TOKENS: u32 = 1024; + +const SOFT_PERCENT: u32 = 85; +const MIN_FREE_TOKENS: u32 = 8192; +const TAIL_DIVISOR: u32 = 10; +const TAIL_CAP_TOKENS: u32 = 50_000; + +pub(crate) const SUMMARY_PREFIX: &str = + "[DS4Server compacted earlier conversation. Durable task-state summary follows.]"; +pub(crate) const SUMMARY_SUFFIX: &str = + "[End compacted summary. Recent conversation continues verbatim below.]"; + +pub(crate) fn should_compact(used: u32, limit: u32) -> bool { + if used == 0 || limit == 0 { + return false; + } + used >= limit.saturating_mul(SOFT_PERCENT) / 100 + || limit.saturating_sub(used) <= MIN_FREE_TOKENS.min(limit / 8) +} + +pub(crate) fn tool_result_needs_compaction(used: u32, limit: u32, result: &str) -> bool { + should_compact(used, limit) + || used + .saturating_add(result.len().min(u32::MAX as usize) as u32) + .saturating_add(TOOL_RESULT_RESERVE_TOKENS) + >= limit +} + +pub(crate) fn bounded_tool_result(used: u32, limit: u32, result: String) -> String { + if used + .saturating_add(result.len().min(u32::MAX as usize) as u32) + .saturating_add(TOOL_RESULT_RESERVE_TOKENS) + < limit + { + result + } else { + "Tool error: the result is too large for the remaining context after compaction. Retry with a smaller read/search/bash output.\n".into() + } +} + +pub(crate) fn tail_budget(context: u32) -> u32 { + (context / TAIL_DIVISOR).clamp(1, TAIL_CAP_TOKENS) +} + +/// Matches the reference token-level rule: start at the target unless a user +/// boundary appears between it and the end. Message starts are token offsets. +pub(crate) fn tail_start(messages: &[ChatTurn], starts: &[u32], bottom: u32, budget: u32) -> usize { + let target = bottom.saturating_sub(budget); + messages + .iter() + .zip(starts) + .position(|(message, start)| message.user && *start >= target) + .or_else(|| starts.iter().position(|start| *start >= target)) + .unwrap_or(messages.len()) +} + +pub(crate) fn summary_prompt(reason: &str) -> String { + format!( + "Internal DS4Server context compaction request. This is not a user request.\n\ +Write a durable task-state summary of the conversation so far. Preserve only facts that matter for continuing the work:\n\ +- user goals, constraints, and preferences\n\ +- files inspected or edited\n\ +- commands run and important results\n\ +- decisions, rejected approaches, known bugs, and pending next steps\n\ +- reloadable bulky data with exact paths/ranges/commands when available\n\n\ +Do not invent facts. Do not include generic narration or raw file contents unless essential.\n\ +After the summary, stop. Do not continue the user task, call tools, or output thinking/DSML control markup.\n\ +Output only the compact summary.\n\nCompaction reason: {reason}\n" + ) +} + +pub(crate) fn summary_system_prompt(base: &str, summary: Option<&str>) -> String { + match summary.filter(|summary| !summary.trim().is_empty()) { + Some(summary) => format!( + "{base}\n\n{SUMMARY_PREFIX}\n{}\n{SUMMARY_SUFFIX}", + summary.trim() + ), + None => base.to_owned(), + } +} + +pub(crate) fn sanitize_summary(summary: &str) -> String { + let end = ["<｜DSML｜", "", "", ""] + .into_iter() + .filter_map(|marker| summary.find(marker)) + .min() + .unwrap_or(summary.len()); + summary[..end].trim().to_owned() +} + +#[cfg(test)] +mod tests { + use super::*; + + fn turn(user: bool) -> ChatTurn { + ChatTurn { + user, + tool: false, + skip_previous_eos: false, + reasoning: None, + reasoning_complete: true, + content: String::new(), + } + } + + #[test] + fn soft_trigger_matches_reference_thresholds() { + assert!(should_compact(85_000, 100_000)); + assert!(should_compact(92_000, 100_000)); + assert!(!should_compact(84_999, 100_000)); + assert!(should_compact(6_963, 8_192)); + assert!(!should_compact(6_962, 8_192)); + } + + #[test] + fn tail_prefers_the_first_user_boundary_after_target() { + let messages = [turn(true), turn(false), turn(true), turn(false)]; + assert_eq!(tail_start(&messages, &[10, 30, 70, 90], 100, 40), 2); + assert_eq!(tail_start(&messages, &[10, 30, 70, 90], 100, 15), 3); + } + + #[test] + fn summary_control_markup_is_private() { + assert_eq!( + sanitize_summary("state kept\nbash"), + "state kept" + ); + } + + #[test] + fn oversized_tool_result_becomes_a_bounded_retry_error() { + let result = "x".repeat(4_000); + assert!(tool_result_needs_compaction(6_000, 10_000, &result)); + let error = bounded_tool_result(4_000, 5_000, result); + assert!(error.starts_with("Tool error:")); + assert!(error.len() < 160); + } +} diff --git a/src/database.rs b/src/database.rs index 74128af..a27dc16 100644 --- a/src/database.rs +++ b/src/database.rs @@ -39,6 +39,7 @@ pub struct Session { pub last_tokens_per_second: Option, /// Raw column value; read it through [`Session::state`]. state: String, + pub compacted_summary: Option, } impl Session { @@ -56,6 +57,7 @@ impl Session { context_limit: 0, last_tokens_per_second: None, state: state.as_id().to_owned(), + compacted_summary: None, } } } @@ -129,6 +131,14 @@ struct NewMessage<'a> { content: &'a str, } +pub struct MessageDraft { + pub user: bool, + pub tool: bool, + pub reasoning: Option, + pub reasoning_complete: bool, + pub content: String, +} + #[derive(Debug)] pub struct ProjectWithSessions { pub project: Project, @@ -391,6 +401,40 @@ impl Database { .map(|_| ()) .map_err(|error| error.to_string()) } + + pub fn replace_with_compacted_transcript( + &mut self, + session_id: i32, + summary: &str, + tail: &[MessageDraft], + ) -> Result, String> { + self.connection + .transaction(|connection| { + diesel::delete(messages::table.filter(messages::session_id.eq(session_id))) + .execute(connection)?; + diesel::update(sessions::table.find(session_id)) + .set(sessions::compacted_summary.eq(Some(summary))) + .execute(connection)?; + let mut stored = Vec::with_capacity(tail.len()); + for message in tail { + stored.push( + diesel::insert_into(messages::table) + .values(NewMessage { + session_id, + user: message.user, + tool: message.tool, + reasoning: message.reasoning.as_deref(), + reasoning_complete: message.reasoning_complete, + content: &message.content, + }) + .returning(StoredMessage::as_returning()) + .get_result(connection)?, + ); + } + Ok(stored) + }) + .map_err(|error: diesel::result::Error| error.to_string()) + } } #[cfg(test)] @@ -506,6 +550,29 @@ mod tests { assert_eq!(messages[2].content, "Tool result"); assert!(!messages[3].user); assert!(!messages[3].tool); + let compacted = reopened + .replace_with_compacted_transcript( + session.id, + "Keep the active task.", + &[MessageDraft { + user: true, + tool: false, + reasoning: None, + reasoning_complete: true, + content: "Recent question".into(), + }], + ) + .unwrap(); + assert_eq!(compacted.len(), 1); + drop(reopened); + let mut reopened = Database::open(&path).unwrap(); + assert_eq!( + reopened.load_projects().unwrap()[0].sessions[0] + .compacted_summary + .as_deref(), + Some("Keep the active task.") + ); + assert_eq!(reopened.load_messages(session.id).unwrap().len(), 1); reopened.delete_session(session.id).unwrap(); assert!(reopened.load_messages(session.id).unwrap().is_empty()); drop(reopened); diff --git a/src/engine.rs b/src/engine.rs index 628a044..5df4924 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -341,6 +341,13 @@ pub(crate) struct GenerationOutput { pub(crate) checkpoint_bytes: u64, } +#[cfg(target_os = "macos")] +pub(crate) struct CompactionOutput { + pub(crate) summary: String, + pub(crate) tail: Vec, + pub(crate) context_tokens: u32, +} + #[cfg(target_os = "macos")] impl Generator { pub(crate) fn open(settings: &EngineSettings, metrics: Arc) -> Result { @@ -496,6 +503,95 @@ impl Generator { Ok(output) } + pub(crate) fn compact( + &mut self, + messages: &[ChatTurn], + settings: &TurnSettings, + reason: &str, + cancelled: &AtomicBool, + mut progress: impl FnMut(u32, u32, Option), + ) -> Result { + self.executor.reset()?; + self.checkpoint = None; + let result = (|| { + let mut private_messages = messages.to_vec(); + private_messages.push(ChatTurn { + user: true, + tool: false, + skip_previous_eos: false, + reasoning: None, + reasoning_complete: true, + content: crate::compaction::summary_prompt(reason), + }); + let mut private_settings = settings.clone(); + private_settings.reasoning_mode = ReasoningMode::Direct; + private_settings.max_generated_tokens = crate::compaction::SUMMARY_MAX_TOKENS; + private_settings.temperature = 0.0; + private_settings.stops.extend([ + "<｜DSML｜".into(), + "".into(), + "".into(), + "".into(), + ]); + let (output, prompt_complete) = self.generate_inner( + &private_messages, + &private_settings, + cancelled, + &mut |_, _| {}, + &mut progress, + )?; + if cancelled.load(Ordering::Relaxed) || !prompt_complete { + return Err("context compaction interrupted".into()); + } + let summary = crate::compaction::sanitize_summary(&output.message.content); + if summary.is_empty() { + return Err("context compaction produced an empty summary".into()); + } + + let full = self.executor.model().render_conversation( + &settings.system_prompt, + messages, + settings.reasoning_mode, + ); + let mut starts = Vec::with_capacity(messages.len()); + for index in 0..messages.len() { + starts.push( + self.executor + .model() + .render_conversation( + &settings.system_prompt, + &messages[..index], + settings.reasoning_mode, + ) + .len() as u32, + ); + } + let start = crate::compaction::tail_start( + messages, + &starts, + full.len() as u32, + crate::compaction::tail_budget(self.executor.context()), + ); + let tail = messages[start..].to_vec(); + let rebuilt_system = + crate::compaction::summary_system_prompt(&settings.system_prompt, Some(&summary)); + let context_tokens = self + .executor + .model() + .render_conversation(&rebuilt_system, &tail, settings.reasoning_mode) + .len() as u32; + Ok(CompactionOutput { + summary, + tail, + context_tokens, + }) + })(); + self.executor.reset()?; + self.checkpoint = None; + result + } + fn select_checkpoint( &mut self, checkpoint: &Path, diff --git a/src/main.rs b/src/main.rs index f4db3d1..e12304d 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,5 +1,6 @@ mod agent; mod app; +mod compaction; mod config; mod database; mod engine; diff --git a/src/runtime.rs b/src/runtime.rs index 6d89d56..202255c 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -1,4 +1,4 @@ -use crate::engine::{ChatTurn, GenerationOutput, Generator}; +use crate::engine::{ChatTurn, CompactionOutput, GenerationOutput, Generator}; use crate::metrics::{Metrics, WorkSource}; use crate::settings::{EngineSettings, TurnSettings}; use std::path::PathBuf; @@ -56,6 +56,7 @@ pub(crate) enum GenerationEvent { tokens_per_second: Option, }, Finished(Result), + Compacted(Result), } struct Command { @@ -63,6 +64,7 @@ struct Command { turn: TurnSettings, messages: Vec, checkpoint: CheckpointTarget, + compact_reason: Option, idle_timeout: Duration, cancel: Arc, events: Sender, @@ -98,6 +100,7 @@ impl GenerationService { turn, messages, checkpoint, + compact_reason: None, idle_timeout, cancel: Arc::clone(&cancel), events, @@ -112,6 +115,35 @@ impl GenerationService { cancel, }) } + + pub(crate) fn compact( + &self, + engine: EngineSettings, + turn: TurnSettings, + messages: Vec, + reason: &str, + idle_timeout: Duration, + ) -> Result { + let cancel = Arc::new(AtomicBool::new(false)); + let (events, receiver) = mpsc::channel(); + self.metrics.request_queued(WorkSource::LocalChat); + self.commands + .send(Command { + engine, + turn, + messages, + checkpoint: CheckpointTarget::OneShot(PathBuf::new()), + compact_reason: Some(reason.to_owned()), + idle_timeout, + cancel: Arc::clone(&cancel), + events, + }) + .map_err(|_| "The model runtime stopped unexpectedly.".to_owned())?; + Ok(ActiveGeneration { + events: receiver, + cancel, + }) + } } fn run(commands: Receiver, metrics: Arc) { @@ -228,6 +260,24 @@ fn run_command( tokens_per_second, }); }; + if let Some(reason) = &command.compact_reason { + let result = generator.compact( + &command.messages, + &command.turn, + reason, + &command.cancel, + &mut progress, + ); + match &result { + Ok(_) => { + metrics.request_finished(source, request_started.elapsed(), 0, 0, 0, None, 0) + } + Err(_) => metrics.request_failed(request_started.elapsed()), + } + let _ = command.events.send(GenerationEvent::Compacted(result)); + *last_used = Instant::now(); + return; + } let result = match command.checkpoint { CheckpointTarget::Local(path) => generator.generate( &path, diff --git a/src/schema.rs b/src/schema.rs index 44f018d..731a280 100644 --- a/src/schema.rs +++ b/src/schema.rs @@ -28,6 +28,7 @@ diesel::table! { context_limit -> Integer, last_tokens_per_second -> Nullable, state -> Text, + compacted_summary -> Nullable, } } diff --git a/src/server/response.rs b/src/server/response.rs index 217bc56..02e0135 100644 --- a/src/server/response.rs +++ b/src/server/response.rs @@ -165,6 +165,12 @@ fn completion_stream_response( send_sse_headers_with_cors(stream, request.cors).map_err(|error| (500, error))?; while let Ok(event) = active.events.recv() { match event { + GenerationEvent::Compacted(_) => { + return Err(( + 500, + "The model runtime returned an unexpected compaction event.".into(), + )); + } GenerationEvent::Chunk { reasoning: false, content, @@ -737,6 +743,12 @@ fn stream_response_with_keepalive( Err(_) => return Ok(()), }; match event { + GenerationEvent::Compacted(_) => { + return Err(( + 500, + "The model runtime returned an unexpected compaction event.".into(), + )); + } GenerationEvent::Loading => {} GenerationEvent::Context { tokens_per_second, ..