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

Refactor application and runtime boundaries

Browse Source
This commit is contained in:
Hermes Agent
2026-07-29 10:40:39 +00:00
parent 902037f947
commit 0f0a1a5409
10 changed files with 1676 additions and 1678 deletions
Download Patch File Download Diff File Expand all files Collapse all files

298
src/runtime.rs
View File

@@ -34,18 +34,6 @@ pub(crate) enum CheckpointTarget {
bootstrap: Option<PathBuf>,
},
Transient(PathBuf),
/// Same transient KV handling as [`CheckpointTarget::Transient`], but asked
/// for by the app itself (session titling) rather than by an HTTP client.
OneShot(PathBuf),
}
impl CheckpointTarget {
fn source(&self) -> WorkSource {
match self {
Self::Local { .. } | Self::OneShot(_) => WorkSource::LocalChat,
Self::Transient(_) => WorkSource::Http,
}
}
}
pub(crate) enum GenerationEvent {
@@ -75,23 +63,60 @@ enum Operation {
}
#[derive(Clone, Copy)]
enum ResponseKind {
Generation,
Compaction,
Measurement,
enum TrackingPolicy {
None,
Queued,
QueuedAndRejected,
}
impl TrackingPolicy {
fn records_queue(self) -> bool {
matches!(self, Self::Queued | Self::QueuedAndRejected)
}
fn records_rejection(self) -> bool {
matches!(self, Self::QueuedAndRejected)
}
}
impl Operation {
fn error_handler(&self) -> fn(String) -> GenerationEvent {
match self {
Self::Generate => |error| GenerationEvent::Finished(Err(error)),
Self::Compact { .. } => |error| GenerationEvent::Compacted(Err(error)),
Self::Measure => |error| GenerationEvent::Measured(Err(error)),
}
}
}
struct Command {
engine: EngineSettings,
turn: TurnSettings,
messages: Vec<ChatTurn>,
checkpoint: CheckpointTarget,
checkpoint: Option<CheckpointTarget>,
source: WorkSource,
operation: Operation,
idle_timeout: Duration,
cancel: Arc<AtomicBool>,
events: Sender<GenerationEvent>,
}
pub(crate) struct CompactionInput {
pub(crate) engine: EngineSettings,
pub(crate) turn: TurnSettings,
pub(crate) messages: Vec<ChatTurn>,
pub(crate) reason: String,
pub(crate) rebuild_system_prompt: String,
pub(crate) checkpoint: PathBuf,
pub(crate) idle_timeout: Duration,
}
struct RuntimeState {
loaded: Option<(EngineSettings, Generator)>,
last_used: Instant,
idle_timeout: Duration,
}
impl GenerationService {
pub(crate) fn spawn(metrics: Arc<Metrics>) -> Result<Self, String> {
let (commands, receiver) = mpsc::channel::<Command>();
@@ -109,70 +134,42 @@ impl GenerationService {
turn: TurnSettings,
messages: Vec<ChatTurn>,
checkpoint: CheckpointTarget,
source: WorkSource,
idle_timeout: Duration,
) -> Result<ActiveGeneration, String> {
let source = checkpoint.source();
let cancel = Arc::new(AtomicBool::new(false));
let (events, receiver) = mpsc::channel();
self.metrics.request_queued(source);
if self
.commands
.send(Command {
self.submit(
CommandRequest {
engine,
turn,
messages,
checkpoint,
checkpoint: Some(checkpoint),
source,
operation: Operation::Generate,
idle_timeout,
cancel: Arc::clone(&cancel),
events,
})
.is_err()
{
self.metrics.request_rejected();
return Err("The model runtime stopped unexpectedly.".to_owned());
}
Ok(ActiveGeneration {
events: receiver,
cancel,
})
},
TrackingPolicy::QueuedAndRejected,
)
}
#[allow(clippy::too_many_arguments)]
pub(crate) fn compact(
&self,
engine: EngineSettings,
turn: TurnSettings,
messages: Vec<ChatTurn>,
reason: &str,
rebuild_system_prompt: String,
checkpoint: PathBuf,
idle_timeout: Duration,
) -> Result<ActiveGeneration, String> {
let cancel = Arc::new(AtomicBool::new(false));
let (events, receiver) = mpsc::channel();
self.commands
.send(Command {
engine,
turn,
messages,
checkpoint: CheckpointTarget::Local {
checkpoint,
pub(crate) fn compact(&self, input: CompactionInput) -> Result<ActiveGeneration, String> {
self.submit(
CommandRequest {
engine: input.engine,
turn: input.turn,
messages: input.messages,
checkpoint: Some(CheckpointTarget::Local {
checkpoint: input.checkpoint,
bootstrap: None,
},
}),
source: WorkSource::LocalChat,
operation: Operation::Compact {
reason: reason.to_owned(),
rebuild_system_prompt,
reason: input.reason,
rebuild_system_prompt: input.rebuild_system_prompt,
},
idle_timeout,
cancel: Arc::clone(&cancel),
events,
})
.map_err(|_| "The model runtime stopped unexpectedly.".to_owned())?;
Ok(ActiveGeneration {
events: receiver,
cancel,
})
idle_timeout: input.idle_timeout,
},
TrackingPolicy::None,
)
}
pub(crate) fn measure_context(
@@ -182,21 +179,47 @@ impl GenerationService {
messages: Vec<ChatTurn>,
idle_timeout: Duration,
) -> Result<ActiveGeneration, String> {
let cancel = Arc::new(AtomicBool::new(false));
let (events, receiver) = mpsc::channel();
self.metrics.request_queued(WorkSource::LocalChat);
self.commands
.send(Command {
self.submit(
CommandRequest {
engine,
turn,
messages,
checkpoint: CheckpointTarget::OneShot(PathBuf::new()),
checkpoint: None,
source: WorkSource::LocalChat,
operation: Operation::Measure,
idle_timeout,
cancel: Arc::clone(&cancel),
events,
})
.map_err(|_| "The model runtime stopped unexpectedly.".to_owned())?;
},
TrackingPolicy::Queued,
)
}
fn submit(
&self,
request: CommandRequest,
tracking: TrackingPolicy,
) -> Result<ActiveGeneration, String> {
let cancel = Arc::new(AtomicBool::new(false));
let (events, receiver) = mpsc::channel();
if tracking.records_queue() {
self.metrics.request_queued(request.source);
}
let command = Command {
engine: request.engine,
turn: request.turn,
messages: request.messages,
checkpoint: request.checkpoint,
source: request.source,
operation: request.operation,
idle_timeout: request.idle_timeout,
cancel: Arc::clone(&cancel),
events,
};
if self.commands.send(command).is_err() {
if tracking.records_rejection() {
self.metrics.request_rejected();
}
return Err("The model runtime stopped unexpectedly.".to_owned());
}
Ok(ActiveGeneration {
events: receiver,
cancel,
@@ -204,44 +227,48 @@ impl GenerationService {
}
}
struct CommandRequest {
engine: EngineSettings,
turn: TurnSettings,
messages: Vec<ChatTurn>,
checkpoint: Option<CheckpointTarget>,
source: WorkSource,
operation: Operation,
idle_timeout: Duration,
}
fn run(commands: Receiver<Command>, metrics: Arc<Metrics>) {
let mut loaded = None::<(EngineSettings, Generator)>;
let mut last_used = Instant::now();
let mut idle_timeout = Duration::from_secs(15 * 60);
let mut state = RuntimeState {
loaded: None,
last_used: Instant::now(),
idle_timeout: Duration::from_secs(15 * 60),
};
loop {
match commands.recv_timeout(Duration::from_secs(1)) {
Ok(command) => {
let request_started = Instant::now();
let source = command.checkpoint.source();
let source = command.source;
let events = command.events.clone();
let response = response_kind(&command.operation);
let error_event = command.operation.error_handler();
let tracked = !matches!(command.operation, Operation::Measure);
if tracked {
metrics.request_started(source);
}
if let Err(error) = catch_runtime_panic(|| {
run_command(
command,
&mut loaded,
&metrics,
request_started,
source,
&mut last_used,
&mut idle_timeout,
);
run_command(command, &mut state, &metrics, request_started, source);
}) {
if tracked {
metrics.request_failed(request_started.elapsed());
}
if loaded.take().is_some() {
if state.loaded.take().is_some() {
metrics.unloaded();
}
let _ = events.send(error_event(response, error));
let _ = events.send(error_event(error));
}
}
Err(mpsc::RecvTimeoutError::Timeout) => {
if loaded.is_some() && last_used.elapsed() >= idle_timeout {
loaded = None;
if state.loaded.is_some() && state.last_used.elapsed() >= state.idle_timeout {
state.loaded = None;
metrics.unloaded();
}
}
@@ -250,39 +277,37 @@ fn run(commands: Receiver<Command>, metrics: Arc<Metrics>) {
}
}
#[allow(clippy::too_many_arguments)]
fn run_command(
command: Command,
loaded: &mut Option<(EngineSettings, Generator)>,
state: &mut RuntimeState,
metrics: &Arc<Metrics>,
request_started: Instant,
source: WorkSource,
last_used: &mut Instant,
idle_timeout: &mut Duration,
) {
let response = response_kind(&command.operation);
let error_event = command.operation.error_handler();
let tracked = !matches!(command.operation, Operation::Measure);
*idle_timeout = command.idle_timeout;
state.idle_timeout = command.idle_timeout;
if command.cancel.load(Ordering::Relaxed) {
if tracked {
metrics.request_failed(request_started.elapsed());
}
let _ = command
.events
.send(error_event(response, "generation cancelled".into()));
.send(error_event("generation cancelled".into()));
return;
}
if loaded
if state
.loaded
.as_ref()
.is_none_or(|(settings, _)| settings != &command.engine)
{
if loaded.take().is_some() {
if state.loaded.take().is_some() {
metrics.unloaded();
}
let _ = command.events.send(GenerationEvent::Loading);
metrics.loading();
let load_started = Instant::now();
*loaded = match Generator::open(&command.engine, Arc::clone(metrics)) {
state.loaded = match Generator::open(&command.engine, Arc::clone(metrics)) {
Ok(generator) => {
let summary = generator.summary();
metrics.loaded(
@@ -298,12 +323,12 @@ fn run_command(
if tracked {
metrics.request_failed(request_started.elapsed());
}
let _ = command.events.send(error_event(response, error));
let _ = command.events.send(error_event(error));
None
}
};
}
if let Some((_, generator)) = loaded {
if let Some((_, generator)) = &mut state.loaded {
let mut prefill_started = None::<(Instant, u32)>;
let mut emit = |reasoning, content| {
let _ = command
@@ -335,7 +360,7 @@ fn run_command(
rebuild_system_prompt,
} = &command.operation
{
let CheckpointTarget::Local { checkpoint, .. } = &command.checkpoint else {
let Some(CheckpointTarget::Local { checkpoint, .. }) = &command.checkpoint else {
unreachable!("compaction checkpoints are local")
};
let result = generator.compact(
@@ -357,16 +382,19 @@ fn run_command(
Err(_) => metrics.request_failed(request_started.elapsed()),
}
let _ = command.events.send(GenerationEvent::Compacted(result));
*last_used = Instant::now();
state.last_used = Instant::now();
return;
}
if matches!(command.operation, Operation::Measure) {
let result = generator.rendered_history_tokens(&command.messages, &command.turn);
let _ = command.events.send(GenerationEvent::Measured(result));
*last_used = Instant::now();
state.last_used = Instant::now();
return;
}
let result = match command.checkpoint {
let result = match command
.checkpoint
.expect("generation requires a checkpoint target")
{
CheckpointTarget::Local {
checkpoint,
bootstrap,
@@ -382,16 +410,14 @@ fn run_command(
let _ = command.events.send(GenerationEvent::Activity(activity));
},
),
CheckpointTarget::Transient(directory) | CheckpointTarget::OneShot(directory) => {
generator.generate_transient(
&directory,
&command.messages,
&command.turn,
&command.cancel,
&mut emit,
&mut progress,
)
}
CheckpointTarget::Transient(directory) => generator.generate_transient(
&directory,
&command.messages,
&command.turn,
&command.cancel,
&mut emit,
&mut progress,
),
};
match &result {
Ok(output) => metrics.request_finished(
@@ -406,23 +432,7 @@ fn run_command(
Err(_) => metrics.request_failed(request_started.elapsed()),
}
let _ = command.events.send(GenerationEvent::Finished(result));
*last_used = Instant::now();
}
}
fn response_kind(operation: &Operation) -> ResponseKind {
match operation {
Operation::Generate => ResponseKind::Generation,
Operation::Compact { .. } => ResponseKind::Compaction,
Operation::Measure => ResponseKind::Measurement,
}
}
fn error_event(kind: ResponseKind, error: String) -> GenerationEvent {
match kind {
ResponseKind::Generation => GenerationEvent::Finished(Err(error)),
ResponseKind::Compaction => GenerationEvent::Compacted(Err(error)),
ResponseKind::Measurement => GenerationEvent::Measured(Err(error)),
state.last_used = Instant::now();
}
}
@@ -461,11 +471,15 @@ mod tests {
#[test]
fn operation_failures_use_the_matching_event() {
assert!(matches!(
error_event(ResponseKind::Compaction, "stopped".into()),
Operation::Compact {
reason: String::new(),
rebuild_system_prompt: String::new(),
}
.error_handler()("stopped".into()),
GenerationEvent::Compacted(Err(error)) if error == "stopped"
));
assert!(matches!(
error_event(ResponseKind::Measurement, "stopped".into()),
Operation::Measure.error_handler()("stopped".into()),
GenerationEvent::Measured(Err(error)) if error == "stopped"
));
}