hugo/DS4Server
1
0
Fork 0
Code Issues 7 Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Support long-running agent turns

Browse Source
This commit is contained in:
Georg Bauer
2026-07-26 11:07:28 +02:00
parent 2a14b93335
commit 3c75b8f6c1
14 changed files with 432 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

145
src/app/generation.rs
View File

@@ -36,6 +36,7 @@ pub(crate) struct ChatMessage {
pub(super) id: i32,
pub(super) user: bool,
pub(super) tool: bool,
pub(super) system: bool,
pub(super) reasoning: Option<String>,
pub(super) reasoning_complete: bool,
pub(super) reasoning_open: bool,
@@ -72,6 +73,7 @@ impl From<StoredMessage> for ChatMessage {
id: message.id,
user: message.user,
tool: message.tool,
system: message.system,
reasoning: message.reasoning,
reasoning_complete: message.reasoning_complete,
reasoning_open: false,
@@ -85,13 +87,27 @@ impl From<StoredMessage> for ChatMessage {
impl App {
pub(super) fn start_generation(&mut self) {
if self.generating || self.selected_project.is_none() {
if self.selected_project.is_none() {
return;
}
let prompt = self.composer.trim().to_owned();
if prompt.is_empty() {
return;
}
if self.generating {
self.queued_inputs.push_back(prompt);
self.composer.clear();
self.activity = Some(format!(
"{} queued input{}",
self.queued_inputs.len(),
if self.queued_inputs.len() == 1 {
""
} else {
"s"
}
));
return;
}
#[cfg(target_os = "macos")]
if prompt == "/compact" {
self.composer.clear();
@@ -136,12 +152,30 @@ impl App {
);
let assistant_reasoning = effective.turn.reasoning_mode != ReasoningMode::Direct;
#[cfg(target_os = "macos")]
let opening_turn = self.selected_session.is_none();
#[cfg(target_os = "macos")]
let mut injected_system = Vec::new();
#[cfg(target_os = "macos")]
if opening_turn {
injected_system.push(crate::agent::datetime_context());
}
#[cfg(target_os = "macos")]
let reminder_injected = self.system_prompt_reminder_due();
#[cfg(target_os = "macos")]
if reminder_injected {
injected_system.push(crate::agent::system_prompt_reminder(
model,
&self.config.generation.system_prompt,
));
}
#[cfg(target_os = "macos")]
let mut messages = self
.conversation
.iter()
.map(|message| ChatTurn {
user: message.user,
tool: message.tool,
system: message.system,
skip_previous_eos: false,
reasoning: message.reasoning.clone(),
reasoning_complete: message.reasoning_complete,
@@ -149,9 +183,20 @@ impl App {
})
.collect::<Vec<_>>();
#[cfg(target_os = "macos")]
messages.extend(injected_system.iter().map(|content| ChatTurn {
user: false,
tool: false,
system: true,
skip_previous_eos: false,
reasoning: None,
reasoning_complete: true,
content: content.clone(),
}));
#[cfg(target_os = "macos")]
messages.push(ChatTurn {
user: true,
tool: false,
system: false,
skip_previous_eos: false,
reasoning: None,
reasoning_complete: true,
@@ -161,7 +206,6 @@ impl App {
#[cfg(target_os = "macos")]
{
// A draft session only reaches the database once there is a turn to store.
let opening_turn = self.selected_session.is_none();
let session_id = match self.selected_session {
Some(session_id) => session_id,
None => {
@@ -184,13 +228,21 @@ impl App {
let Some(database) = &mut self.database else {
return;
};
let saved = match database.start_chat_turn(session_id, &prompt, assistant_reasoning) {
let mut saved = match database.start_chat_turn(
session_id,
&prompt,
&injected_system,
assistant_reasoning,
) {
Ok(turn) => turn,
Err(error) => {
self.error = Some(format!("Could not save the chat turn: {error}"));
return;
}
};
if reminder_injected {
self.system_prompt_seen_at = self.context_used;
}
let idle_timeout =
Duration::from_secs(self.config.idle_timeout_minutes.max(1) as u64 * 60);
self.active_generation = match service.generate(
@@ -207,13 +259,18 @@ impl App {
return;
}
};
let user = ChatMessage::from(saved.0);
let mut assistant = ChatMessage::from(saved.1);
let mut assistant = ChatMessage::from(saved.pop().unwrap());
let user = ChatMessage::from(saved.pop().unwrap());
for message in saved {
self.conversation.push(ChatMessage::from(message));
}
assistant.reasoning_open = assistant_reasoning;
self.composer.clear();
self.conversation.push(user);
self.conversation.push(assistant);
self.generating = true;
self.stop_requested = false;
self.activity = Some("Loading model…".into());
self.tokens_per_second = None;
self.error = None;
if opening_turn {
@@ -229,6 +286,10 @@ impl App {
}
}
fn system_prompt_reminder_due(&self) -> bool {
crate::agent::prompt_reminder_due(self.context_used, self.system_prompt_seen_at)
}
pub(super) fn poll_generation(&mut self) -> bool {
#[cfg(target_os = "macos")]
if self.active_compaction.is_some() {
@@ -242,6 +303,7 @@ impl App {
self.active_tools = None;
if cancelled {
self.generating = false;
self.activity = Some("Stopped".into());
return false;
}
if let Err(error) = self.continue_after_tool_result(&result) {
@@ -254,6 +316,7 @@ impl App {
Err(error) => {
self.active_tools = None;
self.generating = false;
self.activity = Some("Failed".into());
self.error = Some(error);
return false;
}
@@ -269,6 +332,8 @@ impl App {
#[cfg(target_os = "macos")]
let mut context_changed = false;
#[cfg(target_os = "macos")]
let mut start_queued = false;
#[cfg(target_os = "macos")]
loop {
match active.events.try_recv() {
Ok(GenerationEvent::Loading) => {}
@@ -297,6 +362,10 @@ impl App {
}
Ok(GenerationEvent::Finished(result)) => {
match result {
Ok(_) if self.stop_requested => {
self.generating = false;
self.activity = Some("Stopped".into());
}
Ok(_) => {
let model = self.config.model;
let content = self
@@ -311,14 +380,23 @@ impl App {
self.error = Some(error);
}
}
Ok(_) => self.generating = false,
Ok(_) => {
self.generating = false;
self.activity = None;
start_queued = !self.queued_inputs.is_empty();
}
Err(error) => {
self.active_tools = Some(crate::agent::error_async(error));
}
}
}
Err(_) if self.stop_requested => {
self.generating = false;
self.activity = Some("Stopped".into());
}
Err(error) => {
self.generating = false;
self.activity = Some("Failed".into());
self.error = Some(error);
}
}
@@ -330,6 +408,7 @@ impl App {
self.generating = false;
self.active_generation = None;
self.error = Some("The model runtime stopped unexpectedly.".into());
self.activity = Some("Failed".into());
break;
}
}
@@ -378,11 +457,22 @@ impl App {
}
}
#[cfg(target_os = "macos")]
if start_queued {
self.start_next_queued();
}
#[cfg(target_os = "macos")]
return transcript_changed;
#[cfg(not(target_os = "macos"))]
false
}
fn start_next_queued(&mut self) {
if let Some(prompt) = self.queued_inputs.pop_front() {
self.composer = prompt;
self.start_generation();
}
}
#[cfg(target_os = "macos")]
fn start_agent_tools(&mut self, calls: Vec<crate::agent::ToolCall>) -> Result<(), String> {
let session_id = self
@@ -437,26 +527,43 @@ impl App {
self.compaction_summary(),
);
let assistant_reasoning = effective.turn.reasoning_mode != ReasoningMode::Direct;
let saved = self
let queued = self.queued_inputs.iter().cloned().collect::<Vec<_>>();
let reminder = self.system_prompt_reminder_due().then(|| {
crate::agent::system_prompt_reminder(model, &self.config.generation.system_prompt)
});
let mut saved = self
.database
.as_mut()
.ok_or_else(|| "The project database is unavailable.".to_owned())?
.continue_tool_turn(session_id, result, assistant_reasoning)
.continue_tool_turn(
session_id,
result,
&queued,
reminder.as_deref(),
assistant_reasoning,
)
.map_err(|error| format!("Could not save the tool turn: {error}"))?;
self.conversation.push(ChatMessage::from(saved.0));
self.queued_inputs.clear();
if reminder.is_some() {
self.system_prompt_seen_at = self.context_used;
}
let mut assistant = ChatMessage::from(saved.pop().unwrap());
for message in saved {
self.conversation.push(ChatMessage::from(message));
}
let messages = self
.conversation
.iter()
.map(|message| ChatTurn {
user: message.user,
tool: message.tool,
system: message.system,
skip_previous_eos: false,
reasoning: message.reasoning.clone(),
reasoning_complete: message.reasoning_complete,
content: message.content.clone(),
})
.collect();
let mut assistant = ChatMessage::from(saved.1);
assistant.reasoning_open = assistant_reasoning;
self.conversation.push(assistant);
let idle_timeout = Duration::from_secs(self.config.idle_timeout_minutes.max(1) as u64 * 60);
@@ -473,6 +580,8 @@ impl App {
)?,
);
self.tokens_per_second = None;
self.activity = Some("Continuing after tools…".into());
self.stop_requested = false;
Ok(())
}
@@ -501,6 +610,7 @@ impl App {
.map(|message| ChatTurn {
user: message.user,
tool: message.tool,
system: message.system,
skip_previous_eos: false,
reasoning: message.reasoning.clone(),
reasoning_complete: message.reasoning_complete,
@@ -559,10 +669,11 @@ impl App {
return false;
}
self.context_used = compacted.context_tokens;
self.system_prompt_seen_at = compacted.context_tokens;
self.generating = false;
self.activity = None;
match request.pending {
PendingContinuation::None => {}
PendingContinuation::None => self.start_next_queued(),
PendingContinuation::User(prompt) => {
self.composer = prompt;
self.skip_compaction_once = true;
@@ -585,8 +696,12 @@ impl App {
}
Err(error) => {
self.generating = false;
self.activity = None;
self.error = Some(error);
if self.stop_requested {
self.activity = Some("Stopped".into());
} else {
self.activity = Some("Failed".into());
self.error = Some(error);
}
return false;
}
}
@@ -618,6 +733,7 @@ impl App {
.map(|message| crate::database::MessageDraft {
user: message.user,
tool: message.tool,
system: message.system,
reasoning: message.reasoning.clone(),
reasoning_complete: message.reasoning_complete,
content: message.content.clone(),
@@ -633,6 +749,7 @@ impl App {
tail.push(crate::database::MessageDraft {
user: false,
tool: true,
system: false,
reasoning: None,
reasoning_complete: true,
content: observation,
@@ -713,6 +830,7 @@ impl App {
.map(|message| ChatTurn {
user: message.user,
tool: message.tool,
system: message.system,
skip_previous_eos: false,
reasoning: None,
reasoning_complete: true,
@@ -725,6 +843,7 @@ impl App {
messages.push(ChatTurn {
user: true,
tool: false,
system: false,
skip_previous_eos: false,
reasoning: None,
reasoning_complete: true,