Send Gradius dialogue banners as rich messages after the answer

crates/openplotva-app/src/dialog_jobs/effects.rs

@@ -570,11 +570,7 @@ fn dialog_intermediate_outbox_batch(
         bot_id,
         chat_id: Some(params.chat_id),
         thread_id: params.thread_id,
-        ordering_key: format!(
-            "dialog:{bot_id}:{}:{}",
-            params.chat_id,
-            params.thread_id.unwrap_or_default()
-        ),
+        ordering_key: dialog_outbox_ordering_key(bot_id, params.chat_id, params.thread_id),
         causation_update_id,
         dialog_job_id: Some(dialog_job_id),
         trigger_message_id: Some(i64::from(params.message_id)),
@@ -585,6 +581,18 @@ fn dialog_intermediate_outbox_batch(
     }
 }
 
+/// Per-chat delivery order shared by dialog messages and the banners queued after them.
+pub(crate) fn dialog_outbox_ordering_key(
+    bot_id: i64,
+    chat_id: i64,
+    thread_id: Option<i32>,
+) -> String {
+    format!(
+        "dialog:{bot_id}:{chat_id}:{}",
+        thread_id.unwrap_or_default()
+    )
+}
+
 fn dialog_answer_batch_id(bot_id: i64, dialog_job_id: i64) -> String {
     format!("dialog-answer:v1:{bot_id}:{dialog_job_id}")
 }
@@ -658,12 +666,7 @@ async fn enqueue_durable_dialog_answer(
         bot_id: outbox.bot_id,
         chat_id: Some(params.chat_id),
         thread_id: params.thread_id,
-        ordering_key: format!(
-            "dialog:{}:{}:{}",
-            outbox.bot_id,
-            params.chat_id,
-            params.thread_id.unwrap_or_default()
-        ),
+        ordering_key: dialog_outbox_ordering_key(outbox.bot_id, params.chat_id, params.thread_id),
         causation_update_id,
         dialog_job_id: Some(dialog_job_id),
         trigger_message_id: Some(i64::from(params.message_id)),

crates/openplotva-app/src/dialog_jobs/tests.rs

@@ -4091,6 +4091,8 @@ fn session_options<'a>(
 
 #[derive(Clone, Default)]
 struct GradiusAppenderStub {
+    banner: Option<crate::gradius_ads::GradiusAdBanner>,
+    queued_banners: Arc<Mutex<Vec<crate::gradius_ads::GradiusBannerTarget>>>,
     requests: Arc<Mutex<Vec<crate::gradius_ads::GradiusAdAppendRequest>>>,
     render_errors: Arc<Mutex<Vec<(i64, String)>>>,
     queued: Arc<Mutex<Vec<(i64, String)>>>,
@@ -4104,17 +4106,38 @@ impl crate::gradius_ads::GradiusAdAppender for GradiusAppenderStub {
     fn append<'a>(
         &'a self,
         request: crate::gradius_ads::GradiusAdAppendRequest,
-    ) -> crate::gradius_ads::GradiusServiceFuture<'a, Option<crate::gradius_ads::GradiusAdTail>>
+    ) -> crate::gradius_ads::GradiusServiceFuture<'a, Option<crate::gradius_ads::GradiusPreparedAd>>
     {
         Box::pin(async move {
             self.requests
                 .lock()
                 .expect("Gradius requests")
                 .push(request);
-            Ok(Some(crate::gradius_ads::GradiusAdTail {
-                opportunity_id: 91,
-                html: TEST_GRADIUS_TAIL.to_owned(),
-            }))
+            if let Some(banner) = &self.banner {
+                return Ok(Some(crate::gradius_ads::GradiusPreparedAd::Banner(
+                    banner.clone(),
+                )));
+            }
+            Ok(Some(crate::gradius_ads::GradiusPreparedAd::Tail(
+                crate::gradius_ads::GradiusAdTail {
+                    opportunity_id: 91,
+                    html: TEST_GRADIUS_TAIL.to_owned(),
+                },
+            )))
+        })
+    }
+
+    fn queue_banner<'a>(
+        &'a self,
+        banner: &'a crate::gradius_ads::GradiusAdBanner,
+        target: crate::gradius_ads::GradiusBannerTarget,
+    ) -> crate::gradius_ads::GradiusServiceFuture<'a, String> {
+        Box::pin(async move {
+            self.queued_banners
+                .lock()
+                .expect("Gradius queued banners")
+                .push(target);
+            Ok(format!("gradius-banner:v1:test:{}", banner.opportunity_id))
         })
     }
 
