refactor: untangle runtime ownership and durable delivery

.env.example

@@ -300,7 +300,6 @@
 # Bootstrap/default client settings for the managed memory_extraction and
 # memory_subject_merge workflows; their model order and retry budgets are edited
 # independently in the routing admin after the first seed.
-# MEMORY_CONSOLIDATION_PROVIDER=aifarm
 # MEMORY_CONSOLIDATION_MODEL=qwen3.8-27b
 # MEMORY_CONSOLIDATION_TIMEOUT_SECONDS=600
 # Startup and each HH:02 UTC tick enqueue all missing completed chat hours

crates/openplotva-app/src/asr.rs

@@ -619,41 +619,19 @@ fn routed_asr_client(
     mut config: DiscoveryAsrConfig,
     attempt: &RoutedAttempt,
 ) -> Result<DiscoveryAsrClient, AsrClientError> {
-    if let Some(endpoint) = attempt
-        .model_base_url
-        .as_deref()
-        .or(attempt.provider_endpoint.as_deref())
-        .map(str::trim)
-        .filter(|value| !value.is_empty())
-    {
+    if attempt.protocol != Some(openplotva_llm::provider_schema::Protocol::DiscoveryAsr) {
+        return Err(AsrClientError::Discovery(
+            "selected protocol does not support this Discovery adapter".to_owned(),
+        ));
+    }
+    if let Some(endpoint) = attempt.endpoint() {
         config.base_url = endpoint.to_owned();
     }
-    if let Some(service_name) = attempt
-        .discovery_service_name
-        .as_deref()
-        .or_else(|| {
-            attempt
-                .provider_config
-                .get("service_name")
-                .and_then(serde_json::Value::as_str)
-        })
-        .map(str::trim)
-        .filter(|value| !value.is_empty())
-    {
+    let (service_name, endpoint_name) = attempt.discovery_names();
+    if let Some(service_name) = service_name {
         config.service_name = service_name.to_owned();
     }
-    if let Some(endpoint_name) = attempt
-        .discovery_endpoint_name
-        .as_deref()
-        .or_else(|| {
-            attempt
-                .provider_config
-                .get("endpoint_name")
-                .and_then(serde_json::Value::as_str)
-        })
-        .map(str::trim)
-        .filter(|value| !value.is_empty())
-    {
+    if let Some(endpoint_name) = endpoint_name {
         config.endpoint_name = endpoint_name.to_owned();
     }
     if attempt.model_name.trim().is_empty() {

crates/openplotva-app/src/callbacks.rs

@@ -3,12 +3,12 @@
 use std::{fmt, future::Future, pin::Pin, sync::Arc};
 
 use carapax::types::{
-    CallbackQuery as TelegramCallbackQuery, MaybeInaccessibleMessage, Update as TelegramUpdate,
-    UpdateType,
+    CallbackQuery as TelegramCallbackQuery, ChatMember as TelegramChatMember,
+    MaybeInaccessibleMessage, Update as TelegramUpdate, UpdateType,
 };
 use openplotva_telegram::{
-    CallbackHandlerKind, CallbackQueryRoute, TelegramOutboundMethod, callback_query_ack_method,
-    callback_query_route, settings_callback_ack_method,
+    CallbackHandlerKind, CallbackQueryRoute, TelegramOutboundMethod, build_get_chat_member_method,
+    callback_query_ack_method, callback_query_route, settings_callback_ack_method,
 };
 use thiserror::Error;
 use time::OffsetDateTime;
@@ -18,6 +18,120 @@ use crate::{
     updates::UpdateHandler,
 };
 
+/// Boxed future returned by Telegram membership APIs.
+pub type DeleteMemberFuture<'a, E> =
+    Pin<Box<dyn Future<Output = Result<TelegramChatMember, E>> + Send + 'a>>;
+
+pub trait DeleteMemberApi {
+    /// Concrete API error.
+    type Error: fmt::Display + Send + Sync + 'static;
+
+    /// Fetch one Telegram chat member.
+    fn get_chat_member<'a>(
+        &'a self,
+        chat_id: i64,
+        user_id: i64,
+    ) -> DeleteMemberFuture<'a, Self::Error>;
+}
+
+impl DeleteMemberApi for openplotva_telegram::TelegramClient {
+    type Error = carapax::api::ExecuteError;
+
+    fn get_chat_member<'a>(
+        &'a self,
+        chat_id: i64,
+        user_id: i64,
+    ) -> DeleteMemberFuture<'a, Self::Error> {
+        Box::pin(async move {
+            self.execute(build_get_chat_member_method(chat_id, user_id))
+                .await
+        })
+    }
+}
+
+/// Boxed future returned by delete callback effects.
+pub type DeleteEffectFuture<'a, E> = Pin<Box<dyn Future<Output = Result<(), E>> + Send + 'a>>;
+
+pub trait DeleteCallbackEffects {
+    /// Concrete effect error.
+    type Error: fmt::Display + Send + Sync + 'static;
+
+    /// Execute one direct Telegram Bot API method.
+    fn execute_delete_method<'a>(
+        &'a self,
+        method: TelegramOutboundMethod,
+    ) -> DeleteEffectFuture<'a, Self::Error>;
+}
+
+impl DeleteCallbackEffects for openplotva_telegram::TelegramClient {
+    type Error = carapax::api::ExecuteError;
+
+    fn execute_delete_method<'a>(
+        &'a self,
+        method: TelegramOutboundMethod,
+    ) -> DeleteEffectFuture<'a, Self::Error> {
+        Box::pin(async move { method.execute_with(self).await.map(|_| ()) })
+    }
+}
+
+#[must_use]
+pub fn telegram_member_can_delete(member: &TelegramChatMember) -> bool {
+    match member {
+        TelegramChatMember::Creator(_) => true,
+        TelegramChatMember::Administrator(admin) => admin.can_delete_messages,
+        TelegramChatMember::Kicked(_)
+        | TelegramChatMember::Left(_)
+        | TelegramChatMember::Member { .. }
+        | TelegramChatMember::Restricted(_) => false,
+    }
+}
+
+pub(crate) async fn try_execute_delete_method<Effects>(
+    effects: &Effects,
+    method: TelegramOutboundMethod,
+    context: &'static str,
+) where
+    Effects: DeleteCallbackEffects + Sync,
+{
+    let method_name = method.method_name();
+    if let Err(error) = effects.execute_delete_method(method).await {
+        tracing::warn!(
+            message = %error,
+            method = method_name,
+            context,
+            "generic delete callback Telegram side effect failed"
+        );
+    }
+}
+
+pub(crate) async fn can_delete_owned_content<MemberApi>(
+    member_api: &MemberApi,
+    query: &TelegramCallbackQuery,
+    data_user_id: i64,
+    data_chat_id: i64,
+) -> bool
+where
+    MemberApi: DeleteMemberApi + Sync,
+{
+    let requester_id = i64::from(query.from.id);
+    if requester_id == data_user_id {
+        return true;
+    }
+
+    match member_api.get_chat_member(data_chat_id, requester_id).await {
+        Ok(member) => telegram_member_can_delete(&member),
+        Err(error) => {
+            tracing::warn!(
+                message = %error,
+                chat_id = data_chat_id,
+                user_id = requester_id,
+                "failed to authorize delete callback"
+            );
+            false
+        }
+    }
+}
+
 /// Boxed future returned by callback-query side effects.
 pub type CallbackQueryFuture<'a, E> = Pin<Box<dyn Future<Output = Result<(), E>> + Send + 'a>>;