fix(sdk): resolve async response secrets outside the event loop

openhands-sdk/openhands/sdk/agent/agent.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import json
 import re
 from collections.abc import Callable
@@ -1022,6 +1023,12 @@ async def _astep(
 
         message: Message = llm_response.message
         response_type = classify_response(message)
+        if response_type is not LLMResponseType.TOOL_CALLS:
+            # Uncached LookupSecrets may call back into this same server.
+            async with conversation._released_state_lock_during_io():
+                message = await asyncio.to_thread(
+                    self._mask_secrets, message, conversation
+                )
 
         match response_type:
             case LLMResponseType.TOOL_CALLS:

tests/sdk/agent/test_async_secret_masking.py

@@ -0,0 +1,103 @@
+import asyncio
+
+import pytest
+
+from openhands.sdk import Agent, Conversation
+from openhands.sdk.event import ActionEvent, MessageEvent
+from openhands.sdk.llm import Message, TextContent
+from openhands.sdk.llm.message import MessageToolCall
+from openhands.sdk.secret import LookupSecret
+from openhands.sdk.testing import TestLLM
+
+
+@pytest.mark.asyncio
+async def test_async_response_masks_loopback_lookup_secret_without_blocking(tmp_path):
+    requested = asyncio.Event()
+
+    async def serve_secret(reader, writer):
+        await reader.readuntil(b"\r\n\r\n")
+        requested.set()
+        body = b"loopback-secret-value"
+        writer.write(
+            b"HTTP/1.1 200 OK\r\nContent-Length: "
+            + str(len(body)).encode()
+            + b"\r\nConnection: close\r\n\r\n"
+            + body
+        )
+        await writer.drain()
+        writer.close()
+        await writer.wait_closed()
+
+    server = await asyncio.start_server(serve_secret, "127.0.0.1", 0)
+    port = server.sockets[0].getsockname()[1]
+    llm = TestLLM.from_messages(
+        [
+            Message(
+                role="assistant",
+                content=[TextContent(text="value loopback-secret-value")],
+            )
+        ]
+    )
+    conversation = Conversation(
+        agent=Agent(llm=llm, tools=[]),
+        workspace=str(tmp_path),
+        visualizer=None,
+        secrets={"TEST_TOKEN": LookupSecret(url=f"http://127.0.0.1:{port}/secret")},
+    )
+    try:
+        conversation.send_message("hello")
+        await asyncio.wait_for(conversation.arun(), timeout=5)
+        assert requested.is_set()
+        messages = [
+            event
+            for event in conversation.state.events
+            if isinstance(event, MessageEvent) and event.source == "agent"
+        ]
+        assert messages[-1].llm_message.content == [
+            TextContent(text="value <secret-hidden>")
+        ]
+    finally:
+        await asyncio.to_thread(conversation.close)
+        server.close()
+        await server.wait_closed()
+
+
+@pytest.mark.asyncio
+async def test_async_tool_thought_matches_sync_behavior(tmp_path):
+    thoughts = []
+    for use_async in [False, True]:
+        message = Message(
+            role="assistant",
+            content=[TextContent(text="existing tool thought")],
+            tool_calls=[
+                MessageToolCall(
+                    id="finish-call",
+                    origin="completion",
+                    name="finish",
+                    arguments='{"message":"done"}',
+                )
+            ],
+        )
+        conversation = Conversation(
+            agent=Agent(llm=TestLLM.from_messages([message]), tools=[]),
+            workspace=str(tmp_path),
+            visualizer=None,
+            secrets={"TOKEN": "existing tool thought"},
+        )
+        try:
+            conversation.send_message("finish")
+            if use_async:
+                await conversation.arun()
+            else:
+                await asyncio.to_thread(conversation.run)
+            thoughts.append(
+                [
+                    event.thought
+                    for event in conversation.state.events
+                    if isinstance(event, ActionEvent)
+                ]
+            )
+        finally:
+            await asyncio.to_thread(conversation.close)
+    assert thoughts[0]
+    assert thoughts[0] == thoughts[1]