@@ -4282,6 +4305,138 @@ async fn session_marks_prepared_gradius_ad_failed_when_answer_enqueue_fails()
     Ok(())
 }
 
+fn gradius_banner_appender() -> GradiusAppenderStub {
+    GradiusAppenderStub {
+        banner: Some(crate::gradius_ads::GradiusAdBanner {
+            opportunity_id: 92,
+            html: "<h2>Баннер</h2><footer>Реклама</footer>".to_owned(),
+            keyboard: Vec::new(),
+        }),
+        ..GradiusAppenderStub::default()
+    }
+}
+
+#[tokio::test]
+async fn session_sends_answer_unchanged_and_queues_gradius_banner_after_it()
+-> Result<(), Box<dyn Error>> {
+    let now = OffsetDateTime::from_unix_timestamp(1_779_193_800)?;
+    let queue = InMemoryTaskQueue::new();
+    let job_id = queue.assign(
+        DIALOG_AIFARM_QUEUE_NAME,
+        new_dialog_job_at(dialog_params("расскажи шутку"), now),
+    );
+    let provider = StepProviderStub::with_steps(vec![Ok(step_text("Базовый ответ &&&banner&&&"))]);
+    let appender = gradius_banner_appender();
+    let toolbox: Arc<dyn openplotva_dialog::DialogToolbox> =
+        Arc::new(SessionToolboxStub::default());
+    let mut wiring = session_wiring(toolbox, None);
+    wiring.gradius = Some(Arc::new(appender.clone()));
+    let effects = EffectsStub::default();
+    let outcomes = crate::dialog_turn::DialogTurnObserver::new(
+        crate::dialog_turn::RuntimeTurnOutcomeBuffer::new(8),
+        None,
+    );
+
+    let report = process_dialog_job_once_in_queue_with_materializer_history_and_retry_at(
+        &queue,
+        &provider,
+        &effects,
+        &BasicDialogInputMaterializer,
+        &NoopDialogToolCallHistoryStore,
+        session_options(now, &outcomes, &wiring),
+    )
+    .await;
+
+    assert!(report.sent_answer, "{report:?}");
+    let sent = effects.sent();
+    assert_eq!(sent.len(), 1);
+    assert!(
+        sent[0].1.starts_with("Базовый ответ &amp;&amp;&amp;banner"),
+        "{}",
+        sent[0].1
+    );
+    assert!(!sent[0].1.contains("Реклама"));
+    assert_eq!(
+        effects.answer_options(),
+        vec![DialogAnswerSendOptions::default()]
+    );
+    assert_eq!(
+        appender.requests.lock().expect("Gradius requests")[0].assistant_text,
+        "Базовый ответ &&&banner&&&"
+    );
+    assert_eq!(
+        appender
+            .queued_banners
+            .lock()
+            .expect("Gradius queued banners")
+            .as_slice(),
+        &[crate::gradius_ads::GradiusBannerTarget {
+            dialog_job_id: job_id,
+            chat_id: dialog_params("").chat_id,
+            thread_id: dialog_params("").thread_id,
+        }]
+    );
+    assert!(appender.queued.lock().expect("Gradius queued").is_empty());
+    assert!(
+        appender
+            .failed_delivery_opportunities
+            .lock()
+            .expect("Gradius failed delivery opportunities")
+            .is_empty()
+    );
+    Ok(())
+}
+
+#[tokio::test]
+async fn session_closes_gradius_banner_without_queueing_when_answer_enqueue_fails()
+-> Result<(), Box<dyn Error>> {
+    let now = OffsetDateTime::from_unix_timestamp(1_779_193_800)?;
+    let queue = InMemoryTaskQueue::new();
+    queue.assign(
+        DIALOG_AIFARM_QUEUE_NAME,
+        new_dialog_job_at(dialog_params("покажи рекламу"), now),
+    );
+    let provider = StepProviderStub::with_steps(vec![Ok(step_text("Базовый ответ"))]);
+    let appender = gradius_banner_appender();
+    let toolbox: Arc<dyn openplotva_dialog::DialogToolbox> =
+        Arc::new(SessionToolboxStub::default());
+    let mut wiring = session_wiring(toolbox, None);
+    wiring.gradius = Some(Arc::new(appender.clone()));
+    let effects = EffectsStub::failing();
+    let outcomes = crate::dialog_turn::DialogTurnObserver::new(
+        crate::dialog_turn::RuntimeTurnOutcomeBuffer::new(8),
+        None,
+    );
+
+    let report = process_dialog_job_once_in_queue_with_materializer_history_and_retry_at(
+        &queue,
+        &provider,
+        &effects,
+        &BasicDialogInputMaterializer,
+        &NoopDialogToolCallHistoryStore,
+        session_options(now, &outcomes, &wiring),
+    )
+    .await;
+
+    assert_eq!(report.send_error.as_deref(), Some("send down"));
+    assert!(
+        appender
+            .queued_banners
+            .lock()
+            .expect("Gradius queued banners")
+            .is_empty()
+    );
+    assert_eq!(
+        appender
+            .failed_delivery_opportunities
+            .lock()
+            .expect("Gradius failed delivery opportunities")
+            .as_slice(),
+        &[(92, "send down".to_owned())]
+    );
+    Ok(())
+}
+
 #[tokio::test]
 async fn session_runs_tool_round_then_final_text() -> Result<(), Box<dyn Error>> {
     let now = OffsetDateTime::from_unix_timestamp(1_779_193_800)?;

crates/openplotva-app/src/dialog_turn/session.rs

@@ -1078,6 +1078,7 @@ where
             let mut contains_advertising = false;
             let mut advertising_tail_bytes = None;
             let mut advertising_opportunity_id = None;
+            let mut advertising_banner = None;
             if let Some(gradius) = cfg.gradius {
                 let request = crate::gradius_ads::GradiusAdAppendRequest {
                     dialog_job_id: ctx.item_id,
@@ -1106,7 +1107,10 @@ where
                     }
                 } else {
                     match gradius.append(request).await {
-                        Ok(Some(tail)) => {
+                        Ok(Some(crate::gradius_ads::GradiusPreparedAd::Banner(banner))) => {
+                            advertising_banner = Some(banner);
+                        }
+                        Ok(Some(crate::gradius_ads::GradiusPreparedAd::Tail(tail))) => {
                             let tail_bytes = tail.html.len();
                             let candidate = format!("{sanitized}\n\n{}", tail.html);
                             if let Err(error) = validate_dialog_answer_deliverable(&candidate) {
@@ -1189,6 +1193,19 @@ where
                     );
                 }
             }
+            if let (Some(banner), Some(gradius)) = (advertising_banner.as_ref(), cfg.gradius) {
+                queue_gradius_banner(
+                    gradius,
+                    banner,
+                    send_result.as_ref().map(|_| ()).map_err(ToString::to_string),
+                    crate::gradius_ads::GradiusBannerTarget {
+                        dialog_job_id: ctx.item_id,
+                        chat_id: active_params.chat_id,
+                        thread_id: active_params.thread_id,
+                    },
+                )
+                .await;
+            }
             if let (Err(send_error), Some(opportunity_id), Some(gradius)) = (
                 send_result.as_ref(),
                 advertising_opportunity_id,
@@ -2342,6 +2359,54 @@ where
     }
 }
 
+/// Queue the banner only after the answer was accepted, so it follows the answer
+/// in the chat's outbox order; a failed answer closes the banner without sending it.
+async fn queue_gradius_banner(
+    gradius: &dyn crate::gradius_ads::GradiusAdAppender,
+    banner: &crate::gradius_ads::GradiusAdBanner,
+    answer: Result<(), String>,
+    target: crate::gradius_ads::GradiusBannerTarget,
+) {
+    let opportunity_id = banner.opportunity_id;
+    match answer {
+        Ok(()) => match gradius.queue_banner(banner, target).await {
+            Ok(batch_id) => tracing::info!(
+                opportunity_id,
+                job_id = target.dialog_job_id,
+                integration_kind = "native_dialogue",
+                placement = "banner",
+                delivery_state = "queued",
+                batch_id,
+                "Gradius banner queued after the dialog answer"
+            ),
+            Err(error) => tracing::warn!(
+                opportunity_id,
+                job_id = target.dialog_job_id,
+                integration_kind = "native_dialogue",
+                placement = "banner",
+                %error,
+                "failed to queue Gradius banner"
+            ),
+        },
+        Err(send_error) => {
+            if let Err(error) = gradius
+                .mark_delivery_failed(opportunity_id, &send_error)
+                .await
+            {
+                tracing::warn!(
+                    opportunity_id,
+                    job_id = target.dialog_job_id,
+                    integration_kind = "native_dialogue",
+                    placement = "banner",
+                    delivery_state = "failed",
+                    %error,
+                    "failed to persist Gradius banner enqueue failure"
+                );
+            }
+        }
+    }
+}
+
 async fn append_session_queued_marker<Queue>(
     queue: &Queue,
     job_id: i64,