From 4e0bac870d741194e4f75f24365853f5d6288e70 Mon Sep 17 00:00:00 2001 From: Valeriy Selitskiy <239034+iamwavecut@users.noreply.github.com> Date: Thu, 15 Oct 2026 16:32:13 +0200 Subject: [PATCH] Fix memory scheduling, Telegram topics, and context diagnostics --- .env.example | 6 +- crates/openplotva-app/src/admin.rs | 4 + crates/openplotva-app/src/checkin.rs | 3 + crates/openplotva-app/src/delete_drawing.rs | 10 + crates/openplotva-app/src/delete_lyrics.rs | 5 + crates/openplotva-app/src/delete_message.rs | 4 + crates/openplotva-app/src/diagnostics.rs | 4 + .../openplotva-app/src/dialog_jobs/effects.rs | 10 + .../openplotva-app/src/dialog_jobs/tests.rs | 35 ++ crates/openplotva-app/src/dialog_messages.rs | 2 + .../src/dialog_turn/obligations.rs | 1 + .../openplotva-app/src/dialog_turn/signal.rs | 1 + crates/openplotva-app/src/gradius_ads.rs | 1 + .../src/gradius_utility_outbox.rs | 2 + crates/openplotva-app/src/help.rs | 5 + crates/openplotva-app/src/image_jobs.rs | 4 + crates/openplotva-app/src/lib.rs | 7 +- crates/openplotva-app/src/memory_runtime.rs | 47 +-- crates/openplotva-app/src/music_jobs.rs | 1 + crates/openplotva-app/src/payments.rs | 15 + crates/openplotva-app/src/rates.rs | 7 + crates/openplotva-app/src/reset.rs | 4 + .../src/routing_admin_reports.rs | 1 + crates/openplotva-app/src/runtime_llm.rs | 101 ++++++ crates/openplotva-app/src/settings.rs | 11 + .../openplotva-app/src/telegram_activity.rs | 1 + crates/openplotva-app/src/translate.rs | 3 + crates/openplotva-app/src/virtual_messages.rs | 90 ++++- crates/openplotva-config/src/lib.rs | 34 -- crates/openplotva-core/src/lib.rs | 3 + crates/openplotva-llm/src/aifarm.rs | 79 +++++ crates/openplotva-memory/src/lib.rs | 8 - crates/openplotva-storage/src/lib.rs | 324 ++++++++++-------- crates/openplotva-telegram/src/outbound.rs | 87 ++++- crates/openplotva-telegram/src/persistence.rs | 100 +++++- crates/openplotva-telegram/src/rich_api.rs | 17 + crates/openplotva-telegram/src/transport.rs | 1 + crates/openplotva-updates/src/lib.rs | 27 ++ 38 files changed, 810 insertions(+), 255 deletions(-) diff --git a/.env.example b/.env.example index 80a07ec0..32058e09 100644 --- a/.env.example +++ b/.env.example @@ -292,12 +292,12 @@ # MEMORY_CONSOLIDATION_PROVIDER=aifarm # MEMORY_CONSOLIDATION_MODEL=qwen3.8-27b # MEMORY_CONSOLIDATION_TIMEOUT_SECONDS=600 -# Each chat hour with enough messages becomes a memory run at HH:02 UTC, after -# the hour completes; MEMORY_MAX_QUEUED_RUNS bounds the active queue. +# Startup and each HH:02 UTC tick enqueue all missing completed chat hours +# within MEMORY_RETENTION_HOURS. Workers drain the queue continuously, with +# fresh windows first. MEMORY_MAX_QUEUED_RUNS is obsolete and ignored. # MEMORY_WORKER_CONCURRENCY=1 # MEMORY_MAX_MESSAGES_PER_RUN=200 # MEMORY_MIN_MESSAGES_PER_RUN=20 -# MEMORY_MAX_QUEUED_RUNS=5000 # Visible-card lists the memory pipeline keeps in RAM (LRU); 0 disables it. # MEMORY_CARD_CACHE_ENTRIES=4096 # MEMORY_MAX_INPUT_TOKENS=10000 diff --git a/crates/openplotva-app/src/admin.rs b/crates/openplotva-app/src/admin.rs index f894a69f..da61f343 100644 --- a/crates/openplotva-app/src/admin.rs +++ b/crates/openplotva-app/src/admin.rs @@ -2456,6 +2456,10 @@ fn admin_text_plan( ephemeral_delete_after: Option, ) -> AdminTextPlan { let chat = ChatRef { + direct_messages_topic_id: message + .direct_messages_topic + .as_ref() + .map(|topic| topic.topic_id), id: message.chat.get_id().into(), is_forum: message.message_thread_id.is_some(), }; diff --git a/crates/openplotva-app/src/checkin.rs b/crates/openplotva-app/src/checkin.rs index 9f878bf9..0105d476 100644 --- a/crates/openplotva-app/src/checkin.rs +++ b/crates/openplotva-app/src/checkin.rs @@ -717,6 +717,7 @@ impl CheckinCommandEffects for CheckinCommandDispatcherEffects { ) -> CheckinCommandEffectFuture<'a, Self::Error> { Box::pin(async move { let options = openplotva_telegram::RichSendOptions { + direct_messages_topic_id: None, message_thread_id: reply_to.thread_id.map(i64::from), reply_to_message_id: Some(i64::from(reply_to.message_id)), allow_sending_without_reply: true, @@ -965,6 +966,7 @@ where fn checkin_chat_ref(message: CheckinGameMessage) -> ChatRef { ChatRef { + direct_messages_topic_id: None, id: message.chat_id, is_forum: message.thread_id.is_some(), } @@ -2142,6 +2144,7 @@ where ) -> CheckinThemeEffectFuture<'a, Self::Error> { Box::pin(async move { let chat = ChatRef { + direct_messages_topic_id: None, id: message.chat_id, is_forum: message.thread_id.is_some(), }; diff --git a/crates/openplotva-app/src/delete_drawing.rs b/crates/openplotva-app/src/delete_drawing.rs index 111a0072..a30a6050 100644 --- a/crates/openplotva-app/src/delete_drawing.rs +++ b/crates/openplotva-app/src/delete_drawing.rs @@ -886,6 +886,7 @@ where effects, EditCaptionMessageRequest { chat: ChatRef { + direct_messages_topic_id: None, id: data.chat_id, is_forum: false, }, @@ -1238,6 +1239,10 @@ fn delete_drawing_control_message( match query.message.as_ref()? { MaybeInaccessibleMessage::Message(message) => Some(DeleteDrawingControlMessage { chat: ChatRef { + direct_messages_topic_id: message + .direct_messages_topic + .as_ref() + .map(|topic| topic.topic_id), id: message.chat.get_id().into(), is_forum: message.message_thread_id.is_some(), }, @@ -1246,6 +1251,7 @@ fn delete_drawing_control_message( MaybeInaccessibleMessage::InaccessibleMessage(message) => { Some(DeleteDrawingControlMessage { chat: ChatRef { + direct_messages_topic_id: None, id: message.chat.get_id().into(), is_forum: false, }, @@ -1337,6 +1343,10 @@ fn base_text_request( fn message_chat_ref(message: &TelegramMessage) -> ChatRef { ChatRef { + direct_messages_topic_id: message + .direct_messages_topic + .as_ref() + .map(|topic| topic.topic_id), id: message.chat.get_id().into(), is_forum: message.message_thread_id.is_some(), } diff --git a/crates/openplotva-app/src/delete_lyrics.rs b/crates/openplotva-app/src/delete_lyrics.rs index fdc86c0c..259df1c8 100644 --- a/crates/openplotva-app/src/delete_lyrics.rs +++ b/crates/openplotva-app/src/delete_lyrics.rs @@ -408,6 +408,10 @@ fn delete_lyrics_control_message( match query.message.as_ref()? { MaybeInaccessibleMessage::Message(message) => Some(DeleteLyricsControlMessage { chat: ChatRef { + direct_messages_topic_id: message + .direct_messages_topic + .as_ref() + .map(|topic| topic.topic_id), id: message.chat.get_id().into(), is_forum: message.message_thread_id.is_some(), }, @@ -416,6 +420,7 @@ fn delete_lyrics_control_message( MaybeInaccessibleMessage::InaccessibleMessage(message) => { Some(DeleteLyricsControlMessage { chat: ChatRef { + direct_messages_topic_id: None, id: message.chat.get_id().into(), is_forum: false, }, diff --git a/crates/openplotva-app/src/delete_message.rs b/crates/openplotva-app/src/delete_message.rs index c9e05a24..086715c5 100644 --- a/crates/openplotva-app/src/delete_message.rs +++ b/crates/openplotva-app/src/delete_message.rs @@ -305,6 +305,10 @@ fn delete_message_callback_target( }; Some(DeleteMessageCallbackTarget { message_chat: ChatRef { + direct_messages_topic_id: message + .direct_messages_topic + .as_ref() + .map(|topic| topic.topic_id), id: message.chat.get_id().into(), is_forum: message.message_thread_id.is_some(), }, diff --git a/crates/openplotva-app/src/diagnostics.rs b/crates/openplotva-app/src/diagnostics.rs index d507e31b..7e7a1fd7 100644 --- a/crates/openplotva-app/src/diagnostics.rs +++ b/crates/openplotva-app/src/diagnostics.rs @@ -488,6 +488,10 @@ fn telegram_chat_is_private(message: &TelegramMessage) -> bool { fn message_chat_ref(message: &TelegramMessage) -> ChatRef { ChatRef { + direct_messages_topic_id: message + .direct_messages_topic + .as_ref() + .map(|topic| topic.topic_id), id: message.chat.get_id().into(), is_forum: message.message_thread_id.is_some(), } diff --git a/crates/openplotva-app/src/dialog_jobs/effects.rs b/crates/openplotva-app/src/dialog_jobs/effects.rs index 510638ca..5b7a5dde 100644 --- a/crates/openplotva-app/src/dialog_jobs/effects.rs +++ b/crates/openplotva-app/src/dialog_jobs/effects.rs @@ -230,6 +230,10 @@ impl DialogJobEffects for DialogDispatcherEffects { ) -> DialogJobReceiptFuture<'a, Self::Error> { Box::pin(async move { let chat = ChatRef { + direct_messages_topic_id: params + .meta + .get("direct_messages_topic_id") + .and_then(serde_json::Value::as_i64), id: params.chat_id, is_forum: params.thread_id.is_some(), }; @@ -384,6 +388,10 @@ impl DialogJobEffects for DialogDispatcherEffects { ) -> DialogJobEffectFuture<'a, Self::Error> { Box::pin(async move { let chat = ChatRef { + direct_messages_topic_id: params + .meta + .get("direct_messages_topic_id") + .and_then(serde_json::Value::as_i64), id: params.chat_id, is_forum: params.thread_id.is_some(), }; @@ -727,6 +735,7 @@ mod tests { #[test] fn durable_answer_payload_disables_preview_on_every_text_part() { let chat = ChatRef { + direct_messages_topic_id: None, id: -100, is_forum: true, }; @@ -785,6 +794,7 @@ mod tests { #[test] fn advertising_final_text_disables_preview_in_durable_payload() { let chat = ChatRef { + direct_messages_topic_id: None, id: 42, is_forum: false, }; diff --git a/crates/openplotva-app/src/dialog_jobs/tests.rs b/crates/openplotva-app/src/dialog_jobs/tests.rs index e404556d..b77ce364 100644 --- a/crates/openplotva-app/src/dialog_jobs/tests.rs +++ b/crates/openplotva-app/src/dialog_jobs/tests.rs @@ -414,6 +414,41 @@ async fn dialog_worker_skips_resend_when_answer_sent_marker_present() -> Result< Ok(()) } +#[tokio::test] +async fn dialog_dispatcher_keeps_direct_messages_topic_through_replay() -> Result<(), Box> +{ + let queue = Arc::new(DispatcherQueue::new( + openplotva_telegram::DispatcherConfig::default(), + )); + let effects = DialogDispatcherEffects::new(Arc::clone(&queue)); + let mut params = dialog_params("hello"); + params.thread_id = None; + params.meta = serde_json::json!({"direct_messages_topic_id": 4_294_967_301_i64}); + effects + .send_dialog_answer(1, None, ¶ms, "pong", DialogAnswerSendOptions::default()) + .await?; + let item = queue.dequeue_immediate().expect("answer"); + let (_, method) = item.into_parts(); + let command = openplotva_telegram::OutboundCommand::try_from_method(method.expect("method"))?; + let payload = command.payload().expect("durable payload"); + assert_eq!(payload["direct_messages_topic_id"], 4_294_967_301_i64); + assert!(payload.get("message_thread_id").is_none()); + let replay = openplotva_telegram::OutboundCommand::decode( + 1, + "sendMessage", + &serde_json::to_vec(payload)?, + )?; + let openplotva_telegram::TelegramOutboundMethod::SendMessage(method) = replay.into_method() + else { + panic!("sendMessage"); + }; + assert_eq!( + serde_json::to_value(method)?["direct_messages_topic_id"], + 4_294_967_301_i64 + ); + Ok(()) +} + #[tokio::test] async fn dialog_dispatcher_effects_allow_sending_without_deleted_reply_target() -> Result<(), Box> { diff --git a/crates/openplotva-app/src/dialog_messages.rs b/crates/openplotva-app/src/dialog_messages.rs index d4dfdc6c..286030ec 100644 --- a/crates/openplotva-app/src/dialog_messages.rs +++ b/crates/openplotva-app/src/dialog_messages.rs @@ -707,6 +707,7 @@ impl crate::dialog_tools::DrawQueueNoticeSender for DrawQueueNoticeDispatcherSen let message_thread_id = i64::from(notice.thread_id.unwrap_or_default()); let is_topic_message = notice.thread_id.is_some(); let chat = ChatRef { + direct_messages_topic_id: None, id: chat_id, is_forum: is_topic_message, }; @@ -2055,6 +2056,7 @@ fn direct_draw_api_photo_request( })?; Some(PhotoMessageRequest { chat: ChatRef { + direct_messages_topic_id: None, id: request.chat_id, is_forum: request.is_forum, }, diff --git a/crates/openplotva-app/src/dialog_turn/obligations.rs b/crates/openplotva-app/src/dialog_turn/obligations.rs index 7ac5a972..924e1a7f 100644 --- a/crates/openplotva-app/src/dialog_turn/obligations.rs +++ b/crates/openplotva-app/src/dialog_turn/obligations.rs @@ -452,6 +452,7 @@ impl DeliveryObligationNotifier for DispatcherDeliveryObligationNotifier { None }; let chat = ChatRef { + direct_messages_topic_id: None, id: target.chat_id, is_forum: target.thread_id.is_some(), }; diff --git a/crates/openplotva-app/src/dialog_turn/signal.rs b/crates/openplotva-app/src/dialog_turn/signal.rs index e7da474e..999dd5b4 100644 --- a/crates/openplotva-app/src/dialog_turn/signal.rs +++ b/crates/openplotva-app/src/dialog_turn/signal.rs @@ -162,6 +162,7 @@ impl DispatcherTerminalUserSignal { async fn send_text_fallback(&self, target: &SignalTarget) -> Result<(), String> { let chat = ChatRef { + direct_messages_topic_id: None, id: target.chat_id, is_forum: target.thread_id.is_some(), }; diff --git a/crates/openplotva-app/src/gradius_ads.rs b/crates/openplotva-app/src/gradius_ads.rs index dd3d8a36..d5afc5a3 100644 --- a/crates/openplotva-app/src/gradius_ads.rs +++ b/crates/openplotva-app/src/gradius_ads.rs @@ -1888,6 +1888,7 @@ mod tests { .execute(&pool) .await?; let chat = ChatRef { + direct_messages_topic_id: None, id: chat_id, is_forum: false, }; diff --git a/crates/openplotva-app/src/gradius_utility_outbox.rs b/crates/openplotva-app/src/gradius_utility_outbox.rs index 190df83e..5dd72b93 100644 --- a/crates/openplotva-app/src/gradius_utility_outbox.rs +++ b/crates/openplotva-app/src/gradius_utility_outbox.rs @@ -353,6 +353,7 @@ pub(crate) fn image_ad_message( return Err("Ephemeral image advertising requires a group chat".to_owned()); } let chat = ChatRef { + direct_messages_topic_id: None, id: chat_id, is_forum: thread_id.is_some(), }; @@ -412,6 +413,7 @@ fn utility_rich_send( chat_id, html: utility_rich_html(html)?, options: RichSendOptions { + direct_messages_topic_id: None, message_thread_id: thread_id.map(i64::from), reply_to_message_id: Some(reply_to), ..Default::default() diff --git a/crates/openplotva-app/src/help.rs b/crates/openplotva-app/src/help.rs index be719bcc..e5856942 100644 --- a/crates/openplotva-app/src/help.rs +++ b/crates/openplotva-app/src/help.rs @@ -192,6 +192,7 @@ where .as_ref() .map_or(plan.reply_to.chat.id, |chat| chat.id); let options = openplotva_telegram::RichSendOptions { + direct_messages_topic_id: None, message_thread_id: if plan.message.message_thread_id != 0 { Some(plan.message.message_thread_id) } else if plan.reply_to.is_topic_message { @@ -582,6 +583,10 @@ fn help_markup(text: &str, url: String) -> ReplyMarkup { fn message_chat_ref(message: &TelegramMessage) -> ChatRef { ChatRef { + direct_messages_topic_id: message + .direct_messages_topic + .as_ref() + .map(|topic| topic.topic_id), id: message.chat.get_id().into(), is_forum: message.message_thread_id.is_some(), } diff --git a/crates/openplotva-app/src/image_jobs.rs b/crates/openplotva-app/src/image_jobs.rs index 190f5123..52500fc4 100644 --- a/crates/openplotva-app/src/image_jobs.rs +++ b/crates/openplotva-app/src/image_jobs.rs @@ -2689,6 +2689,7 @@ fn nsfw_blocked_message_request( ReplyMessageRef { message_id: i64::from(message_id), chat: ChatRef { + direct_messages_topic_id: None, id: chat_id, is_forum: is_topic_message, }, @@ -2709,6 +2710,7 @@ fn generated_image_message_request( let is_topic_message = message_thread_id != 0; PhotoMessageRequest { chat: ChatRef { + direct_messages_topic_id: None, id: chat_id, is_forum: is_topic_message, }, @@ -2757,6 +2759,7 @@ fn placeholder_image_media_group_request( let is_topic_message = message_thread_id != 0; MediaGroupMessageRequest { chat: ChatRef { + direct_messages_topic_id: None, id: chat_id, is_forum: is_topic_message, }, @@ -2795,6 +2798,7 @@ fn replacement_image_message_request( ) -> EditMediaMessageRequest { EditMediaMessageRequest { chat: ChatRef { + direct_messages_topic_id: None, id: chat_id, is_forum: false, }, diff --git a/crates/openplotva-app/src/lib.rs b/crates/openplotva-app/src/lib.rs index 15b2588d..5aae2dd8 100644 --- a/crates/openplotva-app/src/lib.rs +++ b/crates/openplotva-app/src/lib.rs @@ -4377,7 +4377,7 @@ async fn admin_memory_runs_list_response( fn admin_memory_policy_json(routes: &StaticWebRoutes) -> serde_json::Value { serde_json::json!({ "min_messages_per_run": routes.memory_enqueue_policy.min_messages_per_run, - "max_queued_runs": routes.memory_enqueue_policy.max_queued_runs, + "max_queued_runs": null, "max_messages_per_run": routes.memory_max_messages_per_run, "max_input_tokens": routes.memory_max_input_tokens, "token_estimator": routes.memory_token_estimator_source.as_ref(), @@ -16715,7 +16715,10 @@ mod tests { assert_eq!(value["count"], 0); assert!(value["runs"].is_null()); assert_eq!(value["policy"]["min_messages_per_run"], 20); - assert_eq!(value["policy"]["max_queued_runs"], 5000); + assert_eq!( + value["policy"].get("max_queued_runs"), + Some(&serde_json::Value::Null) + ); assert!(value["policy"].get("max_daily_enqueued_runs").is_none()); assert_eq!(value["policy"]["token_estimator"], "heuristic"); assert!(value["enqueue_rollups"].is_null()); diff --git a/crates/openplotva-app/src/memory_runtime.rs b/crates/openplotva-app/src/memory_runtime.rs index aaf9ed33..c9363c13 100644 --- a/crates/openplotva-app/src/memory_runtime.rs +++ b/crates/openplotva-app/src/memory_runtime.rs @@ -1424,12 +1424,7 @@ where if enqueue_runs { match store - .ensure_hourly_runs( - now, - cfg.retention, - memory_enqueue_lookback(now, cfg.retention), - cfg.enqueue_policy, - ) + .ensure_hourly_runs(now, cfg.retention, cfg.retention, cfg.enqueue_policy) .await { Ok(queued) => report.queued_runs = queued, @@ -1485,12 +1480,7 @@ where if enqueue_runs { match store - .ensure_hourly_runs( - now, - cfg.retention, - memory_enqueue_lookback(now, cfg.retention), - cfg.enqueue_policy, - ) + .ensure_hourly_runs(now, cfg.retention, cfg.retention, cfg.enqueue_policy) .await { Ok(queued) => report.queued_runs = queued, @@ -2523,26 +2513,6 @@ fn estimate_extraction_tokens( /// Minutes past the hour at which the scheduler enqueues the hour that just /// completed, so history writes that trail the hour boundary are included. const MEMORY_ENQUEUE_MINUTE: u8 = 2; -/// Completed hours an hourly tick looks back over; a tick that ran late or a -/// short outage still lands its hours. -pub const MEMORY_RECENT_LOOKBACK: Duration = Duration::from_secs(2 * 60 * 60); -/// UTC hour whose tick rescans the whole retention window for hours a longer -/// outage or a full queue left without a run. -const MEMORY_FULL_SCAN_HOUR: u8 = 3; - -/// History span one enqueue tick scans: the whole retention window once a day, -/// otherwise the recent hours. Startup scans recent hours too, so a deploy does -/// not refill the queue while the process catches up; the daily scan covers -/// hours missed while it was down. -#[must_use] -pub fn memory_enqueue_lookback(now: OffsetDateTime, retention: Duration) -> Duration { - if now.to_offset(time::UtcOffset::UTC).hour() == MEMORY_FULL_SCAN_HOUR { - retention - } else { - MEMORY_RECENT_LOOKBACK.min(retention) - } -} - /// Time from `now` until the next enqueue tick at `HH:02` UTC. #[must_use] pub fn next_memory_enqueue_delay(now: OffsetDateTime) -> Duration { @@ -4678,7 +4648,6 @@ pub fn memory_service_worker_config_from_memory_config( retention: positive_hours(config.retention_hours, 7 * 24), enqueue_policy: openplotva_memory::MemoryRunEnqueuePolicy { min_messages_per_run: config.min_messages_per_run, - max_queued_runs: config.max_queued_runs, }, process: MemoryRunProcessConfig { lease: positive_seconds(config.consolidation_timeout_seconds), @@ -5663,7 +5632,6 @@ mod tests { fn memory_service_worker_config_includes_enqueue_policy() { let config = AppConfig::from_raw(openplotva_config::RawConfig { memory_min_messages_per_run: Some("21".to_owned()), - memory_max_queued_runs: Some("4321".to_owned()), ..openplotva_config::RawConfig::default() }) .expect("config"); @@ -5671,7 +5639,6 @@ mod tests { let cfg = memory_service_worker_config_from_memory_config(&config.memory); assert_eq!(cfg.enqueue_policy.min_messages_per_run, 21); - assert_eq!(cfg.enqueue_policy.max_queued_runs, 4321); } #[test] @@ -6600,7 +6567,7 @@ mod tests { } #[tokio::test] - async fn every_scheduler_tick_enqueues_completed_hours() { + async fn startup_and_every_scheduler_tick_cover_the_full_retention_window() { let store = FakeMemoryWriteStore { ensure_hourly_count: 2, ..FakeMemoryWriteStore::default() @@ -6646,12 +6613,8 @@ mod tests { .collect(); assert_eq!( lookbacks, - vec![ - MEMORY_RECENT_LOOKBACK, - cfg.retention, - MEMORY_RECENT_LOOKBACK - ], - "hourly ticks and startup scan recent hours; the daily tick scans the retention window" + vec![cfg.retention; 3], + "daytime, nighttime, and startup recover all missed retained hours" ); } diff --git a/crates/openplotva-app/src/music_jobs.rs b/crates/openplotva-app/src/music_jobs.rs index bf6aacd3..394cc9a7 100644 --- a/crates/openplotva-app/src/music_jobs.rs +++ b/crates/openplotva-app/src/music_jobs.rs @@ -1155,6 +1155,7 @@ where footer_html: &footer_html, }); let options = openplotva_telegram::RichSendOptions { + direct_messages_topic_id: None, message_thread_id: plan.thread_id.map(i64::from), reply_to_message_id: Some(i64::from(plan.message_id)), allow_sending_without_reply: true, diff --git a/crates/openplotva-app/src/payments.rs b/crates/openplotva-app/src/payments.rs index 422908ea..cf7cf98b 100644 --- a/crates/openplotva-app/src/payments.rs +++ b/crates/openplotva-app/src/payments.rs @@ -2147,6 +2147,7 @@ where ) -> PaymentEffectsFuture<'a, Self::Error> { Box::pin(async move { let chat = openplotva_telegram::ChatRef { + direct_messages_topic_id: None, id: message.chat_id, is_forum: false, }; @@ -2390,6 +2391,7 @@ pub fn invoice_button_send_message_method( ) -> Result { openplotva_telegram::validate_text_message_text(&message.text, &message.parse_mode)?; let chat = openplotva_telegram::ChatRef { + direct_messages_topic_id: None, id: message.chat_id, is_forum: false, }; @@ -2415,6 +2417,7 @@ fn invoice_error_text_send_message_methods( parse_mode: &str, ) -> Result, openplotva_telegram::OutboundBuildError> { let chat = openplotva_telegram::ChatRef { + direct_messages_topic_id: None, id: chat_id, is_forum: false, }; @@ -2549,6 +2552,7 @@ pub fn payment_redirect_send_message_method( ) -> Result { openplotva_telegram::validate_text_message_text(&message.text, "")?; let chat = openplotva_telegram::ChatRef { + direct_messages_topic_id: None, id: message.chat_id, is_forum: message.message_thread_id.is_some(), }; @@ -2663,6 +2667,7 @@ impl VipStatusEffects for RichPaymentEffects { }) }); let options = openplotva_telegram::RichSendOptions { + direct_messages_topic_id: None, message_thread_id: message.message_thread_id.map(i64::from), reply_to_message_id: Some(i64::from(message.reply_to_message_id)), allow_sending_without_reply: true, @@ -2821,6 +2826,7 @@ pub fn vip_status_send_message_method( openplotva_telegram::TELEGRAM_PARSE_MODE_HTML, )?; let chat = openplotva_telegram::ChatRef { + direct_messages_topic_id: None, id: message.chat_id, is_forum: message.message_thread_id.is_some(), }; @@ -3211,6 +3217,10 @@ fn vip_callback_message(query: &TelegramCallbackQuery) -> Option Some(VipCallbackMessage { chat: openplotva_telegram::ChatRef { + direct_messages_topic_id: message + .direct_messages_topic + .as_ref() + .map(|topic| topic.topic_id), id: message.chat.get_id().into(), is_forum: message.message_thread_id.is_some(), }, @@ -3218,6 +3228,7 @@ fn vip_callback_message(query: &TelegramCallbackQuery) -> Option Some(VipCallbackMessage { chat: openplotva_telegram::ChatRef { + direct_messages_topic_id: None, id: message.chat.get_id().into(), is_forum: false, }, @@ -4876,6 +4887,10 @@ async fn send_admin_command_ephemeral_text( text: &str, ) -> Result<(), openplotva_telegram::OutboundBuildError> { let chat = openplotva_telegram::ChatRef { + direct_messages_topic_id: message + .direct_messages_topic + .as_ref() + .map(|topic| topic.topic_id), id: message.chat.get_id().into(), is_forum: message.message_thread_id.is_some(), }; diff --git a/crates/openplotva-app/src/rates.rs b/crates/openplotva-app/src/rates.rs index b434b5dc..a26b89d6 100644 --- a/crates/openplotva-app/src/rates.rs +++ b/crates/openplotva-app/src/rates.rs @@ -1291,6 +1291,7 @@ impl RatesEffects for RatesRichEffects { .as_ref() .map_or(plan.reply_to.chat.id, |chat| chat.id); let options = openplotva_telegram::RichSendOptions { + direct_messages_topic_id: None, message_thread_id: if plan.message.message_thread_id != 0 { Some(plan.message.message_thread_id) } else if plan.reply_to.is_topic_message { @@ -1355,6 +1356,7 @@ impl RatesSideEffectDispatcher for RatesToolRichEffects { } let options = openplotva_telegram::RichSendOptions { + direct_messages_topic_id: None, message_thread_id: meta.thread_id.map(i64::from), reply_to_message_id: Some(i64::from(meta.message_id)), allow_sending_without_reply: true, @@ -1442,6 +1444,7 @@ impl RatesSideEffectDispatcher for RatesToolDispatcherEffects { } let chat = ChatRef { + direct_messages_topic_id: None, id: meta.chat_id, is_forum: meta.thread_id.is_some(), }; @@ -2055,6 +2058,10 @@ fn unavailable_rates_plan(message: &TelegramMessage) -> RatesTextPlan { fn rates_text_plan(message: &TelegramMessage, text: String) -> RatesTextPlan { let chat = ChatRef { + direct_messages_topic_id: message + .direct_messages_topic + .as_ref() + .map(|topic| topic.topic_id), id: message.chat.get_id().into(), is_forum: message.message_thread_id.is_some(), }; diff --git a/crates/openplotva-app/src/reset.rs b/crates/openplotva-app/src/reset.rs index d19c2bd6..0399fa1c 100644 --- a/crates/openplotva-app/src/reset.rs +++ b/crates/openplotva-app/src/reset.rs @@ -359,6 +359,10 @@ where fn message_chat_ref(message: &TelegramMessage) -> ChatRef { ChatRef { + direct_messages_topic_id: message + .direct_messages_topic + .as_ref() + .map(|topic| topic.topic_id), id: message.chat.get_id().into(), is_forum: message.message_thread_id.is_some(), } diff --git a/crates/openplotva-app/src/routing_admin_reports.rs b/crates/openplotva-app/src/routing_admin_reports.rs index c3aa6782..3675a7cd 100644 --- a/crates/openplotva-app/src/routing_admin_reports.rs +++ b/crates/openplotva-app/src/routing_admin_reports.rs @@ -194,6 +194,7 @@ pub fn build_admin_report_dispatch( virtual_id: &str, ) -> Result { let chat = ChatRef { + direct_messages_topic_id: None, id: admin_id, is_forum: false, }; diff --git a/crates/openplotva-app/src/runtime_llm.rs b/crates/openplotva-app/src/runtime_llm.rs index 5ee07af5..73e1f56e 100644 --- a/crates/openplotva-app/src/runtime_llm.rs +++ b/crates/openplotva-app/src/runtime_llm.rs @@ -1277,6 +1277,63 @@ fn capture_raw_body(value: Option<&Value>, policy: RawBodyPolicy) -> RawBodyCapt } } +/// Fixed-key counters survive raw-body omission and retention without retaining +/// message text, URLs, tool arguments, names, or caller-supplied field names. +fn request_shape(value: &Value) -> Option { + let messages = value.get("messages")?.as_array()?; + let mut shape = json!({ + "version": 1, "messages": messages.len(), "text_chars": 0, + "system_chars": 0, "user_chars": 0, "assistant_chars": 0, + "tool_chars": 0, "other_chars": 0, "max_message_chars": 0, + "image_parts": 0, "audio_parts": 0, "video_parts": 0, + "other_parts": 0, "tool_calls": 0, + "tool_definitions": value.get("tools").and_then(Value::as_array).map_or(0, Vec::len), + "tool_definition_bytes": value.get("tools").and_then(|v| serde_json::to_vec(v).ok()).map_or(0, |v| v.len()) + }); + let add = |shape: &mut Value, key: &str, count: usize| { + shape[key] = json!( + shape[key] + .as_u64() + .unwrap_or_default() + .saturating_add(count as u64) + ); + }; + for message in messages { + let content = &message["content"]; + let mut chars = content.as_str().map_or(0, |text| text.chars().count()); + if let Some(parts) = content.as_array() { + for part in parts { + match part["type"].as_str() { + Some("text") => { + chars += part["text"].as_str().map_or(0, |text| text.chars().count()) + } + Some("image_url") => add(&mut shape, "image_parts", 1), + Some("input_audio") => add(&mut shape, "audio_parts", 1), + Some("video_url") => add(&mut shape, "video_parts", 1), + _ => add(&mut shape, "other_parts", 1), + } + } + } + let role_key = match message["role"].as_str() { + Some("system" | "developer") => "system_chars", + Some("user") => "user_chars", + Some("assistant") => "assistant_chars", + Some("tool") => "tool_chars", + _ => "other_chars", + }; + add(&mut shape, role_key, chars); + add(&mut shape, "text_chars", chars); + let max_chars = shape["max_message_chars"].as_u64().unwrap_or_default(); + shape["max_message_chars"] = json!(max_chars.max(chars as u64)); + add( + &mut shape, + "tool_calls", + message["tool_calls"].as_array().map_or(0, Vec::len), + ); + } + Some(shape) +} + impl LlmRequestEvent { fn from_trace(trace: &RuntimeLlmRequestData, raw_bodies: RawBodyPolicy) -> Self { let mut inference_params = trace @@ -1287,6 +1344,9 @@ impl LlmRequestEvent { let raw_request = capture_raw_body(trace.raw_request.as_ref(), raw_bodies); let raw_response = capture_raw_body(trace.raw_response.as_ref(), raw_bodies); if let Some(params) = inference_params.as_object_mut() { + if let Some(shape) = trace.raw_request.as_ref().and_then(request_shape) { + params.insert("request_shape".to_owned(), shape); + } params.insert( "raw_request_capture".to_owned(), Value::String(raw_request.status.to_owned()), @@ -2128,6 +2188,47 @@ mod tests { ); } + #[test] + fn oversized_request_keeps_content_free_shape_diagnostics() { + let trace = RuntimeLlmRequestData { + raw_request: Some(json!({ + "messages": [ + {"role": "system", "content": "secret".repeat(12_000)}, + {"role": "user", "content": [ + {"type": "text", "text": "привет"}, + {"type": "image_url", "image_url": {"url": "https://private.invalid/image"}} + ]}, + {"role": "assistant", "content": null, "tool_calls": [{"id": "private-call", "function": {"name": "secret_tool", "arguments": "private args"}}]} + ], + "tools": [{"type": "function", "function": {"name": "secret_tool"}}] + })), + ..RuntimeLlmRequestData::default() + }; + for enabled in [true, false] { + let event = LlmRequestEvent::from_trace( + &trace, + RawBodyPolicy { + enabled, + max_bytes: 65_536, + }, + ); + assert!(event.raw_request.is_none()); + let shape = &event.inference_params["request_shape"]; + assert_eq!(shape["messages"], 3); + assert_eq!(shape["system_chars"], 72_000); + assert_eq!(shape["user_chars"], 6); + assert_eq!(shape["text_chars"], 72_006); + assert_eq!(shape["max_message_chars"], 72_000); + assert_eq!(shape["image_parts"], 1); + assert_eq!(shape["tool_calls"], 1); + assert_eq!(shape["tool_definitions"], 1); + let diagnostic = shape.to_string(); + for private in ["secret", "private", "привет", "https://"] { + assert!(!diagnostic.contains(private), "{diagnostic}"); + } + } + } + #[test] fn raw_body_scrub_updates_only_aged_rows_with_bodies() { assert!(SQL_SCRUB_OLD_LLM_RAW_BODIES.contains("UPDATE llm_request_events")); diff --git a/crates/openplotva-app/src/settings.rs b/crates/openplotva-app/src/settings.rs index 0f69ea68..cbfdccce 100644 --- a/crates/openplotva-app/src/settings.rs +++ b/crates/openplotva-app/src/settings.rs @@ -712,6 +712,7 @@ impl NewMembersGreetingSender for TelegramJoinGreetingSender { let queue = DispatcherQueue::new(DispatcherConfig::default()); let chat = ChatRef { + direct_messages_topic_id: None, id: message.chat_id, is_forum: message.thread_id.is_some(), }; @@ -1512,6 +1513,10 @@ where } let chat = ChatRef { + direct_messages_topic_id: message + .direct_messages_topic + .as_ref() + .map(|topic| topic.topic_id), id: message.chat.get_id().into(), is_forum: false, }; @@ -2368,6 +2373,7 @@ where NextId: FnMut() -> String, { let chat = ChatRef { + direct_messages_topic_id: None, id: params.chat_id, is_forum: false, }; @@ -2412,6 +2418,7 @@ where NextId: FnMut() -> String, { let chat = ChatRef { + direct_messages_topic_id: None, id: params.chat_id, is_forum: false, }; @@ -3137,6 +3144,10 @@ pub fn admin_chat_member_upsert_from_telegram( fn message_chat_ref(message: &TelegramMessage) -> ChatRef { ChatRef { + direct_messages_topic_id: message + .direct_messages_topic + .as_ref() + .map(|topic| topic.topic_id), id: message.chat.get_id().into(), is_forum: message.message_thread_id.is_some(), } diff --git a/crates/openplotva-app/src/telegram_activity.rs b/crates/openplotva-app/src/telegram_activity.rs index f3eead91..cbd17de1 100644 --- a/crates/openplotva-app/src/telegram_activity.rs +++ b/crates/openplotva-app/src/telegram_activity.rs @@ -177,6 +177,7 @@ where let request = ChatActionRequest { chat: ChatRef { + direct_messages_topic_id: None, id: chat_id, is_forum: thread_id.is_some(), }, diff --git a/crates/openplotva-app/src/translate.rs b/crates/openplotva-app/src/translate.rs index 97e1b99a..77f160e2 100644 --- a/crates/openplotva-app/src/translate.rs +++ b/crates/openplotva-app/src/translate.rs @@ -202,6 +202,7 @@ impl TranslateEffects for TranslateDispatcherEffects { ) -> TranslateEffectFuture<'a, Self::Error> { Box::pin(async move { let chat = ChatRef { + direct_messages_topic_id: None, id: message.chat_id, is_forum: message.thread_id.is_some(), }; @@ -1314,6 +1315,7 @@ pub fn translate_result_plan_from_control_job( translated: String, ) -> TranslateResultPlan { let chat = ChatRef { + direct_messages_topic_id: None, id: params.chat_id, is_forum: params.thread_id.is_some(), }; @@ -2195,6 +2197,7 @@ mod tests { let effects = TranslateDispatcherEffects::new(Arc::clone(&queue)) .with_virtual_id_factory(Arc::new(|| "translate-vmsg-1".to_owned())); let chat = ChatRef { + direct_messages_topic_id: None, id: -10042, is_forum: true, }; diff --git a/crates/openplotva-app/src/virtual_messages.rs b/crates/openplotva-app/src/virtual_messages.rs index b910bbde..5375e110 100644 --- a/crates/openplotva-app/src/virtual_messages.rs +++ b/crates/openplotva-app/src/virtual_messages.rs @@ -669,11 +669,14 @@ where index + 1 == total_parts, )? }; - let mut dispatcher_message = - DispatcherMessage::new(fingerprint_text_message_part(chat.id, &part), &virtual_id) - .with_method(TelegramOutboundMethod::from(method)) - .with_bypass_chat_restrictions(req.bypass_chat_restrictions) - .with_protected(req.protected); + let mut dispatcher_message = DispatcherMessage::new( + fingerprint_text_message_part(chat.id, &part) + .with_direct_messages_topic(chat.direct_messages_topic_id), + &virtual_id, + ) + .with_method(TelegramOutboundMethod::from(method)) + .with_bypass_chat_restrictions(req.bypass_chat_restrictions) + .with_protected(req.protected); if let Some(debounce_key) = req.debounce_key { dispatcher_message = dispatcher_message.with_debounce_key(debounce_key); } @@ -706,11 +709,14 @@ where let chat = message_target_chat(req.message.chat.as_ref(), req.reply_to)?; let method = build_rich_message_method(req.message, chat, req.reply_to)?; let virtual_id = next_virtual_id(); - let mut dispatcher_message = - DispatcherMessage::new(fingerprint_rich_message(chat.id, &method.html), &virtual_id) - .with_method(TelegramOutboundMethod::from(method)) - .with_bypass_chat_restrictions(req.bypass_chat_restrictions) - .with_protected(req.protected); + let mut dispatcher_message = DispatcherMessage::new( + fingerprint_rich_message(chat.id, &method.html) + .with_direct_messages_topic(chat.direct_messages_topic_id), + &virtual_id, + ) + .with_method(TelegramOutboundMethod::from(method)) + .with_bypass_chat_restrictions(req.bypass_chat_restrictions) + .with_protected(req.protected); if let Some(debounce_key) = req.debounce_key { dispatcher_message = dispatcher_message.with_debounce_key(debounce_key); } @@ -1259,6 +1265,59 @@ mod tests { } } + #[tokio::test] + async fn direct_messages_topics_keep_identical_replies_separate_after_replay() + -> Result<(), Box> { + let queue = DispatcherQueue::new(DispatcherConfig { + dedupe_config: openplotva_telegram::DebouncerConfig { + enabled: true, + default_window: Duration::from_secs(30), + max_cache_size: 1000, + per_chat_settings: Default::default(), + }, + ..DispatcherConfig::default() + }); + for topic in [4_294_967_301_i64, 4_294_967_302_i64] { + let mut request = text_request("same reply"); + request + .chat + .as_mut() + .expect("chat") + .direct_messages_topic_id = Some(topic); + let result = queue_text_message_parts( + &queue, + QueueTextRequest { + message: &request, + reply_to: None, + immediate_first: false, + bypass_chat_restrictions: false, + ephemeral_delete_after: None, + protected: false, + debounce_key: None, + }, + || format!("topic-{topic}"), + ) + .await?; + assert_eq!(result.enqueued_count(), 1, "topic {topic}"); + } + let expected = queue.snapshot(); + let saved = persistent_queue_from_drain(queue.drain_for_shutdown(), 10)?; + let replay = openplotva_telegram::persistent_queue_replay_from_items(saved.items); + for (before, after) in expected.regular.iter().zip(&replay.items) { + assert_eq!(before.fingerprint_key, after.fingerprint.to_string()); + } + let restored = DispatcherQueue::new(DispatcherConfig::default()); + for item in replay.items { + restored.restore(item); + } + let actual = restored.snapshot(); + assert_eq!(actual.regular.len(), 2); + for (before, after) in expected.regular.iter().zip(&actual.regular) { + assert_eq!(before.fingerprint_key, after.fingerprint_key); + } + Ok(()) + } + #[tokio::test] async fn queue_text_message_parts_inserts_virtual_ids_before_dispatch_enqueue() -> Result<(), Box> { @@ -1364,6 +1423,7 @@ mod tests { let queue = DispatcherQueue::new(DispatcherConfig::default()); let request = openplotva_telegram::StickerMessageRequest { chat: Some(ChatRef { + direct_messages_topic_id: None, id: 42, is_forum: true, }), @@ -1415,6 +1475,7 @@ mod tests { let queue = DispatcherQueue::new(DispatcherConfig::default()); let request = PhotoMessageRequest { chat: ChatRef { + direct_messages_topic_id: None, id: 42, is_forum: true, }, @@ -1467,6 +1528,7 @@ mod tests { let queue = DispatcherQueue::new(DispatcherConfig::default()); let request = AudioMessageRequest { chat: ChatRef { + direct_messages_topic_id: None, id: 42, is_forum: true, }, @@ -1598,6 +1660,7 @@ mod tests { let queue = DispatcherQueue::new(DispatcherConfig::default()); let request = EditMediaMessageRequest { chat: ChatRef { + direct_messages_topic_id: None, id: 42, is_forum: true, }, @@ -1654,6 +1717,7 @@ mod tests { let queue = DispatcherQueue::new(DispatcherConfig::default()); let request = EditTextMessageRequest { chat: ChatRef { + direct_messages_topic_id: None, id: 42, is_forum: false, }, @@ -1924,6 +1988,7 @@ mod tests { let method = build_text_message_method( &text_request("hello"), ChatRef { + direct_messages_topic_id: None, id: 42, is_forum: false, }, @@ -1970,6 +2035,7 @@ mod tests { let queue = DispatcherQueue::new(DispatcherConfig::default()); let request = EditTextMessageRequest { chat: ChatRef { + direct_messages_topic_id: None, id: 42, is_forum: false, }, @@ -2032,6 +2098,7 @@ mod tests { fn text_request(text: &str) -> TextMessageRequest { TextMessageRequest { chat: Some(ChatRef { + direct_messages_topic_id: None, id: 42, is_forum: false, }), @@ -2047,6 +2114,7 @@ mod tests { fn media_group_request() -> MediaGroupMessageRequest { MediaGroupMessageRequest { chat: ChatRef { + direct_messages_topic_id: None, id: 42, is_forum: true, }, @@ -2079,6 +2147,7 @@ mod tests { let method = build_text_message_method( &text_request("hello"), ChatRef { + direct_messages_topic_id: None, id: 42, is_forum: false, }, @@ -2106,6 +2175,7 @@ mod tests { let method = build_text_message_method( &text_request("hello"), ChatRef { + direct_messages_topic_id: None, id: 42, is_forum: false, }, diff --git a/crates/openplotva-config/src/lib.rs b/crates/openplotva-config/src/lib.rs index 1d2fe0f7..c10d37f5 100644 --- a/crates/openplotva-config/src/lib.rs +++ b/crates/openplotva-config/src/lib.rs @@ -383,8 +383,6 @@ pub const DEFAULT_MEMORY_CONSOLIDATION_MODEL: &str = "Gemma 4 26B Heretic"; pub const DEFAULT_MEMORY_MIN_MESSAGES_PER_RUN: i32 = 20; -pub const DEFAULT_MEMORY_MAX_QUEUED_RUNS: i32 = 5_000; - /// Visible-card lists the memory pipeline keeps in RAM; `0` disables the cache. pub const DEFAULT_MEMORY_CARD_CACHE_ENTRIES: i32 = 4_096; @@ -1051,7 +1049,6 @@ pub struct MemoryConfig { pub worker_concurrency: i32, pub max_messages_per_run: i32, pub min_messages_per_run: i32, - pub max_queued_runs: i32, pub card_cache_entries: i32, pub max_input_tokens: i32, pub tokenizer_model: String, @@ -1579,8 +1576,6 @@ pub struct RawConfig { pub memory_max_messages_per_run: Option, /// `MEMORY_MIN_MESSAGES_PER_RUN`. pub memory_min_messages_per_run: Option, - /// `MEMORY_MAX_QUEUED_RUNS`. - pub memory_max_queued_runs: Option, /// `MEMORY_CARD_CACHE_ENTRIES`. pub memory_card_cache_entries: Option, /// `MEMORY_MAX_INPUT_TOKENS`. @@ -2907,14 +2902,6 @@ impl AppConfig { DEFAULT_MEMORY_MIN_MESSAGES_PER_RUN, )?, )?, - max_queued_runs: validate_positive_i32( - "MEMORY_MAX_QUEUED_RUNS", - parse_i32( - "MEMORY_MAX_QUEUED_RUNS", - raw.memory_max_queued_runs, - DEFAULT_MEMORY_MAX_QUEUED_RUNS, - )?, - )?, card_cache_entries: validate_non_negative_i32( "MEMORY_CARD_CACHE_ENTRIES", parse_i32( @@ -3407,7 +3394,6 @@ impl RawConfig { memory_worker_concurrency: env("MEMORY_WORKER_CONCURRENCY"), memory_max_messages_per_run: env("MEMORY_MAX_MESSAGES_PER_RUN"), memory_min_messages_per_run: env("MEMORY_MIN_MESSAGES_PER_RUN"), - memory_max_queued_runs: env("MEMORY_MAX_QUEUED_RUNS"), memory_card_cache_entries: env("MEMORY_CARD_CACHE_ENTRIES"), memory_max_input_tokens: env("MEMORY_MAX_INPUT_TOKENS"), memory_tokenizer_model: env("MEMORY_TOKENIZER_MODEL"), @@ -4222,10 +4208,6 @@ mod tests { config.memory.min_messages_per_run, super::DEFAULT_MEMORY_MIN_MESSAGES_PER_RUN ); - assert_eq!( - config.memory.max_queued_runs, - super::DEFAULT_MEMORY_MAX_QUEUED_RUNS - ); assert_eq!(config.memory.max_input_tokens, 10_000); assert_eq!(config.memory.tokenizer_model, "google/gemma-4-26B-A4B-it"); assert_eq!(config.memory.embedder_service_name, "embedder"); @@ -4934,7 +4916,6 @@ mod tests { memory_worker_concurrency: Some("2".to_owned()), memory_max_messages_per_run: Some("55".to_owned()), memory_min_messages_per_run: Some("20".to_owned()), - memory_max_queued_runs: Some("5000".to_owned()), memory_max_input_tokens: Some("6000".to_owned()), memory_tokenizer_model: Some("tokenizer".to_owned()), memory_tokenizer_file: Some("/tmp/tokenizer.json".to_owned()), @@ -4971,7 +4952,6 @@ mod tests { assert_eq!(config.memory.worker_concurrency, 2); assert_eq!(config.memory.max_messages_per_run, 55); assert_eq!(config.memory.min_messages_per_run, 20); - assert_eq!(config.memory.max_queued_runs, 5000); assert_eq!(config.memory.max_input_tokens, 6000); assert_eq!(config.memory.tokenizer_model, "tokenizer"); assert_eq!(config.memory.tokenizer_file, "/tmp/tokenizer.json"); @@ -5008,7 +4988,6 @@ mod tests { fn memory_policy_defaults_and_rejects_non_positive_values() -> Result<(), super::ConfigError> { let config = AppConfig::from_raw(RawConfig::default())?; assert_eq!(config.memory.min_messages_per_run, 20); - assert_eq!(config.memory.max_queued_runs, 5000); assert_eq!( config.memory.card_cache_entries, super::DEFAULT_MEMORY_CARD_CACHE_ENTRIES @@ -5032,19 +5011,6 @@ mod tests { }) )); - let error = AppConfig::from_raw(RawConfig { - memory_max_queued_runs: Some("-1".to_owned()), - ..RawConfig::default() - }) - .err(); - assert!(matches!( - error, - Some(super::ConfigError::NonPositiveInteger { - name: "MEMORY_MAX_QUEUED_RUNS", - value: -1, - }) - )); - Ok(()) } diff --git a/crates/openplotva-core/src/lib.rs b/crates/openplotva-core/src/lib.rs index de4bb10d..000656ba 100644 --- a/crates/openplotva-core/src/lib.rs +++ b/crates/openplotva-core/src/lib.rs @@ -238,6 +238,9 @@ pub fn filter_non_terminator_tool_calls(calls: &[ToolCall]) -> Vec { #[derive(Clone, Debug, Default, Deserialize, PartialEq, Serialize)] pub struct ChatMessageMeta { + /// Destination topic in a channel's direct messages chat (up to 52 bits). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub direct_messages_topic_id: Option, /// Server-granted permission for one spontaneous gift; never copied from a tool argument. #[serde(default, skip_serializing_if = "std::ops::Not::not")] pub agent_gift: bool, diff --git a/crates/openplotva-llm/src/aifarm.rs b/crates/openplotva-llm/src/aifarm.rs index 90c8ec32..7975a12b 100644 --- a/crates/openplotva-llm/src/aifarm.rs +++ b/crates/openplotva-llm/src/aifarm.rs @@ -166,6 +166,18 @@ pub struct ChatVideoUrlPart { pub url: String, } +/// Content-free sizes of the dialog inputs selected for one model call. +#[derive(Clone, Debug, PartialEq, Serialize)] +pub struct DialogContextShape { + pub history_messages: usize, + pub history_text_chars: usize, + pub current_text_chars: usize, + pub reference_blocks: usize, + pub reference_text_chars: usize, + pub shield_text_chars: usize, + pub transcript_messages: usize, +} + /// OpenAI-compatible chat completion request subset used by AIFarm routing. #[derive(Clone, Debug, Default, Deserialize, PartialEq, Serialize)] pub struct ChatCompletionRequest { @@ -260,6 +272,9 @@ pub struct ChatCompletionRequest { /// Trace metadata for low-level call observation. Never serialized to the wire. #[serde(skip)] pub trace: Option, + /// Local diagnostics, excluded from provider payloads. + #[serde(skip)] + pub dialog_context: Option, /// Tool calls written as markup in the text execute instead of rejecting the sample. /// Client-side policy, never serialized to the wire. #[serde(skip)] @@ -3291,6 +3306,22 @@ where let mut request = ChatCompletionRequest { model, messages, + dialog_context: Some(DialogContextShape { + history_messages: history.len(), + history_text_chars: history + .iter() + .map(|message| message.text.chars().count()) + .sum(), + current_text_chars: input.message.text.chars().count(), + reference_blocks: input.reference_context.len(), + reference_text_chars: input + .reference_context + .iter() + .map(|text| text.chars().count()) + .sum(), + shield_text_chars: input.shield_context.chars().count(), + transcript_messages: transcript.len(), + }), stream: false, max_tokens: self.cfg.max_tokens_for_input(input), temperature: self.cfg.temperature, @@ -3854,6 +3885,9 @@ fn aifarm_trace_transport(result: &CompletionResult) -> Option { fn aifarm_trace_inference_params(request: &ChatCompletionRequest) -> Option { let mut map = serde_json::Map::new(); + if let Some(shape) = &request.dialog_context { + map.insert("dialog_context".to_owned(), json!(shape)); + } insert_positive_i32(&mut map, "max_tokens", request.max_tokens); insert_opt_f64(&mut map, "temperature", request.temperature); insert_opt_f64(&mut map, "top_p", request.top_p); @@ -11999,6 +12033,51 @@ mod tests { assert!(error.to_string().contains("tool protocol error"), "{error}"); } + #[test] + fn dialog_context_diagnostics_count_selected_history_without_storing_content() + -> Result<(), Box> { + let provider = AifarmDialogProvider::with_transport( + AifarmDialogConfig::default(), + FakeTransport::new(Vec::new()), + ); + let mut input = base_input(); + input.message.text = "привет".to_owned(); + input.reference_context = vec!["private-reference".repeat(100)]; + input.shield_context = "shield".to_owned(); + input.history = vec![HistoryMessage::default(); 50]; + let history = vec![ + HistoryMessage { + text: "secret-history".to_owned(), + ..HistoryMessage::default() + }; + 3 + ]; + let request = provider.step_request_with_history( + &input, + &history, + &[], + &openplotva_dialog::ToolsMode::FinalOnly, + 1, + )?; + let params = super::aifarm_trace_inference_params(&request).expect("inference parameters"); + let shape = ¶ms["dialog_context"]; + assert_eq!(shape["history_messages"], 3); + assert_eq!(shape["history_text_chars"], 42); + assert_eq!(shape["current_text_chars"], 6); + assert_eq!(shape["reference_blocks"], 1); + assert_eq!(shape["reference_text_chars"], 1700); + assert_eq!(shape["shield_text_chars"], 6); + assert_eq!(shape["transcript_messages"], 0); + assert!(!params.to_string().contains("secret-history")); + assert!(!params.to_string().contains("private-reference")); + assert!( + serde_json::to_value(&request)? + .get("dialog_context") + .is_none() + ); + Ok(()) + } + #[test] fn dialog_request_prefers_input_enable_thinking_over_config() -> Result<(), AifarmDialogError> { let provider = AifarmDialogProvider::with_transport( diff --git a/crates/openplotva-memory/src/lib.rs b/crates/openplotva-memory/src/lib.rs index 60afa9db..5fde5395 100644 --- a/crates/openplotva-memory/src/lib.rs +++ b/crates/openplotva-memory/src/lib.rs @@ -19,7 +19,6 @@ pub const PURPOSE: &str = "memory"; pub const DEFAULT_MEMORY_MAX_OUTPUT_TOKENS: i32 = 4000; pub const DEFAULT_MEMORY_MIN_MESSAGES_PER_RUN: i32 = 20; -pub const DEFAULT_MEMORY_MAX_QUEUED_RUNS: i32 = 5_000; pub const DISCOVERY_PRIORITY_MEMORY: i32 = 10; pub const DEFAULT_MEMORY_CONSOLIDATION_MODEL: &str = "Gemma 4 26B Heretic"; pub const PROMPT_VERSION: &str = "chat_memory_daily_v7"; @@ -177,14 +176,12 @@ pub struct RetrievalScope { #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub struct MemoryRunEnqueuePolicy { pub min_messages_per_run: i32, - pub max_queued_runs: i32, } impl Default for MemoryRunEnqueuePolicy { fn default() -> Self { Self { min_messages_per_run: DEFAULT_MEMORY_MIN_MESSAGES_PER_RUN, - max_queued_runs: DEFAULT_MEMORY_MAX_QUEUED_RUNS, } } } @@ -199,11 +196,6 @@ impl MemoryRunEnqueuePolicy { } else { self.min_messages_per_run }, - max_queued_runs: if self.max_queued_runs <= 0 { - fallback.max_queued_runs - } else { - self.max_queued_runs - }, } } } diff --git a/crates/openplotva-storage/src/lib.rs b/crates/openplotva-storage/src/lib.rs index f389966c..a7d21174 100644 --- a/crates/openplotva-storage/src/lib.rs +++ b/crates/openplotva-storage/src/lib.rs @@ -557,14 +557,6 @@ fn memory_message_count_bucket(message_count: i32) -> &'static str { } } -fn memory_enqueue_remaining_capacity( - policy: openplotva_memory::MemoryRunEnqueuePolicy, - active_queue_depth: i64, -) -> Option { - let remaining = i64::from(policy.normalized().max_queued_runs) - active_queue_depth; - (remaining > 0).then(|| i32::try_from(remaining).unwrap_or(i32::MAX)) -} - fn optional_trimmed_string(value: &str) -> Option { let value = value.trim(); (!value.is_empty()).then(|| value.to_owned()) @@ -1018,22 +1010,45 @@ WHERE status = 'processing' AND (leased_until IS NULL OR leased_until < CURRENT_TIMESTAMP)"#; /// Memory runs whose window ended longer ago than this are abandoned: never -/// claimed and not counted against the queue cap. +/// claimed and not counted as active queue depth. pub const MEMORY_RUN_MAX_CLAIM_AGE: Duration = Duration::from_secs(7 * 24 * 60 * 60); -/// Claims the oldest claimable memory run first, so a backlog drains in age -/// order instead of starving behind each new daily batch. Runs whose window ended +/// Freshly enqueued windows and their continuations precede historical backfill. +/// Freshness is measured at initial enqueue, so waiting does not demote a run. +/// Each lane still drains oldest first. Runs whose window ended /// more than `$4` seconds ago are never claimed (and are not counted as queue -/// depth); cleanup removes them separately. The ordering is the exact backward -/// scan of the `idx_memory_runs_claim_*_order` indexes, so no extra index is -/// needed: within one window it prefers continuations, then newer ids. -pub const SQL_CLAIM_MEMORY_RUN: &str = r#"WITH current_processing AS ( +/// depth); cleanup removes them separately. Within a window, continuations +/// precede the initial run, with newer ids breaking ties. +pub const SQL_CLAIM_MEMORY_RUN: &str = r#"WITH fresh_candidate AS ( + SELECT r.id, -1 AS priority + FROM memory_runs r + WHERE r.prompt_version = $1 + AND r.range_end_at >= CURRENT_TIMESTAMP - make_interval(secs => $4::double precision) + AND r.attempts < 5 + AND ((r.status = 'processing' AND r.leased_until < CURRENT_TIMESTAMP) + OR (r.status IN ('queued', 'failed') AND (r.leased_until IS NULL OR r.leased_until < CURRENT_TIMESTAMP))) + AND EXISTS ( + SELECT 1 FROM memory_runs initial + WHERE initial.chat_id = r.chat_id AND initial.thread_id = r.thread_id + AND initial.range_start_at = r.range_start_at AND initial.range_end_at = r.range_end_at + AND initial.prompt_version = r.prompt_version + AND initial.cursor_after_at = 'epoch'::timestamptz + AND initial.cursor_after_message_id = 0 AND initial.cursor_after_entry_id = '' + AND initial.range_end_at >= initial.created_at - interval '2 hours' + ) + ORDER BY r.range_end_at ASC, r.range_start_at ASC, + r.cursor_after_at DESC, r.cursor_after_message_id DESC, r.id DESC + FOR UPDATE OF r SKIP LOCKED + LIMIT 1 +), +current_processing AS ( SELECT id, 0 AS priority FROM ( SELECT id FROM memory_runs WHERE prompt_version = $1 AND status = 'processing' + AND NOT EXISTS (SELECT 1 FROM fresh_candidate) AND leased_until < CURRENT_TIMESTAMP AND attempts < 5 AND range_end_at >= CURRENT_TIMESTAMP - make_interval(secs => $4::double precision) @@ -1053,6 +1068,7 @@ current_failed AS ( FROM memory_runs WHERE prompt_version = $1 AND status = 'failed' + AND NOT EXISTS (SELECT 1 FROM fresh_candidate) AND attempts < 5 AND (leased_until IS NULL OR leased_until < CURRENT_TIMESTAMP) AND NOT EXISTS (SELECT 1 FROM current_processing) @@ -1073,6 +1089,7 @@ current_queued AS ( FROM memory_runs WHERE prompt_version = $1 AND status = 'queued' + AND NOT EXISTS (SELECT 1 FROM fresh_candidate) AND (leased_until IS NULL OR leased_until < CURRENT_TIMESTAMP) AND NOT EXISTS (SELECT 1 FROM current_processing) AND NOT EXISTS (SELECT 1 FROM current_failed) @@ -1087,6 +1104,8 @@ current_queued AS ( ) current_queued ), current_candidate AS ( + SELECT id, priority FROM fresh_candidate + UNION ALL SELECT id, priority FROM current_processing UNION ALL SELECT id, priority FROM current_failed @@ -1216,41 +1235,6 @@ pub const SQL_RETRY_FAILED_MEMORY_RUNS: &str = "UPDATE memory_runs SET status = pub const SQL_COUNT_ACTIVE_MEMORY_RUNS: &str = "SELECT count(*)::bigint FROM memory_runs WHERE prompt_version = $1 AND status IN ('queued', 'processing') AND range_end_at >= CURRENT_TIMESTAMP - make_interval(secs => $2::double precision)"; -pub const SQL_RECORD_MEMORY_ENQUEUE_BACKPRESSURE: &str = r#"WITH dims AS ( - SELECT jsonb_build_object( - 'decision', 'backpressure', - 'scope', 'all', - 'message_count_bucket', 'all' - ) AS dimensions -) -INSERT INTO telemetry_rollups ( - source, kind, granularity, bucket_start, bucket_end, dimensions_hash, dimensions, metrics -) -SELECT - 'memory', - 'consolidation_enqueue', - 'day', - $2::timestamptz, - $3::timestamptz, - md5(dimensions::text), - dimensions, - jsonb_build_object( - 'window_count', 0, - 'message_count', 0, - 'min_range_start', $2::timestamptz, - 'max_range_end', $3::timestamptz, - 'prompt_version', $1::text, - 'queue_depth', $4::bigint, - 'min_messages_per_run', $5::int, - 'max_queued_runs', $6::int - ) -FROM dims -ON CONFLICT (source, kind, granularity, bucket_start, dimensions_hash) DO UPDATE SET - bucket_end = EXCLUDED.bucket_end, - dimensions = EXCLUDED.dimensions, - metrics = EXCLUDED.metrics, - updated_at = CURRENT_TIMESTAMP"#; - pub const SQL_RECORD_MEMORY_ENQUEUE_ROLLUPS: &str = r#"WITH grouped_windows AS ( SELECT h.chat_id, @@ -1270,7 +1254,7 @@ pub const SQL_RECORD_MEMORY_ENQUEUE_ROLLUPS: &str = r#"WITH grouped_windows AS ( AND h.kind = 'text' GROUP BY 1, 2, 3 ), -ranked_windows AS ( +checked_windows AS ( SELECT *, EXISTS ( @@ -1284,23 +1268,7 @@ ranked_windows AS ( AND r.cursor_after_at = '1970-01-01 00:00:00+00'::timestamptz AND r.cursor_after_message_id = 0 AND r.cursor_after_entry_id = '' - ) AS already_enqueued, - sum( - CASE WHEN message_count >= $5 AND NOT EXISTS ( - SELECT 1 - FROM memory_runs r - WHERE r.chat_id = grouped_windows.chat_id - AND r.thread_id = grouped_windows.canonical_thread_id - AND r.range_start_at = grouped_windows.range_start_at - AND r.range_end_at = grouped_windows.range_end_at - AND r.prompt_version = $1 - AND r.cursor_after_at = '1970-01-01 00:00:00+00'::timestamptz - AND r.cursor_after_message_id = 0 - AND r.cursor_after_entry_id = '' - ) THEN 1 ELSE 0 END - ) OVER ( - ORDER BY range_start_at DESC, message_count DESC, chat_id ASC, canonical_thread_id ASC - ) AS enqueue_rank + ) AS already_enqueued FROM grouped_windows ), classified AS ( @@ -1308,8 +1276,7 @@ classified AS ( CASE WHEN message_count < $5 THEN 'below_threshold' WHEN already_enqueued THEN 'already_enqueued' - WHEN enqueue_rank <= $8 THEN 'enqueued' - ELSE 'queue_cap' + ELSE 'enqueued' END AS decision, CASE WHEN canonical_thread_id = 0 THEN 'chat' ELSE 'thread' END AS scope, CASE @@ -1323,7 +1290,7 @@ classified AS ( range_start_at, range_end_at, message_count - FROM ranked_windows + FROM checked_windows ), prepared AS ( SELECT @@ -1340,8 +1307,8 @@ prepared AS ( 'prompt_version', $1::text, 'queue_depth', $6::bigint, 'min_messages_per_run', $5::int, - 'max_queued_runs', $7::int, - 'effective_limit', $8::int + 'max_queued_runs', NULL, + 'effective_limit', NULL ) AS metrics FROM classified GROUP BY decision, scope, message_count_bucket @@ -1405,7 +1372,6 @@ active_windows AS ( AND r.cursor_after_entry_id = '' ) ORDER BY range_start_at DESC, message_count DESC, chat_id ASC, canonical_thread_id ASC - LIMIT $6 ) INSERT INTO memory_runs ( chat_id, @@ -1417,7 +1383,6 @@ INSERT INTO memory_runs ( ) SELECT chat_id, thread_id, range_start_at, range_end_at, $1, message_count FROM active_windows -WHERE $6 > 0 ON CONFLICT (chat_id, thread_id, range_start_at, range_end_at, prompt_version, cursor_after_at, cursor_after_message_id, cursor_after_entry_id) DO NOTHING"#; pub const SQL_SKIP_SUPERSEDED_MEMORY_RUNS: &str = "UPDATE memory_runs SET status = 'skipped', lease_owner = '', leased_until = NULL, error = '', completed_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP WHERE prompt_version <> $1 AND (status IN ('queued', 'failed') OR (status = 'processing' AND leased_until < CURRENT_TIMESTAMP))"; @@ -5838,8 +5803,8 @@ impl PostgresMemoryStore { /// Enqueue a run for every chat hour that completed before `now`, started /// within the last `lookback` (at most `retention`), and has enough - /// messages: newest first, while the active queue has room. The enqueue - /// rollups still describe whole UTC days of the scanned range. + /// messages. Queue depth does not limit admission. The enqueue rollups + /// describe whole UTC days of the scanned range. pub async fn ensure_hourly_runs( &self, now: OffsetDateTime, @@ -5862,17 +5827,6 @@ impl PostgresMemoryStore { let mut window_end = completed_hours_end; let mut total = 0; let active_queue_depth = self.active_memory_run_count().await?; - if memory_enqueue_remaining_capacity(policy, active_queue_depth).is_none() { - self.record_memory_enqueue_backpressure( - earliest_day_start, - completed_hours_end, - active_queue_depth, - policy, - ) - .await?; - self.skip_superseded_runs().await?; - return Ok(total); - } while window_end > earliest_day_start { // One rollup bucket per UTC day: today's runs to the last completed // hour, then whole earlier days. @@ -5880,23 +5834,16 @@ impl PostgresMemoryStore { .date() .midnight() .assume_utc(); - let Some(limit) = memory_enqueue_remaining_capacity( - policy, - active_queue_depth.saturating_add(i64::try_from(total).unwrap_or(i64::MAX)), - ) else { - break; - }; self.record_memory_enqueue_rollups( window_start, window_end, earliest, active_queue_depth, policy, - limit, ) .await?; total += self - .ensure_run_window(window_start, window_end, scan_from, policy, limit) + .ensure_run_window(window_start, window_end, scan_from, policy) .await?; window_end = window_start; } @@ -5912,25 +5859,6 @@ impl PostgresMemoryStore { .await?) } - async fn record_memory_enqueue_backpressure( - &self, - window_start: OffsetDateTime, - window_end: OffsetDateTime, - active_queue_depth: i64, - policy: openplotva_memory::MemoryRunEnqueuePolicy, - ) -> Result<(), StorageError> { - sqlx::query(SQL_RECORD_MEMORY_ENQUEUE_BACKPRESSURE) - .bind(openplotva_memory::PROMPT_VERSION) - .bind(window_start) - .bind(window_end) - .bind(active_queue_depth) - .bind(policy.min_messages_per_run) - .bind(policy.max_queued_runs) - .execute(&self.pool) - .await?; - Ok(()) - } - async fn record_memory_enqueue_rollups( &self, window_start: OffsetDateTime, @@ -5938,7 +5866,6 @@ impl PostgresMemoryStore { earliest: OffsetDateTime, active_queue_depth: i64, policy: openplotva_memory::MemoryRunEnqueuePolicy, - limit: i32, ) -> Result<(), StorageError> { sqlx::query(SQL_RECORD_MEMORY_ENQUEUE_ROLLUPS) .bind(openplotva_memory::PROMPT_VERSION) @@ -5947,8 +5874,6 @@ impl PostgresMemoryStore { .bind(earliest) .bind(policy.min_messages_per_run) .bind(active_queue_depth) - .bind(policy.max_queued_runs) - .bind(limit) .execute(&self.pool) .await?; Ok(()) @@ -5960,7 +5885,6 @@ impl PostgresMemoryStore { window_end: OffsetDateTime, earliest: OffsetDateTime, policy: openplotva_memory::MemoryRunEnqueuePolicy, - limit: i32, ) -> Result { Ok(sqlx::query(SQL_ENSURE_HOURLY_MEMORY_RUNS) .bind(openplotva_memory::PROMPT_VERSION) @@ -5968,7 +5892,6 @@ impl PostgresMemoryStore { .bind(window_end) .bind(earliest) .bind(policy.min_messages_per_run) - .bind(limit) .execute(&self.pool) .await? .rows_affected()) @@ -11818,7 +11741,7 @@ mod tests { super::SQL_CLAIM_MEMORY_RUN .matches("range_end_at >= CURRENT_TIMESTAMP - make_interval(secs => $4") .count(), - 6, + 7, "every claim lane must skip abandoned runs" ); assert!(super::SQL_COMPLETE_MEMORY_RUN.contains("cards_superseded = $4")); @@ -11845,20 +11768,15 @@ mod tests { assert!( super::SQL_ENSURE_HOURLY_MEMORY_RUNS.contains("WHERE NOT EXISTS (\n SELECT 1") ); - assert!(super::SQL_ENSURE_HOURLY_MEMORY_RUNS.contains("LIMIT $6")); assert!( super::SQL_SELECT_MEMORY_RUN_MESSAGES .contains("WHEN COALESCE(c.is_forum, false) AND e.thread_id <> 0") ); assert!(super::SQL_COUNT_ACTIVE_MEMORY_RUNS.contains("status IN ('queued', 'processing')")); assert!(super::SQL_RECORD_MEMORY_ENQUEUE_ROLLUPS.contains("message_count_bucket")); - assert!(super::SQL_RECORD_MEMORY_ENQUEUE_ROLLUPS.contains("'queue_cap'")); assert!(!super::SQL_RECORD_MEMORY_ENQUEUE_ROLLUPS.contains("daily")); - assert!(!super::SQL_RECORD_MEMORY_ENQUEUE_BACKPRESSURE.contains("daily")); assert!(super::SQL_RECORD_MEMORY_ENQUEUE_ROLLUPS.contains("'below_threshold'")); assert!(super::SQL_RECORD_MEMORY_ENQUEUE_ROLLUPS.contains("'already_enqueued'")); - assert!(super::SQL_RECORD_MEMORY_ENQUEUE_BACKPRESSURE.contains("'backpressure'")); - assert!(super::SQL_RECORD_MEMORY_ENQUEUE_BACKPRESSURE.contains("'min_range_start'")); assert!(super::SQL_LIST_MEMORY_ENQUEUE_ROLLUPS.contains("kind = 'consolidation_enqueue'")); assert!(super::SQL_ENSURE_HOURLY_MEMORY_RUNS.contains("h.bucket_day >=")); assert!(super::SQL_SKIP_SUPERSEDED_MEMORY_RUNS.contains("prompt_version <> $1")); @@ -11980,31 +11898,13 @@ mod tests { } #[test] - fn memory_enqueue_policy_normalizes_capacity_and_buckets_messages() { + fn memory_enqueue_rollups_bucket_messages() { assert_eq!(super::memory_message_count_bucket(1), "1"); assert_eq!(super::memory_message_count_bucket(3), "2_3"); assert_eq!(super::memory_message_count_bucket(10), "4_10"); assert_eq!(super::memory_message_count_bucket(19), "11_19"); assert_eq!(super::memory_message_count_bucket(20), "20_50"); assert_eq!(super::memory_message_count_bucket(51), "51_plus"); - - let policy = openplotva_memory::MemoryRunEnqueuePolicy { - min_messages_per_run: 20, - max_queued_runs: 5000, - }; - assert_eq!( - super::memory_enqueue_remaining_capacity(policy, 4_900), - Some(100) - ); - assert_eq!( - super::memory_enqueue_remaining_capacity(policy, 0), - Some(5_000), - "only the active queue bounds the enqueue; there is no daily cap" - ); - assert_eq!( - super::memory_enqueue_remaining_capacity(policy, 5_000), - None - ); } #[test] @@ -12216,6 +12116,137 @@ mod tests { assert_eq!(r#ref.message_id, 20); } + #[tokio::test] + async fn live_memory_claim_prioritizes_fresh_work_even_after_it_ages() + -> Result<(), Box> { + let Ok(dsn) = env::var("OPENPLOTVA_TEST_POSTGRES_DSN") else { + return Ok(()); + }; + let pool = PgPoolOptions::new() + .max_connections(1) + .connect(&dsn) + .await?; + super::run_migrations_on(&pool).await?; + sqlx::query("CREATE TEMP TABLE memory_runs (LIKE public.memory_runs INCLUDING ALL)") + .execute(&pool) + .await?; + let version = openplotva_memory::PROMPT_VERSION; + // Backfill retries must not jump ahead of a fresh window. A continuation + // keeps its initial window's priority even if processing took hours. + sqlx::query("INSERT INTO memory_runs (id, chat_id, range_start_at, range_end_at, prompt_version, status, created_at, leased_until, cursor_after_at) VALUES + (1, 1, now()-interval '3 days', now()-interval '3 days'+interval '1 hour', $1, 'processing', now(), now()-interval '1 minute', 'epoch'), + (2, 2, now()-interval '2 days', now()-interval '2 days'+interval '1 hour', $1, 'failed', now(), NULL, 'epoch'), + (3, 3, now()-interval '5 hours', now()-interval '4 hours', $1, 'completed', now()-interval '4 hours', NULL, 'epoch'), + (4, 3, now()-interval '5 hours', now()-interval '4 hours', $1, 'queued', now(), NULL, now()-interval '4 hours 30 minutes'), + (5, 4, now()-interval '1 hour', now(), $1, 'queued', now(), NULL, 'epoch')") + .bind(version).execute(&pool).await?; + let store = super::PostgresMemoryStore::new(pool.clone()); + for expected in [4, 5, 1, 2] { + let run = store + .claim_run("freshness-test", Duration::from_secs(60)) + .await? + .expect("claim"); + assert_eq!(run.id, expected); + store.complete_run(run.id, Default::default()).await?; + } + assert!( + store + .claim_run("freshness-test", Duration::from_secs(60)) + .await? + .is_none() + ); + pool.close().await; + Ok(()) + } + + #[tokio::test] + async fn live_memory_enqueue_has_no_queue_cap_and_keeps_fresh_work_first() + -> Result<(), Box> { + let Ok(dsn) = env::var("OPENPLOTVA_TEST_POSTGRES_DSN") else { + return Ok(()); + }; + let pool = PgPoolOptions::new() + .max_connections(1) + .connect(&dsn) + .await?; + super::run_migrations_on(&pool).await?; + sqlx::raw_sql("CREATE TEMP TABLE memory_runs (LIKE public.memory_runs INCLUDING ALL); + CREATE TEMP TABLE chat_history_entries (LIKE public.chat_history_entries INCLUDING DEFAULTS); + CREATE TEMP TABLE telegram_chats_effective (id bigint, is_forum boolean); + CREATE TEMP TABLE telemetry_rollups (LIKE public.telemetry_rollups INCLUDING ALL);").execute(&pool).await?; + let now = time::OffsetDateTime::now_utc(); + sqlx::query("INSERT INTO chat_history_entries (bucket_day, chat_id, thread_id, message_id, entry_id, kind, role, occurred_at, sender_id, payload) + SELECT ($1::timestamptz-interval '2 days')::date, n, 0, m, n::text||':'||m::text, 'text', 'user', $1::timestamptz-interval '2 days', 1, '{}'::jsonb + FROM generate_series(1, 5001) n CROSS JOIN generate_series(1, 2) m") + .bind(now).execute(&pool).await?; + sqlx::query("INSERT INTO chat_history_entries (bucket_day, chat_id, thread_id, message_id, entry_id, kind, role, occurred_at, sender_id, payload) + SELECT occurred::date, chat_id, 0, m, chat_id::text||':'||m::text, 'text', 'user', occurred, 1, '{}'::jsonb + FROM (VALUES (9002, date_trunc('hour', $1::timestamptz)), (9003, $1::timestamptz-interval '8 days')) AS excluded(chat_id, occurred) + CROSS JOIN generate_series(1, 2) m") + .bind(now).execute(&pool).await?; + sqlx::query("INSERT INTO chat_history_entries (bucket_day, chat_id, thread_id, message_id, entry_id, kind, role, occurred_at, sender_id, payload) + VALUES (($1::timestamptz-interval '2 days')::date, 9004, 0, 1, 'below-threshold', 'text', 'user', $1::timestamptz-interval '2 days', 1, '{}'::jsonb)") + .bind(now).execute(&pool).await?; + let store = super::PostgresMemoryStore::new(pool.clone()); + let policy = openplotva_memory::MemoryRunEnqueuePolicy { + min_messages_per_run: 2, + }; + let retention = Duration::from_secs(7 * 24 * 60 * 60); + assert_eq!( + store + .ensure_hourly_runs(now, retention, retention, policy) + .await?, + 5001 + ); + let enqueued: i64 = sqlx::query_scalar("SELECT count(*) FROM memory_runs") + .fetch_one(&pool) + .await?; + assert_eq!( + enqueued, 5001, + "incomplete, expired, and below-threshold windows are excluded" + ); + let rollups: i64 = sqlx::query_scalar("SELECT COALESCE(sum((metrics->>'window_count')::bigint), 0)::bigint FROM telemetry_rollups + WHERE source='memory' AND dimensions->>'decision'='enqueued'") + .fetch_one(&pool).await?; + assert_eq!(rollups, 5001); + assert_eq!( + store + .ensure_hourly_runs(now, retention, retention, policy) + .await?, + 0, + "a repeated full scan must not duplicate pending windows" + ); + sqlx::query("INSERT INTO chat_history_entries (bucket_day, chat_id, thread_id, message_id, entry_id, kind, role, occurred_at, sender_id, payload) + SELECT ($1::timestamptz-interval '1 hour')::date, 9001, 0, m, 'fresh:'||m::text, 'text', 'user', $1::timestamptz-interval '1 hour', 1, '{}'::jsonb + FROM generate_series(1, 2) m").bind(now).execute(&pool).await?; + assert_eq!( + store + .ensure_hourly_runs(now, retention, retention, policy) + .await?, + 1 + ); + let run = store + .claim_run("freshness-test", Duration::from_secs(60)) + .await? + .expect("fresh claim"); + assert_eq!(run.chat_id, 9001); + store.complete_run(run.id, Default::default()).await?; + assert_eq!( + store + .ensure_hourly_runs(now, retention, retention, policy) + .await?, + 0, + "completed windows must not be reprocessed on the next scan" + ); + let backfill = store + .claim_run("freshness-test", Duration::from_secs(60)) + .await? + .expect("backfill claim"); + assert!((1..=5001).contains(&backfill.chat_id)); + pool.close().await; + Ok(()) + } + #[tokio::test] async fn live_memory_runs_are_enqueued_through_the_last_complete_hour() -> Result<(), Box> { @@ -12267,7 +12298,6 @@ mod tests { let policy = openplotva_memory::MemoryRunEnqueuePolicy { min_messages_per_run: 3, - max_queued_runs: 1_000_000, }; let retention = Duration::from_secs(7 * 24 * 60 * 60); let windows = || async { @@ -12289,7 +12319,7 @@ mod tests { assert_eq!( windows().await?, vec![(this_morning, this_morning + hour)], - "an hourly tick looks back only over the recent hours" + "an explicit narrow scan only includes recent hours" ); let full = store diff --git a/crates/openplotva-telegram/src/outbound.rs b/crates/openplotva-telegram/src/outbound.rs index 6237f45b..1b9cfe2a 100644 --- a/crates/openplotva-telegram/src/outbound.rs +++ b/crates/openplotva-telegram/src/outbound.rs @@ -78,6 +78,15 @@ pub struct MessageFingerprint { pub debounce_key: Option, } +impl MessageFingerprint { + pub fn with_direct_messages_topic(mut self, topic_id: Option) -> Self { + if let Some(topic_id) = topic_id.filter(|id| *id != 0) { + self.content_hash = hash_content(&format!("dm:{topic_id}:{}", self.content_hash)); + } + self + } +} + impl fmt::Display for MessageFingerprint { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { write!( @@ -95,6 +104,8 @@ impl fmt::Display for MessageFingerprint { /// Minimal chat fields needed by the outbound builder. #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub struct ChatRef { + /// Destination in a channel's direct messages chat, independent of forum threads. + pub direct_messages_topic_id: Option, /// Telegram chat ID. pub id: i64, /// Whether the chat is a forum supergroup. @@ -309,6 +320,7 @@ pub struct ReplyParametersPlan { /// Public sticker payload mirror for asserting form-only `carapax` methods. #[derive(Clone, Debug, Eq, PartialEq)] pub struct StickerMessagePlan { + pub direct_messages_topic_id: Option, /// Telegram target chat ID. pub chat_id: i64, /// Telegram sticker file ID. @@ -355,6 +367,7 @@ pub struct PhotoMessageRequest { /// Public photo payload mirror for asserting form-only `carapax` methods. #[derive(Clone, Debug, Eq, PartialEq)] pub struct PhotoMessagePlan { + pub direct_messages_topic_id: Option, /// Telegram target chat ID. pub chat_id: i64, /// Photo file, URL, or upload bytes. @@ -404,6 +417,7 @@ pub struct AudioMessageRequest { /// Public audio payload mirror for asserting form-only `carapax` methods. #[derive(Clone, Debug, Eq, PartialEq)] pub struct AudioMessagePlan { + pub direct_messages_topic_id: Option, /// Telegram target chat ID. pub chat_id: i64, /// Audio file, URL, or upload bytes. @@ -445,6 +459,7 @@ pub struct MediaGroupMessageRequest { /// Public media-group payload mirror for asserting form-only `carapax` methods. #[derive(Clone, Debug, Eq, PartialEq)] pub struct MediaGroupMessagePlan { + pub direct_messages_topic_id: Option, /// Telegram target chat ID. pub chat_id: i64, pub message_thread_id: Option, @@ -596,6 +611,9 @@ pub fn build_text_message_method( text = text.with_format(parse_mode); } let mut method = SendMessage::new(chat.id, text); + if let Some(topic_id) = chat.direct_messages_topic_id { + method = method.with_direct_messages_topic_id(topic_id); + } if req.disable_notification { method = method.with_disable_notification(true); } @@ -653,6 +671,7 @@ pub fn build_rich_message_method( .transpose() .map_err(|error| OutboundBuildError::PersistencePayload(error.to_string()))?; let options = RichSendOptions { + direct_messages_topic_id: chat.direct_messages_topic_id, message_thread_id: reply_to .and_then(reply_thread_id) .filter(|thread_id| *thread_id != 0) @@ -1146,6 +1165,7 @@ pub fn build_sticker_message_plan( }; Ok(StickerMessagePlan { + direct_messages_topic_id: chat.direct_messages_topic_id, chat_id: chat.id, file_id: req.file_id.clone(), message_thread_id, @@ -1165,6 +1185,7 @@ pub fn build_photo_message_plan( req: &PhotoMessageRequest, ) -> Result { Ok(PhotoMessagePlan { + direct_messages_topic_id: req.chat.direct_messages_topic_id, chat_id: req.chat.id, photo: req.photo.clone(), message_thread_id: (req.message_thread_id != 0).then_some(req.message_thread_id), @@ -1185,6 +1206,7 @@ pub fn build_audio_message_method( pub fn build_audio_message_plan(req: &AudioMessageRequest) -> AudioMessagePlan { AudioMessagePlan { + direct_messages_topic_id: req.chat.direct_messages_topic_id, chat_id: req.chat.id, audio: req.audio.clone(), message_thread_id: (req.message_thread_id != 0).then_some(req.message_thread_id), @@ -1204,6 +1226,7 @@ pub fn build_media_group_message_method( pub fn build_media_group_message_plan(req: &MediaGroupMessageRequest) -> MediaGroupMessagePlan { MediaGroupMessagePlan { + direct_messages_topic_id: req.chat.direct_messages_topic_id, chat_id: req.chat.id, message_thread_id: (req.message_thread_id != 0).then_some(req.message_thread_id), disable_notification: req.disable_notification, @@ -1282,6 +1305,7 @@ pub fn fingerprint_sticker_message_plan(plan: &StickerMessagePlan) -> MessageFin MESSAGE_TYPE_STICKER, hash_content(&plan.file_id), ) + .with_direct_messages_topic(plan.direct_messages_topic_id) } pub fn fingerprint_photo_message_plan(plan: &PhotoMessagePlan) -> MessageFingerprint { @@ -1291,6 +1315,7 @@ pub fn fingerprint_photo_message_plan(plan: &PhotoMessagePlan) -> MessageFingerp MESSAGE_TYPE_PHOTO, hash_content(content.as_ref()), ) + .with_direct_messages_topic(plan.direct_messages_topic_id) } pub fn fingerprint_audio_message_plan(plan: &AudioMessagePlan) -> MessageFingerprint { @@ -1300,6 +1325,7 @@ pub fn fingerprint_audio_message_plan(plan: &AudioMessagePlan) -> MessageFingerp MESSAGE_TYPE_AUDIO, hash_content(content.as_ref()), ) + .with_direct_messages_topic(plan.direct_messages_topic_id) } pub fn validate_text_message_text(text: &str, render_as: &str) -> Result<(), OutboundBuildError> { @@ -1369,6 +1395,7 @@ impl StickerMessagePlan { let mut value = go_base_file_config_json( self.chat_id, self.message_thread_id, + self.direct_messages_topic_id, self.disable_notification, self.reply_parameters, json!(self.file_id), @@ -1385,6 +1412,9 @@ impl StickerMessagePlan { if let Some(thread_id) = self.message_thread_id { method = method.with_message_thread_id(thread_id); } + if let Some(topic_id) = self.direct_messages_topic_id { + method = method.with_direct_messages_topic_id(topic_id); + } if self.disable_notification { method = method.with_disable_notification(true); } @@ -1412,6 +1442,7 @@ impl PhotoMessagePlan { let mut value = go_base_file_config_json( self.chat_id, self.message_thread_id, + self.direct_messages_topic_id, self.disable_notification, self.reply_parameters, self.photo.go_file_value(), @@ -1433,6 +1464,9 @@ impl PhotoMessagePlan { if let Some(thread_id) = self.message_thread_id { method = method.with_message_thread_id(thread_id); } + if let Some(topic_id) = self.direct_messages_topic_id { + method = method.with_direct_messages_topic_id(topic_id); + } if self.disable_notification { method = method.with_disable_notification(true); } @@ -1489,6 +1523,7 @@ impl AudioMessagePlan { let mut value = go_base_file_config_json( self.chat_id, self.message_thread_id, + self.direct_messages_topic_id, self.disable_notification, self.reply_parameters, self.audio.go_file_value(), @@ -1511,6 +1546,9 @@ impl AudioMessagePlan { if let Some(thread_id) = self.message_thread_id { method = method.with_message_thread_id(thread_id); } + if let Some(topic_id) = self.direct_messages_topic_id { + method = method.with_direct_messages_topic_id(topic_id); + } if self.disable_notification { method = method.with_disable_notification(true); } @@ -1564,6 +1602,7 @@ impl MediaGroupMessagePlan { let mut fields = go_base_chat_config_fields( self.chat_id, self.message_thread_id, + self.direct_messages_topic_id, self.disable_notification, self.reply_parameters, ); @@ -1591,6 +1630,9 @@ impl MediaGroupMessagePlan { if let Some(thread_id) = self.message_thread_id { method = method.with_message_thread_id(thread_id); } + if let Some(topic_id) = self.direct_messages_topic_id { + method = method.with_direct_messages_topic_id(topic_id); + } if self.disable_notification { method = method.with_disable_notification(true); } @@ -1681,6 +1723,7 @@ fn go_persistence_payload( fn go_base_file_config_json( chat_id: i64, message_thread_id: Option, + direct_messages_topic_id: Option, disable_notification: bool, reply_parameters: Option, file: Value, @@ -1688,6 +1731,7 @@ fn go_base_file_config_json( let mut fields = go_base_chat_config_fields( chat_id, message_thread_id, + direct_messages_topic_id, disable_notification, reply_parameters, ); @@ -1698,6 +1742,7 @@ fn go_base_file_config_json( fn go_base_chat_config_fields( chat_id: i64, message_thread_id: Option, + direct_messages_topic_id: Option, disable_notification: bool, reply_parameters: Option, ) -> Map { @@ -1710,7 +1755,10 @@ fn go_base_chat_config_fields( "MessageThreadID".to_owned(), json!(message_thread_id.unwrap_or_default()), ); - fields.insert("DirectMessagesTopicID".to_owned(), json!(0)); + fields.insert( + "DirectMessagesTopicID".to_owned(), + json!(direct_messages_topic_id.unwrap_or_default()), + ); fields.insert("ProtectContent".to_owned(), json!(false)); fields.insert("ReplyMarkup".to_owned(), Value::Null); fields.insert( @@ -1859,13 +1907,18 @@ mod tests { fn private_chat(id: i64) -> ChatRef { ChatRef { + direct_messages_topic_id: None, id, is_forum: false, } } fn forum_chat(id: i64) -> ChatRef { - ChatRef { id, is_forum: true } + ChatRef { + direct_messages_topic_id: None, + id, + is_forum: true, + } } fn base_text_request(text: &str) -> TextMessageRequest { @@ -1963,6 +2016,7 @@ mod tests { #[test] fn fingerprint_photo_plan_hashes_reusable_file_id_without_formatting() { let plan = PhotoMessagePlan { + direct_messages_topic_id: None, chat_id: 42, photo: PhotoSource::FileId("photo-file-id".to_owned()), message_thread_id: None, @@ -1984,6 +2038,7 @@ mod tests { #[test] fn fingerprint_photo_plan_hashes_uploaded_file_like_go_file_bytes() { let plan = PhotoMessagePlan { + direct_messages_topic_id: None, chat_id: 42, photo: PhotoSource::Bytes { file_name: "plotva_image_provider_1.jpg".to_owned(), @@ -2006,6 +2061,7 @@ mod tests { #[test] fn fingerprint_media_plans_use_go_message_type_names() { let sticker = StickerMessagePlan { + direct_messages_topic_id: None, chat_id: 42, file_id: "sticker-file".to_owned(), message_thread_id: None, @@ -2013,6 +2069,7 @@ mod tests { reply_parameters: None, }; let audio = AudioMessagePlan { + direct_messages_topic_id: None, chat_id: 7, audio: AudioSource::FileId("song-file-id".to_owned()), message_thread_id: None, @@ -2351,6 +2408,7 @@ mod tests { fn sticker_plan_persistence_payload_matches_go_sticker_config() -> Result<(), Box> { let plan = StickerMessagePlan { + direct_messages_topic_id: None, chat_id: 42, file_id: "sticker-file".to_owned(), message_thread_id: Some(77), @@ -2466,6 +2524,7 @@ mod tests { fn photo_plan_persistence_payload_matches_go_photo_config() -> Result<(), Box> { let plan = PhotoMessagePlan { + direct_messages_topic_id: None, chat_id: 42, photo: PhotoSource::Url("https://example.test/image.png".to_owned()), message_thread_id: Some(77), @@ -2503,6 +2562,7 @@ mod tests { fn photo_plan_persistence_payload_encodes_uploaded_bytes_like_go_file_bytes() -> Result<(), Box> { let plan = PhotoMessagePlan { + direct_messages_topic_id: None, chat_id: 42, photo: PhotoSource::Bytes { file_name: "image.png".to_owned(), @@ -2592,6 +2652,7 @@ mod tests { fn media_group_plan_persistence_payload_matches_go_media_group_config() -> Result<(), Box> { let plan = MediaGroupMessagePlan { + direct_messages_topic_id: None, chat_id: 42, message_thread_id: Some(77), disable_notification: true, @@ -2699,6 +2760,7 @@ mod tests { fn audio_plan_persistence_payload_matches_go_audio_config() -> Result<(), Box> { let plan = AudioMessagePlan { + direct_messages_topic_id: None, chat_id: 42, audio: AudioSource::FileId("song-file-id".to_owned()), message_thread_id: Some(77), @@ -3596,4 +3658,25 @@ mod tests { "clear must omit the reaction field, got {value}" ); } + #[test] + fn direct_messages_topic_survives_photo_plan_persistence() { + let request = PhotoMessageRequest { + chat: ChatRef { + direct_messages_topic_id: Some(4_294_967_301), + id: -100, + is_forum: false, + }, + message_thread_id: 0, + disable_notification: false, + photo: PhotoSource::FileId("photo-file".to_owned()), + caption: String::new(), + render_as: String::new(), + has_spoiler: false, + reply_parameters: None, + }; + let plan = build_photo_message_plan(&request).expect("plan"); + let saved = plan.to_persistence_payload().expect("save"); + let value: Value = serde_json::from_slice(&saved.message).expect("json"); + assert_eq!(value["DirectMessagesTopicID"], 4_294_967_301_i64); + } } diff --git a/crates/openplotva-telegram/src/persistence.rs b/crates/openplotva-telegram/src/persistence.rs index fa79e930..40bea932 100644 --- a/crates/openplotva-telegram/src/persistence.rs +++ b/crates/openplotva-telegram/src/persistence.rs @@ -650,6 +650,12 @@ fn replay_text_method(value: &Value) -> Option { text = text.with_format(mode); } let mut method = SendMessage::new(chat_id, text); + if let Some(topic_id) = non_zero_field_i64( + value, + &["DirectMessagesTopicID", "direct_messages_topic_id"], + ) { + method = method.with_direct_messages_topic_id(topic_id); + } if field_bool(value, &["DisableNotification", "disable_notification"]).unwrap_or(false) { method = method.with_disable_notification(true); } @@ -748,6 +754,10 @@ fn reaction_emoji(value: &Value) -> Option { fn replay_sticker_method(value: &Value) -> Option { let chat_id = field_i64(value, &["ChatID", "chat_id"])?; let plan = StickerMessagePlan { + direct_messages_topic_id: non_zero_field_i64( + value, + &["DirectMessagesTopicID", "direct_messages_topic_id"], + ), chat_id, file_id: string_file_value(field_value(value, &["File", "sticker"])?)?, message_thread_id: non_zero_field_i64(value, &["MessageThreadID", "message_thread_id"]), @@ -761,6 +771,10 @@ fn replay_sticker_method(value: &Value) -> Option { fn replay_photo_method(value: &Value) -> Option { let chat_id = field_i64(value, &["ChatID", "chat_id"])?; let plan = PhotoMessagePlan { + direct_messages_topic_id: non_zero_field_i64( + value, + &["DirectMessagesTopicID", "direct_messages_topic_id"], + ), chat_id, photo: photo_source_value(field_value(value, &["File", "photo"])?)?, message_thread_id: non_zero_field_i64(value, &["MessageThreadID", "message_thread_id"]), @@ -777,6 +791,10 @@ fn replay_photo_method(value: &Value) -> Option { fn replay_audio_method(value: &Value) -> Option { let chat_id = field_i64(value, &["ChatID", "chat_id"])?; let plan = AudioMessagePlan { + direct_messages_topic_id: non_zero_field_i64( + value, + &["DirectMessagesTopicID", "direct_messages_topic_id"], + ), chat_id, audio: audio_source_value(field_value(value, &["File", "audio"])?)?, message_thread_id: non_zero_field_i64(value, &["MessageThreadID", "message_thread_id"]), @@ -793,6 +811,10 @@ fn replay_media_group_method(value: &Value) -> Option { let chat_id = field_i64(value, &["ChatID", "chat_id"])?; let media = field_value(value, &["Media", "media"])?.as_array()?; let plan = MediaGroupMessagePlan { + direct_messages_topic_id: non_zero_field_i64( + value, + &["DirectMessagesTopicID", "direct_messages_topic_id"], + ), chat_id, message_thread_id: non_zero_field_i64(value, &["MessageThreadID", "message_thread_id"]), disable_notification: field_bool(value, &["DisableNotification", "disable_notification"]) @@ -835,7 +857,7 @@ fn fingerprint_from_value( fallback_chat_id: i64, ) -> MessageFingerprint { let chat_id = field_i64(value, &["ChatID", "chat_id"]).unwrap_or(fallback_chat_id); - match method_kind { + let fingerprint = match method_kind { TelegramOutboundMethodKind::SendMessage => MessageFingerprint { chat_id, message_type: "text".to_owned(), @@ -890,7 +912,13 @@ fn fingerprint_from_value( content_hash: hash_content(&value.to_string()), debounce_key: None, }, - } + }; + let topic_id = non_zero_field_i64( + value, + &["DirectMessagesTopicID", "direct_messages_topic_id"], + ) + .or_else(|| non_zero_field_i64(&value["options"], &["direct_messages_topic_id"])); + fingerprint.with_direct_messages_topic(topic_id) } fn file_fingerprint(chat_id: i64, message_type: &str, value: Option<&Value>) -> MessageFingerprint { @@ -1250,6 +1278,7 @@ mod tests { chat_id, html: html.to_owned(), options: RichSendOptions { + direct_messages_topic_id: None, message_thread_id: Some(77), reply_to_message_id: Some(9), allow_sending_without_reply: true, @@ -1680,14 +1709,65 @@ mod tests { )); } + #[tokio::test] + async fn direct_messages_topic_survives_media_replay_on_the_wire() + -> Result<(), Box> { + let topic = 4_294_967_301_i64; + let fixtures = [ + ( + TelegramOutboundMethodKind::SendPhoto, + json!({"File":"photo"}), + ), + ( + TelegramOutboundMethodKind::SendAudio, + json!({"File":"audio"}), + ), + ( + TelegramOutboundMethodKind::SendSticker, + json!({"File":"sticker"}), + ), + ( + TelegramOutboundMethodKind::SendMediaGroup, + json!({"Media":[{"type":"photo","media":"photo1"},{"type":"photo","media":"photo2"}]}), + ), + ]; + for (kind, mut payload) in fixtures { + payload["ChatID"] = json!(42); + payload["DirectMessagesTopicID"] = json!(topic); + payload["MessageThreadID"] = json!(0); + let method = + replay_outbound_method(kind, &serde_json::to_vec(&payload)?).expect("media replay"); + let wire = match method { + TelegramOutboundMethod::SendPhoto(method) => { + crate::test_api::bot_api_payload(*method).await? + } + TelegramOutboundMethod::SendAudio(method) => { + crate::test_api::bot_api_payload(*method).await? + } + TelegramOutboundMethod::SendSticker(method) => { + crate::test_api::bot_api_payload(*method).await? + } + TelegramOutboundMethod::SendMediaGroup(method) => { + crate::test_api::bot_api_payload(*method).await? + } + _ => panic!("unexpected method"), + }; + assert_eq!(wire["direct_messages_topic_id"], topic, "{kind:?}"); + assert!(wire.get("message_thread_id").is_none(), "{kind:?}"); + } + Ok(()) + } + #[test] fn reply_missing_fallback_replays_create_methods_without_reply_target() -> Result<(), Box> { - let text = carapax::types::SendMessage::new(42, "hello").with_reply_parameters( - carapax::types::ReplyParameters::new(9) - .with_chat_id(42) - .with_allow_sending_without_reply(true), - ); + let text = carapax::types::SendMessage::new(42, "hello") + .with_direct_messages_topic_id(4_294_967_301_i64) + .with_reply_parameters( + carapax::types::ReplyParameters::new(9) + .with_chat_id(42) + .with_allow_sending_without_reply(true), + ); let TelegramOutboundMethod::SendMessage(text) = replay_outbound_method_without_reply(&TelegramOutboundMethod::from(text)) .expect("text fallback") @@ -1696,6 +1776,7 @@ mod tests { }; let text_payload = serde_json::to_value(text.as_ref())?; assert!(text_payload.get("reply_parameters").is_none()); + assert_eq!(text_payload["direct_messages_topic_id"], 4_294_967_301_i64); let TelegramOutboundMethod::SendRichMessage(rich) = replay_outbound_method_without_reply(&rich_method(42, "hello")) @@ -1718,6 +1799,7 @@ mod tests { .with_method(sticker_method(42, "sticker-file-id")) .with_persistence_payload( StickerMessagePlan { + direct_messages_topic_id: None, chat_id: 42, file_id: "sticker-file-id".to_owned(), message_thread_id: None, @@ -1758,6 +1840,7 @@ mod tests { -> Result<(), Box> { let queue = DispatcherQueue::new(DispatcherConfig::default()); let command = StickerMessagePlan { + direct_messages_topic_id: None, chat_id: 42, file_id: "sticker-file-id".to_owned(), message_thread_id: None, @@ -2053,6 +2136,7 @@ mod tests { fn replay_reconstructs_current_form_backed_media_methods() -> Result<(), Box> { let photo = PhotoMessagePlan { + direct_messages_topic_id: None, chat_id: 42, photo: PhotoSource::Url("https://example.test/image.png".to_owned()), message_thread_id: Some(77), @@ -2064,6 +2148,7 @@ mod tests { } .to_persistence_payload()?; let audio = AudioMessagePlan { + direct_messages_topic_id: None, chat_id: 42, audio: AudioSource::Bytes { file_name: "song.mp3".to_owned(), @@ -2077,6 +2162,7 @@ mod tests { } .to_persistence_payload()?; let media_group = MediaGroupMessagePlan { + direct_messages_topic_id: None, chat_id: 42, message_thread_id: None, disable_notification: false, diff --git a/crates/openplotva-telegram/src/rich_api.rs b/crates/openplotva-telegram/src/rich_api.rs index 413e5c08..e8ae02dc 100644 --- a/crates/openplotva-telegram/src/rich_api.rs +++ b/crates/openplotva-telegram/src/rich_api.rs @@ -37,6 +37,8 @@ impl fmt::Debug for RichApiClient { /// Optional parameters shared by rich-message sends. #[derive(Debug, Default, Clone, Deserialize, PartialEq, Serialize)] pub struct RichSendOptions { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub direct_messages_topic_id: Option, pub message_thread_id: Option, pub reply_to_message_id: Option, pub allow_sending_without_reply: bool, @@ -220,6 +222,9 @@ fn build_send_body(chat_id: i64, html: &str, options: &RichSendOptions) -> Value let mut body = Map::new(); body.insert("chat_id".to_owned(), json!(chat_id)); body.insert("rich_message".to_owned(), json!({ "html": html })); + if let Some(topic) = options.direct_messages_topic_id { + body.insert("direct_messages_topic_id".to_owned(), json!(topic)); + } if let Some(thread) = options.message_thread_id { body.insert("message_thread_id".to_owned(), json!(thread)); } @@ -307,6 +312,17 @@ fn parse_api_response(text: &str) -> Resultx", &RichSendOptions::default()); @@ -319,6 +335,7 @@ mod tests { #[test] fn send_body_with_reply_thread_and_markup() { let options = RichSendOptions { + direct_messages_topic_id: None, message_thread_id: Some(7), reply_to_message_id: Some(11), allow_sending_without_reply: true, diff --git a/crates/openplotva-telegram/src/transport.rs b/crates/openplotva-telegram/src/transport.rs index e348fecf..efc13106 100644 --- a/crates/openplotva-telegram/src/transport.rs +++ b/crates/openplotva-telegram/src/transport.rs @@ -943,6 +943,7 @@ mod tests { fn chat(id: i64) -> ChatRef { ChatRef { + direct_messages_topic_id: None, id, is_forum: false, } diff --git a/crates/openplotva-updates/src/lib.rs b/crates/openplotva-updates/src/lib.rs index 1e15d3f2..134e9c4b 100644 --- a/crates/openplotva-updates/src/lib.rs +++ b/crates/openplotva-updates/src/lib.rs @@ -1574,6 +1574,9 @@ pub fn build_message_meta( vision_description: &str, ) -> ChatMessageMeta { let mut meta = ChatMessageMeta { + direct_messages_topic_id: message + .and_then(|m| m.direct_messages_topic.as_ref()) + .map(|topic| topic.topic_id), annotation: message.map(forward_annotation).unwrap_or_default(), vision_description: vision_description.trim().to_owned(), sender_type: sender.sender_type.clone(), @@ -5695,6 +5698,30 @@ mod tests { Ok(()) } + #[test] + fn message_meta_preserves_direct_messages_topic_as_i64() -> Result<(), Box> { + let update = serde_json::from_value(json!({ + "update_id": 140, + "message": { + "message_id": 72, "date": 1_710_000_000, + "chat": sample_private_chat_json(), "from": sample_user_json(), + "text": "hello", "direct_messages_topic": {"topic_id": 4_294_967_301_i64} + } + }))?; + let message = update_message(&update)?; + let meta = super::build_message_meta( + Some(message), + super::resolve_message_sender(Some(message)), + &[], + "", + ); + assert_eq!( + serde_json::to_value(meta)?["direct_messages_topic_id"], + 4_294_967_301_i64 + ); + Ok(()) + } + #[test] fn build_message_meta_matches_go_fetcher_shape() -> Result<(), Box> { let update = serde_json::from_value(json!({