Preserve pending drawing requests across initiator follow-ups

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

@@ -6511,6 +6511,139 @@ async fn explicit_image_request_never_publishes_a_promise_after_exhausted_repair
     Ok(())
 }
 
+#[tokio::test]
+async fn pending_image_request_survives_initiator_thanks() -> Result<(), Box<dyn Error>> {
+    let now = OffsetDateTime::from_unix_timestamp(1_779_193_800)?;
+    for observed in [true, false] {
+        let registry = Arc::new(crate::dialog_turn::DialogSessionRegistry::new());
+        let key = crate::dialog_turn::SessionKey::new(42, Some(9));
+        let toolbox = Arc::new(SessionToolboxStub::with_queued_draw("image-ticket"));
+        let wiring = crate::dialog_turn::SessionWorkerWiring {
+            registry: registry.clone(),
+            ..session_wiring(toolbox.clone(), None)
+        };
+        let queue = InMemoryTaskQueue::new();
+        let mut params = dialog_params("нарисуй рыбу");
+        params.meta = serde_json::json!({"requested_tool": "draw_image"});
+        let job = queue.assign(DIALOG_AIFARM_QUEUE_NAME, new_dialog_job_at(params, now));
+        let materializer = MaterializerStub::default();
+        let mut thanks = dialog_params("Спасибо большое Плотва");
+        thanks.message_id += 1;
+        if observed {
+            materializer
+                .observed
+                .lock()
+                .expect("observed")
+                .push(serde_json::json!({
+                    "message_id": thanks.message_id, "user_id": thanks.user_id,
+                    "text": thanks.message_text, "meta": {}
+                }));
+        }
+        let draw = step_tools(
+            "",
+            vec![(
+                "draw",
+                openplotva_dialog::ToolStep {
+                    step: "draw_image".into(),
+                    prompt: "A fish".into(),
+                    ..Default::default()
+                },
+            )],
+        );
+        let provider = StepProviderStub::with_steps(if observed {
+            vec![Ok(draw)]
+        } else {
+            vec![Ok(step_text("Сейчас подумаю.")), Ok(draw)]
+        })
+        .with_on_call(Box::new(move |index| {
+            if !observed && index == 1 {
+                assert!(registry.inject(
+                    key,
+                    job,
+                    crate::dialog_turn::InjectedMessage {
+                        params: thanks.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,
+            &materializer,
+            &NoopDialogToolCallHistoryStore,
+            session_options(now, &outcomes, &wiring),
+        )
+        .await;
+        assert!(report.completed, "{report:?}");
+        for request in provider.requests() {
+            assert_eq!(
+                request.required_tool.as_deref(),
+                Some("draw_image"),
+                "observed={observed}"
+            );
+        }
+        let draws = toolbox.draw_requests.lock().expect("draws");
+        assert_eq!(draws.len(), 1);
+        assert_eq!(draws[0].context.message_text, "нарисуй рыбу");
+    }
+    Ok(())
+}
+
+#[tokio::test]
+async fn pending_image_request_can_be_cancelled_before_the_first_step() -> Result<(), Box<dyn Error>>
+{
+    let now = OffsetDateTime::from_unix_timestamp(1_779_193_800)?;
+    for text in [
+        "Плотва, не рисуй",
+        "Отмена!",
+        "стоп",
+        "не надо рисовать",
+        "stop drawing",
+    ] {
+        let queue = InMemoryTaskQueue::new();
+        let mut params = dialog_params("нарисуй рыбу");
+        params.meta = serde_json::json!({"requested_tool":"draw_image"});
+        let message_id = params.message_id + 1;
+        let user_id = params.user_id;
+        queue.assign(DIALOG_AIFARM_QUEUE_NAME, new_dialog_job_at(params, now));
+        let materializer = MaterializerStub::default();
+        materializer
+            .observed
+            .lock()
+            .expect("observed")
+            .push(serde_json::json!({
+                "message_id":message_id, "user_id":user_id, "text":text, "meta":{}
+            }));
+        let toolbox = Arc::new(SessionToolboxStub::with_queued_draw("image-ticket"));
+        let wiring = session_wiring(toolbox.clone(), None);
+        let provider = StepProviderStub::with_steps(vec![Ok(step_text("Отменила."))]);
+        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,
+            &materializer,
+            &NoopDialogToolCallHistoryStore,
+            session_options(now, &outcomes, &wiring),
+        )
+        .await;
+        assert!(report.completed, "{text}: {report:?}");
+        assert_eq!(provider.requests()[0].required_tool, None, "{text}");
+        assert!(toolbox.draw_requests.lock().expect("draws").is_empty());
+    }
+    Ok(())
+}
+
 #[tokio::test]
 async fn newer_initiator_image_intent_replaces_the_previous_attempt() -> Result<(), Box<dyn Error>>
 {

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

@@ -319,6 +319,39 @@ fn pending_image_tool<'a>(
         })
 }
 
+fn replaces_pending_image_request(pending: bool, requested_tool: &str, text: &str) -> bool {
+    if !pending || matches!(requested_tool, "draw_image" | "draw_api") {
+        return true;
+    }
+    let text = text.trim().to_lowercase();
+    let text = text
+        .strip_prefix("плотва")
+        .or_else(|| text.strip_prefix("plotva"))
+        .unwrap_or(&text)
+        .trim_matches(|c: char| c.is_whitespace() || ",.!?".contains(c));
+    matches!(
+        text,
+        "отмена" | "стоп" | "cancel" | "/cancel" | "/cancel_drawing"
+    ) || [
+        "не рисуй",
+        "не надо рисовать",
+        "не нужно рисовать",
+        "перестань рисовать",
+        "отмени рисунок",
+        "отмени генерацию",
+        "не генерируй",
+        "cancel drawing",
+        "stop drawing",
+        "do not draw",
+        "don't draw",
+    ]
+    .iter()
+    .any(|prefix| {
+        text.strip_prefix(prefix)
+            .is_some_and(|tail| tail.chars().next().is_none_or(|c| !c.is_alphanumeric()))
+    })
+}
+
 fn record_rejected_final(agent: &mut openplotva_agent::AgentLoop, text: &str, reason: &str) {
     agent.transcript.push(SessionMessage::Assistant {
         text: text.to_owned(),
@@ -571,7 +604,11 @@ where
                     agent.observed_message_id = agent.observed_message_id.max(id);
                 }
                 message["can_change_task"] = Value::Bool(message.get("user_id").and_then(Value::as_i64) == Some(active_params.user_id));
-                if message["can_change_task"] == true {
+                if message["can_change_task"] == true && replaces_pending_image_request(
+                    pending_image_tool(&tool_context, &recorded_tool_calls, requested_tool_call_offset).is_some(),
+                    message["meta"]["requested_tool"].as_str().unwrap_or_default(),
+                    message["text"].as_str().unwrap_or_default(),
+                ) {
                     tool_context.message_id = message["message_id"].as_i64().and_then(|id| i32::try_from(id).ok()).unwrap_or(tool_context.message_id);
                     tool_context.message_meta.requested_tool = message["meta"]["requested_tool"].as_str().unwrap_or_default().to_owned();
                     requested_tool_call_offset = recorded_tool_calls.len();
@@ -590,7 +627,12 @@ where
                 if let Ok(meta) = serde_json::from_value::<openplotva_core::ChatMessageMeta>(params.meta.clone()) {
                     extend_context_images(&mut tool_context, params.message_id, meta.attachments);
                 }
-                if params.message_id > previously_observed_message_id && params.message_id >= tool_context.message_id && params.user_id == active_params.user_id {
+                if params.message_id > previously_observed_message_id && params.message_id >= tool_context.message_id && params.user_id == active_params.user_id
+                    && replaces_pending_image_request(
+                        pending_image_tool(&tool_context, &recorded_tool_calls, requested_tool_call_offset).is_some(),
+                        params.meta["requested_tool"].as_str().unwrap_or_default(),
+                        &params.message_text,
+                    ) {
                     tool_context.message_id = params.message_id;
                     tool_context.message_meta.requested_tool = params.meta["requested_tool"].as_str().unwrap_or_default().to_owned();
                     requested_tool_call_offset = recorded_tool_calls.len();