Fix memory scheduling, Telegram topics, and context diagnostics

.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

crates/openplotva-app/src/admin.rs

@@ -2456,6 +2456,10 @@ fn admin_text_plan(
     ephemeral_delete_after: Option<Duration>,
 ) -> 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(),
     };

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(),
             };

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(),
     }

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,
                 },

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(),
         },

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(),
     }

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,
         };

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<dyn Error>>
+{
+    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, &params, "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<dyn Error>> {

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,
         },

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(),
             };

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(),
         };

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,
             };

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()

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(),
     }

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,
         },

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());