fix: align create_terminal return type with runtime handle

Signed-off-by: Chojan Shang <psiace@apache.org>

Author: PsiACE

src/acp/interfaces.py

@@ -64,6 +64,7 @@
     WriteTextFileRequest,
     WriteTextFileResponse,
 )
+from .terminal import TerminalHandle
 from .utils import param_model
 
 __all__ = ["Agent", "Client"]
@@ -111,7 +112,7 @@ async def create_terminal(
         env: list[EnvVariable] | None = None,
         output_byte_limit: int | None = None,
         **kwargs: Any,
-    ) -> CreateTerminalResponse: ...
+    ) -> CreateTerminalResponse | TerminalHandle: ...
 
     @param_model(TerminalOutputRequest)
     async def terminal_output(self, session_id: str, terminal_id: str, **kwargs: Any) -> TerminalOutputResponse: ...

src/acp/terminal.py

@@ -20,6 +20,10 @@ def __init__(self, terminal_id: str, session_id: str, conn: Connection) -> None:
         self._session_id = session_id
         self._conn = conn
 
+    @property
+    def terminal_id(self) -> str:
+        return self.id
+
     async def current_output(self) -> TerminalOutputResponse:
         response = await self._conn.send_request(
             CLIENT_METHODS["terminal_output"],

tests/test_rpc.py

@@ -10,6 +10,7 @@
     Agent,
     AuthenticateResponse,
     Client,
+    CreateTerminalResponse,
     InitializeResponse,
     LoadSessionResponse,
     NewSessionResponse,
@@ -24,13 +25,15 @@
     update_agent_message_text,
     update_tool_call,
 )
+from acp.core import AgentSideConnection, ClientSideConnection
 from acp.schema import (
     AgentMessageChunk,
     AllowedOutcome,
     AudioContentBlock,
     ClientCapabilities,
     DeniedOutcome,
     EmbeddedResourceContentBlock,
+    EnvVariable,
     HttpMcpServer,
     ImageContentBlock,
     Implementation,
@@ -130,6 +133,56 @@ async def test_session_notifications_flow(connect, client):
     assert client.notifications[0].session_id == "sess"
 
 
+@pytest.mark.asyncio
+async def test_on_connect_create_terminal_handle(server):
+    class _TerminalAgent(Agent):
+        __test__ = False
+
+        def __init__(self) -> None:
+            self._conn: Client | None = None
+            self.handle_id: str | None = None
+
+        def on_connect(self, conn: Client) -> None:
+            self._conn = conn
+
+        async def prompt(
+            self,
+            prompt: list[TextContentBlock],
+            session_id: str,
+            **kwargs: Any,
+        ) -> PromptResponse:
+            assert self._conn is not None
+            handle = await self._conn.create_terminal(command="echo", session_id=session_id)
+            self.handle_id = handle.terminal_id
+            return PromptResponse(stop_reason="end_turn")
+
+    class _TerminalClient(TestClient):
+        __test__ = False
+
+        async def create_terminal(
+            self,
+            command: str,
+            session_id: str,
+            args: list[str] | None = None,
+            cwd: str | None = None,
+            env: list[EnvVariable] | None = None,
+            output_byte_limit: int | None = None,
+            **kwargs: Any,
+        ) -> CreateTerminalResponse:
+            return CreateTerminalResponse(terminal_id="term-123")
+
+    agent = _TerminalAgent()
+    client = _TerminalClient()
+    agent_conn = AgentSideConnection(agent, server.server_writer, server.server_reader, listening=True)
+    client_conn = ClientSideConnection(client, server.client_writer, server.client_reader)
+
+    await client_conn.prompt(session_id="sess", prompt=[TextContentBlock(type="text", text="start")])
+    assert agent.handle_id == "term-123"
+
+    await client_conn.close()
+    await agent_conn.close()
+
+
 @pytest.mark.asyncio
 async def test_concurrent_reads(connect, client):
     for i in range(5):