Skip to content

Keep only live taskman work in memory and drop unused startup load #253

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
iamwavecut merged 2 commits into from
Oct 13, 2026
Merged

Keep only live taskman work in memory and drop unused startup load #253

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3f3200f
Keep only live taskman work in memory and drop unused startup load
iamwavecut Oct 13, 2026
05cb980
Remove the outbox retry test's rows so other live tests do not claim …
iamwavecut Oct 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
105 changes: 56 additions & 49 deletions crates/openplotva-app/src/admin.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -966,7 +966,7 @@ where

match command.name {
ADMIN_QUEUE_STATUS_COMMAND => {
let text = admin_queue_status_text(&runtime);
let text = admin_queue_status_text(&runtime).await;
send_admin_queue_result(
runtime.effects,
message,
Expand Down Expand Up @@ -1835,10 +1835,10 @@ fn admin_queue_disabled_text() -> &'static str {
"❌ Персистентная система очередей отключена"
}

fn admin_queue_status_text<Effects>(runtime: &AdminQueueRuntime<'_, Effects>) -> String {
async fn admin_queue_status_text<Effects>(runtime: &AdminQueueRuntime<'_, Effects>) -> String {
let queue_names = runtime.config.queue_names();
let pending_by_queue = queue_counts_by_status(runtime.taskman, "pending");
let processing_by_queue = queue_counts_by_status(runtime.taskman, "processing");
let pending_by_queue = queue_counts_by_status(runtime.taskman, "pending").await;
let processing_by_queue = queue_counts_by_status(runtime.taskman, "processing").await;
let eta_by_queue = queue_eta_seconds(runtime.taskman, &queue_names);
let mut lines = Vec::with_capacity(queue_names.len() + 5);
lines.push("📊 <b>Статус очередей:</b>\n".to_owned());
Expand Down Expand Up @@ -1867,8 +1867,8 @@ fn admin_queue_status_text<Effects>(runtime: &AdminQueueRuntime<'_, Effects>) ->

async fn admin_queue_stats_text<Effects>(runtime: &AdminQueueRuntime<'_, Effects>) -> String {
let queue_names = runtime.config.queue_names();
let pending_by_queue = queue_counts_by_status(runtime.taskman, "pending");
let processing_by_queue = queue_counts_by_status(runtime.taskman, "processing");
let pending_by_queue = queue_counts_by_status(runtime.taskman, "pending").await;
let processing_by_queue = queue_counts_by_status(runtime.taskman, "processing").await;
let (total_pending, total_processing) =
queue_stats_totals(&queue_names, &pending_by_queue, &processing_by_queue);
let mut lines = Vec::new();
Expand All @@ -1883,11 +1883,14 @@ async fn admin_queue_stats_text<Effects>(runtime: &AdminQueueRuntime<'_, Effects
&processing_by_queue,
));
lines.push("\n🕒 <b>Старейшие:</b>".to_owned());
lines.extend(queue_stats_oldest_lines(
runtime.taskman,
&queue_names,
runtime.config.default_processing_timeout,
));
lines.extend(
queue_stats_oldest_lines(
runtime.taskman,
&queue_names,
runtime.config.default_processing_timeout,
)
.await,
);
lines.push("\n📮 <b>Очередь отправки:</b>".to_owned());
lines.extend(queue_stats_dispatcher_lines(
runtime
Expand Down Expand Up @@ -1915,18 +1918,20 @@ async fn admin_queue_stats_text<Effects>(runtime: &AdminQueueRuntime<'_, Effects
lines.join("\n")
}

fn queue_counts_by_status(
async fn queue_counts_by_status(
taskman: Option<&dyn RuntimeTaskmanInspector>,
status: &str,
) -> BTreeMap<String, i32> {
let Some(taskman) = taskman else {
return BTreeMap::new();
};
let result = taskman.list_jobs(RuntimeTaskmanJobsFilter {
status: vec![status.to_owned()],
limit: 1,
..RuntimeTaskmanJobsFilter::default()
});
let result = taskman
.list_jobs(RuntimeTaskmanJobsFilter {
status: vec![status.to_owned()],
limit: 1,
..RuntimeTaskmanJobsFilter::default()
})
.await;
match result {
Ok(result) => queue_count_map(&result.summary.by_queue),
Err(error) => {
Expand Down Expand Up @@ -2032,44 +2037,46 @@ fn queue_stats_count_lines(
.collect()
}

fn queue_stats_oldest_lines(
async fn queue_stats_oldest_lines(
taskman: Option<&dyn RuntimeTaskmanInspector>,
queue_names: &[String],
timeout: Duration,
) -> Vec<String> {
queue_names
.iter()
.map(|queue_name| {
let pending_age = queue_oldest_age(taskman, queue_name, "pending", "created_at");
let processing_age = queue_oldest_age(taskman, queue_name, "processing", "started_at");
let processing_flag = processing_age
.map(|age| queue_stats_processing_flag(age, timeout))
.unwrap_or_default();
format!(
"• {queue_name}: pending {}, processing {}{}",
format_queue_optional_age(pending_age),
format_queue_optional_age(processing_age),
processing_flag
)
})
.collect()
let mut lines = Vec::with_capacity(queue_names.len());
for queue_name in queue_names {
let pending_age = queue_oldest_age(taskman, queue_name, "pending", "created_at").await;
let processing_age =
queue_oldest_age(taskman, queue_name, "processing", "started_at").await;
let processing_flag = processing_age
.map(|age| queue_stats_processing_flag(age, timeout))
.unwrap_or_default();
lines.push(format!(
"• {queue_name}: pending {}, processing {}{}",
format_queue_optional_age(pending_age),
format_queue_optional_age(processing_age),
processing_flag
));
}
lines
}

fn queue_oldest_age(
async fn queue_oldest_age(
taskman: Option<&dyn RuntimeTaskmanInspector>,
queue_name: &str,
status: &str,
time_field: &str,
) -> Option<Duration> {
let taskman = taskman?;
let result = taskman.list_jobs(RuntimeTaskmanJobsFilter {
status: vec![status.to_owned()],
queue: vec![queue_name.to_owned()],
sort_by: time_field.to_owned(),
sort_dir: "asc".to_owned(),
limit: 1,
..RuntimeTaskmanJobsFilter::default()
});
let result = taskman
.list_jobs(RuntimeTaskmanJobsFilter {
status: vec![status.to_owned()],
queue: vec![queue_name.to_owned()],
sort_by: time_field.to_owned(),
sort_dir: "asc".to_owned(),
limit: 1,
..RuntimeTaskmanJobsFilter::default()
})
.await;
let result = match result {
Ok(result) => result,
Err(error) => {
Expand Down Expand Up @@ -2559,7 +2566,7 @@ fn format_go_fraction(value: u64, unit: u64, suffix: &str) -> String {
mod tests {
use super::*;
use openplotva_server::{
RuntimeGeminiCachePurgerFuture, RuntimeTaskmanDiagnosticsData,
RuntimeGeminiCachePurgerFuture, RuntimeTaskmanDiagnosticsData, RuntimeTaskmanJobListFuture,
RuntimeTaskmanJobListResultData, RuntimeTaskmanJobSummaryData,
RuntimeTaskmanQueueDiagnosticsData, RuntimeUpdatesInspectorFuture,
};
Expand Down Expand Up @@ -4320,10 +4327,10 @@ mod tests {
}

impl RuntimeTaskmanInspector for TaskmanInspectorStub {
fn list_jobs(
&self,
fn list_jobs<'a>(
&'a self,
filter: RuntimeTaskmanJobsFilter,
) -> Result<RuntimeTaskmanJobListResultData, String> {
) -> RuntimeTaskmanJobListFuture<'a> {
let status = filter.status.first().map(String::as_str).unwrap_or("");
let by_queue = match status {
"pending" => self.pending.clone(),
Expand Down Expand Up @@ -4351,7 +4358,7 @@ mod tests {
} else {
Vec::new()
};
Ok(RuntimeTaskmanJobListResultData {
Box::pin(std::future::ready(Ok(RuntimeTaskmanJobListResultData {
total: items.len() as i32,
offset: 0,
limit: filter.limit,
Expand All @@ -4360,7 +4367,7 @@ mod tests {
by_queue: json!(by_queue),
},
items,
})
})))
}

fn job<'a>(&'a self, _id: i64) -> openplotva_server::RuntimeTaskmanJobFuture<'a> {
Expand Down
6 changes: 3 additions & 3 deletions crates/openplotva-app/src/dialog_jobs/queue.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,8 @@
use std::fmt;

use openplotva_taskman::{
InMemoryTaskQueue, JobType, Priority, StatelessJobItem, TaskQueueError, TaskQueueJobEvent,
TaskQueueWorkItem,
AGENT_CHECKPOINT_EVENT_STAGE, InMemoryTaskQueue, JobType, Priority, StatelessJobItem,
TaskQueueError, TaskQueueJobEvent, TaskQueueWorkItem,
};
use time::OffsetDateTime;

Expand Down Expand Up @@ -146,7 +146,7 @@ impl DialogJobWorkerQueue for InMemoryTaskQueue {
at: OffsetDateTime,
) -> DialogJobWorkerFuture<'a, (), Self::Error> {
Box::pin(async move {
if event.stage == "agent_checkpoint" {
if event.stage == AGENT_CHECKPOINT_EVENT_STAGE {
self.upsert_job_event(job_id, event, at)
} else {
self.append_job_event(job_id, event, at)
Expand Down
31 changes: 21 additions & 10 deletions crates/openplotva-app/src/dialog_jobs/tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,9 +13,10 @@ use openplotva_llm::{
retry::{FailureReason, ProviderError},
};
use openplotva_taskman::{
DEFAULT_LLM_JOB_MAX_ATTEMPTS, DIALOG_AIFARM_QUEUE_NAME, DialogJobParams, HIGH_PRIORITY,
InMemoryTaskQueue, JobStatus, LLM_JOB_RETRY_EXHAUSTED_STAGE, LLM_JOB_RETRY_STAGE, Priority,
TEXT_QUEUE_NAME, TaskQueueError, TaskQueueJobEvent, new_dialog_job_at,
AGENT_CHECKPOINT_EVENT_STAGE, DEFAULT_LLM_JOB_MAX_ATTEMPTS, DIALOG_AIFARM_QUEUE_NAME,
DialogJobParams, HIGH_PRIORITY, InMemoryTaskQueue, JobStatus, LLM_JOB_RETRY_EXHAUSTED_STAGE,
LLM_JOB_RETRY_STAGE, Priority, TEXT_QUEUE_NAME, TaskQueueError, TaskQueueJobEvent,
new_dialog_job_at,
};
use openplotva_telegram::DispatcherQueue;
use time::{Duration as TimeDuration, OffsetDateTime};
Expand Down Expand Up @@ -7280,6 +7281,20 @@ async fn agent_retry_restores_context_and_completed_tool_results() -> Result<(),
)
.await;
assert!(first.retry_requeued, "{first:?}");
let checkpoints = |queue: &InMemoryTaskQueue| {
queue
.record(id)
.expect("record")
.events
.iter()
.filter(|event| event.stage == AGENT_CHECKPOINT_EVENT_STAGE)
.count()
};
assert_eq!(
checkpoints(&queue),
1,
"a requeued turn keeps one checkpoint"
);
let second = process_dialog_job_once_in_queue_with_materializer_history_and_retry_at(
&queue,
&provider,
Expand All @@ -7297,14 +7312,10 @@ async fn agent_retry_restores_context_and_completed_tool_results() -> Result<(),
message,
openplotva_dialog::SessionMessage::ToolResult { .. }
)));
let record = queue.record(id).expect("record");
assert_eq!(
record
.events
.iter()
.filter(|event| event.stage == "agent_checkpoint")
.count(),
1
checkpoints(&queue),
0,
"a finished turn drops its checkpoint"
);
Ok(())
}
Expand Down
6 changes: 3 additions & 3 deletions crates/openplotva-app/src/dialog_turn/session.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ use openplotva_dialog::{
},
};
use openplotva_llm::ChatStepProvider;
use openplotva_taskman::TaskQueueJobEvent;
use openplotva_taskman::{AGENT_CHECKPOINT_EVENT_STAGE, TaskQueueJobEvent};
use serde_json::Value;
use time::{Duration as TimeDuration, OffsetDateTime, format_description::well_known::Rfc3339};

Expand Down Expand Up @@ -466,7 +466,7 @@ where
.item_events
.iter()
.rev()
.find(|event| event.stage == "agent_checkpoint")
.find(|event| event.stage == AGENT_CHECKPOINT_EVENT_STAGE)
.and_then(|event| event.data.get("checkpoint"))
.map(|value| serde_json::from_str::<(DialogInput, SessionState)>(value))
.transpose();
Expand Down Expand Up @@ -603,7 +603,7 @@ where
let duplicate_guard_history = &duplicate_guard_history;
let run_id = &run_id;
openplotva_agent::run((state, report), |(state, report)| async move {
if let Err(error) = persist_agent_event(queue, effects, ctx.item_id, "agent_checkpoint", BTreeMap::from([("checkpoint".into(), serde_json::to_string(&(base_input, &state)).expect("serializable agent state"))]), ctx.now).await {
if let Err(error) = persist_agent_event(queue, effects, ctx.item_id, AGENT_CHECKPOINT_EVENT_STAGE, BTreeMap::from([("checkpoint".into(), serde_json::to_string(&(base_input, &state)).expect("serializable agent state"))]), ctx.now).await {
return std::ops::ControlFlow::Break(agent_persistence_failed(error));
}
let SessionState { mut requested_tool_call_offset, mut tool_context, mut gift_used, mut budget, mut agent, mut sent, mut side_effect_tickets, mut recorded_tool_calls, mut regenerations, mut anti_loop, mut repeated_final_repair, mut requires_novel_final, mut draws_scheduled, mut songs_scheduled, mut reacted_message_ids, mut tool_result_cache, mut media_reference_aliases, mut successful_web_search, mut web_source_urls, mut links, mut search_citation_repairs, mut repeated_tool_call } = state;
Expand Down
Loading
Loading