From e45542359fd3caee6527d220c72551edd2d3b537 Mon Sep 17 00:00:00 2001
From: Lanre Shittu <136805224+Shizoqua@users.noreply.github.com>
Date: Sun, 1 Nov 2026 09:10:30 +0000
Subject: [PATCH] FIX: Read manual XPIA input off the event loop

---
 pyrit/executor/workflow/xpia.py           |  6 ++-
 tests/unit/executor/workflow/test_xpia.py | 50 +++++++++++++++++++++++
 2 files changed, 54 insertions(+), 2 deletions(-)

diff --git a/pyrit/executor/workflow/xpia.py b/pyrit/executor/workflow/xpia.py
index 8381fde697..a707af6dd5 100644
--- a/pyrit/executor/workflow/xpia.py
+++ b/pyrit/executor/workflow/xpia.py
@@ -1,6 +1,7 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT license.
 
+import asyncio
 import logging
 import uuid
 from dataclasses import dataclass, field
@@ -661,10 +662,11 @@ async def _setup_async(self, *, context: XPIAContext) -> None:
                 modified to include the manual processing callback.
         """
 
-        # Create the manual input callback
+        # Create the manual input callback. The console read runs in a worker thread so other
+        # async work keeps running while the operator triggers the processing target.
         async def manual_input_async() -> str:
             prompt = "Please trigger the processing target's execution and paste the output here:"
-            return input(prompt)
+            return await asyncio.to_thread(input, prompt)
 
         # Set the processing callback on the context
         context.processing_callback = manual_input_async
diff --git a/tests/unit/executor/workflow/test_xpia.py b/tests/unit/executor/workflow/test_xpia.py
index 243e431b40..a2a1615983 100644
--- a/tests/unit/executor/workflow/test_xpia.py
+++ b/tests/unit/executor/workflow/test_xpia.py
@@ -1,6 +1,8 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT license.
 
+import asyncio
+import threading
 import uuid
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -9,6 +11,7 @@
 
 from pyrit.executor.workflow.xpia import (
     XPIAContext,
+    XPIAManualProcessingWorkflow,
     XPIAResult,
     XPIAStatus,
     XPIAWorkflow,
@@ -698,3 +701,50 @@ async def test_xpia_test_setup_raises_when_processing_prompt_is_none(
         assert context.processing_callback is not None
         with pytest.raises(RuntimeError, match="context.processing_prompt is not initialized"):
             await context.processing_callback()
+
+
+@pytest.fixture
+def manual_workflow(mock_attack_setup_target: MagicMock, mock_scorer: MagicMock) -> XPIAManualProcessingWorkflow:
+    """Create a manual processing XPIA workflow for testing."""
+    return XPIAManualProcessingWorkflow(attack_setup_target=mock_attack_setup_target, scorer=mock_scorer)
+
+
+@pytest.mark.usefixtures("patch_central_database")
+class TestXPIAManualProcessingWorkflow:
+    """Tests for the manual processing callback created during setup."""
+
+    async def test_setup_async_sets_callback_that_returns_operator_input(
+        self, manual_workflow: XPIAManualProcessingWorkflow, valid_message: Message
+    ) -> None:
+        context = XPIAContext(attack_content=valid_message)
+        await manual_workflow._setup_async(context=context)
+        assert context.processing_callback is not None
+
+        with patch("builtins.input", return_value="processed output") as mock_input:
+            response = await context.processing_callback()
+
+        assert response == "processed output"
+        mock_input.assert_called_once_with(
+            "Please trigger the processing target's execution and paste the output here:"
+        )
+
+    async def test_manual_input_does_not_block_the_event_loop(
+        self, manual_workflow: XPIAManualProcessingWorkflow, valid_message: Message
+    ) -> None:
+        context = XPIAContext(attack_content=valid_message)
+        await manual_workflow._setup_async(context=context)
+        assert context.processing_callback is not None
+        event_loop_released_input = threading.Event()
+
+        def wait_for_event_loop(prompt: str) -> str:
+            # Only a task on the event loop sets this, so the read can finish only if the loop keeps running.
+            assert event_loop_released_input.wait(timeout=5)
+            return "processed output"
+
+        async def release_input_async() -> None:
+            event_loop_released_input.set()
+
+        with patch("builtins.input", side_effect=wait_for_event_loop):
+            response, _ = await asyncio.gather(context.processing_callback(), release_input_async())
+
+        assert response == "processed output"