features/a2a_stdio_transport.feature

@@ -0,0 +1,155 @@
+Feature: A2A stdio transport for local-mode communication
+  As a developer using local-mode agent communication
+  I want the A2aStdioTransport to handle subprocess communication correctly
+  So that JSON-RPC messages are sent and received reliably
+
+  @coverage
+  Scenario: Transport initializes with no process
+    Given a new A2aStdioTransport instance
+    Then the stdio transport should not be connected
+    And the stdio transport process should be None
+
+  @coverage
+  Scenario: Send raises when not connected
+    Given a new A2aStdioTransport instance
+    When I try to send a request without connecting
+    Then a RuntimeError should be raised about not connected
+
+  @coverage
+  Scenario: Send raises for non-A2aRequest input
+    Given a connected A2aStdioTransport with a mock process
+    When I try to send a non-A2aRequest object
+    Then a TypeError should be raised about A2aRequest
+
+  @coverage
+  Scenario: Send succeeds with valid request and mock response
+    Given a connected A2aStdioTransport with a mock process
+    And the mock process returns a valid JSON-RPC response
+    When I send a valid A2aRequest
+    Then I should receive an A2aResponse
+
+  @coverage
+  Scenario: Send raises on invalid JSON response
+    Given a connected A2aStdioTransport with a mock process
+    And the mock process returns invalid JSON
+    When I try to send a valid A2aRequest
+    Then a RuntimeError should be raised about invalid JSON
+
+  @coverage
+  Scenario: Send raises when subprocess closes unexpectedly
+    Given a connected A2aStdioTransport with a mock process
+    And the mock process returns empty response
+    When I try to send a valid A2aRequest
+    Then a RuntimeError should be raised about closed unexpectedly
+
+  @coverage
+  Scenario: Send raises when stdin is unavailable
+    Given a connected A2aStdioTransport with a mock process
+    And the mock process has no stdin
+    When I try to send a valid A2aRequest
+    Then a RuntimeError should be raised about stdin
+
+  @coverage
+  Scenario: Send raises when stdout is unavailable
+    Given a connected A2aStdioTransport with a mock process
+    And the mock process has no stdout but valid stdin
+    When I try to send a valid A2aRequest
+    Then a RuntimeError should be raised about stdout
+
+  @coverage
+  Scenario: Connect raises for empty agent path
+    Given a new A2aStdioTransport instance
+    When I try to connect with empty agent path
+    Then a ValueError should be raised about agent_path
+
+  @coverage
+  Scenario: Connect raises when already connected
+    Given a connected A2aStdioTransport with a mock process
+    When I try to connect again with a valid path
+    Then a RuntimeError should be raised about already connected
+
+  @coverage
+  Scenario: Disconnect is a no-op when not connected
+    Given a new A2aStdioTransport instance
+    When I call disconnect
+    Then no stdio transport error should be raised
+
+  @coverage
+  Scenario: Disconnect closes stdin and waits for process
+    Given a connected A2aStdioTransport with a mock process
+    When I call disconnect
+    Then the stdio transport should not be connected
+    And mock stdin close should have been called
+    And mock process wait should have been called
+
+  @coverage
+  Scenario: Disconnect terminates when wait times out
+    Given a connected A2aStdioTransport with a mock process
+    And the mock process times out on first wait
+    When I call disconnect
+    Then the stdio transport should not be connected
+    And mock process terminate should have been called
+
+  @coverage
+  Scenario: Connect with Python module path
+    Given a new A2aStdioTransport instance
+    And subprocess Popen is mocked to succeed
+    When I connect with agent path "cleveragents.a2a.agent"
+    Then the stdio transport should be connected
+
+  @coverage
+  Scenario: Connect with executable path
+    Given a new A2aStdioTransport instance
+    And subprocess Popen is mocked to succeed
+    When I connect with agent path "/usr/local/bin/agent"
+    Then the stdio transport should be connected
+
+  @coverage
+  Scenario: Connect with .py file path
+    Given a new A2aStdioTransport instance
+    And subprocess Popen is mocked to succeed
+    When I connect with agent path "agent.py"
+    Then the stdio transport should be connected
+
+  @coverage
+  Scenario: Connect raises for file not found
+    Given a new A2aStdioTransport instance
+    And subprocess Popen raises FileNotFoundError
+    When I try to connect with agent path "/nonexistent/agent"
+    Then a RuntimeError should be raised about agent not found
+
+  # -------------------------------------------------------------------
+  # TransportSelector — selection logic
+  # -------------------------------------------------------------------
+
+  @coverage
+  Scenario: TransportSelector selects stdio transport when no server URL
+    When I select a transport with no server URL
+    Then the selected transport should be an A2aStdioTransport
+
+  @coverage
+  Scenario: TransportSelector selects stdio transport when server URL is None
+    When I select a transport with server URL None
+    Then the selected transport should be an A2aStdioTransport
+
+  @coverage
+  Scenario: TransportSelector selects HTTP transport when server URL is provided
+    When I select a transport with server URL "http://localhost:8080"
+    Then the selected transport should be an A2aHttpTransport
+
+  @coverage
+  Scenario: TransportSelector selects stdio transport when server URL is empty string
+    When I select a transport with server URL ""
+    Then the selected transport should be an A2aStdioTransport
+
+  @coverage
+  Scenario: TransportSelector returns a new instance each call for stdio
+    When I select a transport with no server URL
+    And I select a transport with no server URL again
+    Then the two stdio transports should be different instances
+
+  @coverage
+  Scenario: TransportSelector returns a new instance each call for HTTP
+    When I select a transport with server URL "http://localhost:8080"
+    And I select a transport with server URL "http://localhost:8080" again
+    Then the two HTTP transports should be different instances

features/mocks/a2a_bad_json_agent.py

@@ -0,0 +1,24 @@
+"""Mock bad-JSON agent for A2A stdio transport tests.
+
+Reads from stdin and writes invalid JSON to stdout.
+Used to test error handling in A2aStdioTransport.
+"""
+
+from __future__ import annotations
+
+import sys
+
+
+def main() -> None:
+    """Read from stdin and write invalid JSON to stdout."""
+    for line in sys.stdin:
+        line = line.strip()
+        if not line:
+            continue
+        # Write invalid JSON to trigger JSONDecodeError in the transport
+        sys.stdout.write("this is not valid json\n")
+        sys.stdout.flush()
+
+
+if __name__ == "__main__":
+    main()

features/mocks/a2a_echo_agent.py

@@ -0,0 +1,39 @@
+"""Mock echo agent for A2A stdio transport tests.
+
+Reads JSON-RPC 2.0 requests from stdin and echoes them back as success
+responses on stdout.  Used as a subprocess in unit tests.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+
+
+def main() -> None:
+    """Read JSON-RPC 2.0 requests from stdin and echo responses to stdout."""
+    for line in sys.stdin:
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            request = json.loads(line)
+            response = {
+                "jsonrpc": "2.0",
+                "id": request.get("id", ""),
+                "result": {"echoed": True, "method": request.get("method", "")},
+            }
+            sys.stdout.write(json.dumps(response) + "\n")
+            sys.stdout.flush()
+        except json.JSONDecodeError:
+            error_response = {
+                "jsonrpc": "2.0",
+                "id": "",
+                "error": {"code": -32700, "message": "Parse error"},
+            }
+            sys.stdout.write(json.dumps(error_response) + "\n")
+            sys.stdout.flush()
+
+
+if __name__ == "__main__":
+    main()

features/mocks/mock_stdio_process.py

@@ -0,0 +1,146 @@
+"""Mock subprocess for A2A stdio transport unit tests.
+
+Provides a fake Popen-compatible object that simulates subprocess
+communication without launching a real process.  Used in BDD unit tests
+to avoid real subprocess overhead and flakiness.
+"""
+
+from __future__ import annotations
+
+import json
+
+
+class MockStdinWriter:
+    """Fake stdin that captures written data."""
+
+    def __init__(self) -> None:
+        self._buffer: list[str] = []
+        self.closed: bool = False
+
+    def write(self, data: str) -> None:
+        if not self.closed:
+            self._buffer.append(data)
+
+    def flush(self) -> None:
+        pass
+
+    def close(self) -> None:
+        self.closed = True
+
+    @property
+    def lines(self) -> list[str]:
+        """Return all written lines."""
+        return self._buffer
+
+
+class MockStdoutReader:
+    """Fake stdout that returns pre-configured responses."""
+
+    def __init__(self, responses: list[str]) -> None:
+        self._responses = iter(responses)
+        self.closed: bool = False
+
+    def readline(self) -> str:
+        if self.closed:
+            return ""
+        try:
+            return next(self._responses)
+        except StopIteration:
+            return ""
+
+
+class MockEchoProcess:
+    """Mock process that echoes JSON-RPC 2.0 requests as success responses.
+
+    Simulates a subprocess that reads JSON-RPC requests and returns
+    success responses with the same id.
+    """
+
+    def __init__(self) -> None:
+        self.pid: int = 99999
+        self.stdin: MockStdinWriter = MockStdinWriter()
+        self._returncode: int | None = None
+
+    def _make_response(self, request_json: str) -> str:
+        """Build a success response for the given request JSON."""
+        try:
+            request = json.loads(request_json.strip())
+            response = {
+                "jsonrpc": "2.0",
+                "id": request.get("id", ""),
+                "result": {"echoed": True, "method": request.get("method", "")},
+            }
+            return json.dumps(response) + "\n"
+        except json.JSONDecodeError:
+            return (
+                json.dumps(
+                    {
+                        "jsonrpc": "2.0",
+                        "id": "",
+                        "error": {"code": -32700, "message": "Parse error"},
+                    }
+                )
+                + "\n"
+            )
+
+    @property
+    def stdout(self) -> MockStdoutReader:
+        """Return a stdout reader that responds to the last written request."""
+        # Build response based on what was written to stdin
+        lines = self.stdin.lines
+        if lines:
+            last_line = lines[-1].strip()
+            if last_line:
+                response = self._make_response(last_line)
+                return MockStdoutReader([response])
+        return MockStdoutReader([""])
+
+    def wait(self, timeout: float | None = None) -> int:
+        self._returncode = 0
+        return 0
+
+    def terminate(self) -> None:
+        self._returncode = -15
+
+    def kill(self) -> None:
+        self._returncode = -9
+
+
+class MockBadJsonProcess:
+    """Mock process that returns invalid JSON responses."""
+
+    def __init__(self) -> None:
+        self.pid: int = 99998
+        self.stdin: MockStdinWriter = MockStdinWriter()
+        self.stdout: MockStdoutReader = MockStdoutReader(["this is not valid json\n"])
+        self._returncode: int | None = None
+
+    def wait(self, timeout: float | None = None) -> int:
+        self._returncode = 0
+        return 0
+
+    def terminate(self) -> None:
+        self._returncode = -15
+
+    def kill(self) -> None:
+        self._returncode = -9
+
+
+class MockClosedStdoutProcess:
+    """Mock process that returns empty stdout (simulates closed stdout)."""
+
+    def __init__(self) -> None:
+        self.pid: int = 99997
+        self.stdin: MockStdinWriter = MockStdinWriter()
+        self.stdout: MockStdoutReader = MockStdoutReader([""])
+        self._returncode: int | None = None
+
+    def wait(self, timeout: float | None = None) -> int:
+        self._returncode = 0
+        return 0
+
+    def terminate(self) -> None:
+        self._returncode = -15
+
+    def kill(self) -> None:
+        self._returncode = -9

features/steps/a2a_stdio_transport_steps.py

@@ -0,0 +1,346 @@
+# pyright: reportRedeclaration=false
+"""Step definitions for A2A stdio transport coverage (coverage boost)."""
+
+from __future__ import annotations
+
+import json
+import subprocess
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+from behave import given, then, when
+
+from cleveragents.a2a.models import A2aRequest, A2aResponse
+from cleveragents.a2a.stdio_transport import A2aStdioTransport
+
+# ── Given steps ──────────────────────────────────────────────────────────
+
+
+@given("a new A2aStdioTransport instance")
+def step_new_transport(context: Any) -> None:
+    context.transport = A2aStdioTransport()
+
+
+@given("a connected A2aStdioTransport with a mock process")
+def step_connected_transport(context: Any) -> None:
+    transport = A2aStdioTransport()
+    mock_proc = MagicMock(spec=subprocess.Popen)
+    mock_proc.stdin = MagicMock()
+    mock_proc.stdout = MagicMock()
+    mock_proc.stderr = MagicMock()
+    mock_proc.pid = 12345
+    mock_proc.wait = MagicMock(return_value=0)
+    transport._process = mock_proc
+    transport._is_connected = True
+    context.transport = transport
+    context.mock_process = mock_proc
+
+
+@given("the mock process returns a valid JSON-RPC response")
+def step_mock_valid_response(context: Any) -> None:
+    resp = {"jsonrpc": "2.0", "id": "test-id", "result": {"status": "ok"}}
+    context.mock_process.stdout.readline.return_value = json.dumps(resp) + "\n"
+
+
+@given("the mock process returns invalid JSON")
+def step_mock_invalid_json(context: Any) -> None:
+    context.mock_process.stdout.readline.return_value = "not-json{{"
+
+
+@given("the mock process returns empty response")
+def step_mock_empty_response(context: Any) -> None:
+    context.mock_process.stdout.readline.return_value = ""
+
+
+@given("the mock process has no stdin")
+def step_mock_no_stdin(context: Any) -> None:
+    context.mock_process.stdin = None
+
+
+@given("the mock process has no stdout but valid stdin")
+def step_mock_no_stdout(context: Any) -> None:
+    context.mock_process.stdin = MagicMock()
+    context.mock_process.stdin.write = MagicMock()
+    context.mock_process.stdin.flush = MagicMock()
+    context.mock_process.stdout = None
+
+
+@given("the mock process times out on first wait")
+def step_mock_timeout(context: Any) -> None:
+    context.mock_process.wait.side_effect = [
+        subprocess.TimeoutExpired("test", 5.0),
+        0,
+    ]
+
+
+@given("subprocess Popen is mocked to succeed")
+def step_mock_popen_success(context: Any) -> None:
+    mock_proc = MagicMock(spec=subprocess.Popen)
+    mock_proc.stdin = MagicMock()
+    mock_proc.stdout = MagicMock()
+    mock_proc.stderr = MagicMock()
+    mock_proc.pid = 99999
+    context.popen_mock = mock_proc
+    patcher = patch(
+        "cleveragents.a2a.stdio_transport.subprocess.Popen", return_value=mock_proc
+    )
+    context.popen_patcher = patcher
+    patcher.start()
+
+    def cleanup() -> None:
+        patcher.stop()
+
+    context.add_cleanup(cleanup)
+
+
+@given("subprocess Popen raises FileNotFoundError")
+def step_mock_popen_fnf(context: Any) -> None:
+    patcher = patch(
+        "cleveragents.a2a.stdio_transport.subprocess.Popen",
+        side_effect=FileNotFoundError("No such file"),
+    )
+    context.popen_patcher = patcher
+    patcher.start()
+
+    def cleanup() -> None:
+        patcher.stop()
+
+    context.add_cleanup(cleanup)
+
+
+# ── When steps ───────────────────────────────────────────────────────────
+
+
+def _make_request() -> A2aRequest:
+    return A2aRequest(method="test.echo", params={"msg": "hello"})
+
+
+@when("I try to send a request without connecting")
+def step_send_without_connect(context: Any) -> None:
+    try:
+        context.transport.send(_make_request())
+        context.raised_error = None
+    except (RuntimeError, TypeError) as exc:
+        context.raised_error = exc
+
+
+@when("I try to send a non-A2aRequest object")
+def step_send_non_request(context: Any) -> None:
+    bad_arg: Any = {"not": "a request"}
+    try:
+        context.transport.send(bad_arg)
+        context.raised_error = None
+    except TypeError as exc:
+        context.raised_error = exc
+
+
+@when("I send a valid A2aRequest")
+def step_send_valid_request(context: Any) -> None:
+    context.response = context.transport.send(_make_request())
+
+
+@when("I try to send a valid A2aRequest")
+def step_try_send_valid_request(context: Any) -> None:
+    try:
+        context.transport.send(_make_request())
+        context.raised_error = None
+    except RuntimeError as exc:
+        context.raised_error = exc
+
+
+@when("I try to connect with empty agent path")
+def step_connect_empty(context: Any) -> None:
+    try:
+        context.transport.connect("")
+        context.raised_error = None
+    except ValueError as exc:
+        context.raised_error = exc
+
+
+@when("I try to connect again with a valid path")
+def step_connect_again(context: Any) -> None:
+    try:
+        context.transport.connect("/some/agent")
+        context.raised_error = None
+    except RuntimeError as exc:
+        context.raised_error = exc
+
+
+@when("I call disconnect")
+def step_disconnect(context: Any) -> None:
+    context.transport.disconnect()
+
+
+@when('I connect with agent path "{path}"')
+def step_connect_with_path(context: Any, path: str) -> None:
+    context.transport.connect(path)
+
+
+@when('I try to connect with agent path "{path}"')
+def step_try_connect_with_path(context: Any, path: str) -> None:
+    try:
+        context.transport.connect(path)
+        context.raised_error = None
+    except (RuntimeError, ValueError) as exc:
+        context.raised_error = exc
+
+
+# ── Then steps ───────────────────────────────────────────────────────────
+
+
+@then("the stdio transport should not be connected")
+def step_stdio_not_connected(context: Any) -> None:
+    assert not context.transport.is_connected(), "Expected transport to be disconnected"
+
+
+@then("the stdio transport should be connected")
+def step_stdio_is_connected(context: Any) -> None:
+    assert context.transport.is_connected(), "Expected transport to be connected"
+
+
+@then("the stdio transport process should be None")
+def step_stdio_process_none(context: Any) -> None:
+    assert context.transport.get_process() is None
+
+
+@then("a RuntimeError should be raised about not connected")
+def step_runtime_not_connected(context: Any) -> None:
+    assert isinstance(context.raised_error, RuntimeError)
+    assert "not connected" in str(context.raised_error).lower()
+
+
+@then("a TypeError should be raised about A2aRequest")
+def step_type_error_request(context: Any) -> None:
+    assert isinstance(context.raised_error, TypeError)
+    assert "A2aRequest" in str(context.raised_error)
+
+
+@then("I should receive an A2aResponse")
+def step_received_response(context: Any) -> None:
+    assert isinstance(context.response, A2aResponse)
+
+
+@then("a RuntimeError should be raised about invalid JSON")
+def step_runtime_invalid_json(context: Any) -> None:
+    assert isinstance(context.raised_error, RuntimeError)
+    assert "invalid json" in str(context.raised_error).lower()
+
+
+@then("a RuntimeError should be raised about closed unexpectedly")
+def step_runtime_closed(context: Any) -> None:
+    assert isinstance(context.raised_error, RuntimeError)
+    assert "closed unexpectedly" in str(context.raised_error).lower()
+
+
+@then("a RuntimeError should be raised about stdin")
+def step_runtime_stdin(context: Any) -> None:
+    assert isinstance(context.raised_error, RuntimeError)
+    assert "stdin" in str(context.raised_error).lower()
+
+
+@then("a RuntimeError should be raised about stdout")
+def step_runtime_stdout(context: Any) -> None:
+    assert isinstance(context.raised_error, RuntimeError)
+    assert "stdout" in str(context.raised_error).lower()
+
+
+@then("a ValueError should be raised about agent_path")
+def step_value_error_path(context: Any) -> None:
+    assert isinstance(context.raised_error, ValueError)
+    assert "agent_path" in str(context.raised_error)
+
+
+@then("a RuntimeError should be raised about already connected")
+def step_runtime_already_connected(context: Any) -> None:
+    assert isinstance(context.raised_error, RuntimeError)
+    assert "already connected" in str(context.raised_error).lower()
+
+
+@then("no stdio transport error should be raised")
+def step_no_stdio_error(context: Any) -> None:
+    # Disconnect on unconnected transport is a no-op
+    pass
+
+
+@then("mock stdin close should have been called")
+def step_stdin_closed(context: Any) -> None:
+    context.mock_process.stdin.close.assert_called_once()
+
+
+@then("mock process wait should have been called")
+def step_wait_called(context: Any) -> None:
+    context.mock_process.wait.assert_called()
+
+
+@then("mock process terminate should have been called")
+def step_terminate_called(context: Any) -> None:
+    context.mock_process.terminate.assert_called_once()
+
+
+@then("a RuntimeError should be raised about agent not found")
+def step_runtime_agent_not_found(context: Any) -> None:
+    assert isinstance(context.raised_error, RuntimeError)
+    assert "agent not found" in str(context.raised_error).lower()
+
+
+# ── TransportSelector steps ──────────────────────────────────────────────
+
+from cleveragents.a2a.transport import A2aHttpTransport  # noqa: E402
+from cleveragents.a2a.transport_selector import TransportSelector  # noqa: E402
+
+
+@when("I select a transport with no server URL")
+def step_select_transport_no_url(context: Any) -> None:
+    context.selected_transport = TransportSelector.select()
+
+
+@when("I select a transport with server URL None")
+def step_select_transport_url_none(context: Any) -> None:
+    context.selected_transport = TransportSelector.select(server_url=None)
+
+
+@when('I select a transport with server URL "{url}"')
+def step_select_transport_with_url(context: Any, url: str) -> None:
+    context.selected_transport = TransportSelector.select(
+        server_url=url if url else None
+    )
+
+
+@when("I select a transport with no server URL again")
+def step_select_transport_no_url_again(context: Any) -> None:
+    context.selected_transport_2 = TransportSelector.select()
+
+
+@when('I select a transport with server URL "{url}" again')
+def step_select_transport_with_url_again(context: Any, url: str) -> None:
+    context.selected_transport_2 = TransportSelector.select(
+        server_url=url if url else None
+    )
+
+
+@then("the selected transport should be an A2aStdioTransport")
+def step_selected_is_stdio(context: Any) -> None:
+    assert isinstance(context.selected_transport, A2aStdioTransport), (
+        f"Expected A2aStdioTransport, got {type(context.selected_transport)}"
+    )
+
+
+@then("the selected transport should be an A2aHttpTransport")
+def step_selected_is_http(context: Any) -> None:
+    assert isinstance(context.selected_transport, A2aHttpTransport), (
+        f"Expected A2aHttpTransport, got {type(context.selected_transport)}"
+    )
+
+
+@then("the two stdio transports should be different instances")
+def step_two_stdio_different(context: Any) -> None:
+    assert context.selected_transport is not context.selected_transport_2, (
+        "Expected two different A2aStdioTransport instances"
+    )
+
+
+@then("the two HTTP transports should be different instances")
+def step_two_http_different(context: Any) -> None:
+    assert context.selected_transport is not context.selected_transport_2, (
+        "Expected two different A2aHttpTransport instances"
+    )

robot/a2a_stdio_transport.robot

@@ -0,0 +1,93 @@
+*** Settings ***
+Documentation    Integration tests for A2A stdio transport and TransportSelector.
+...              Verifies subprocess lifecycle management, JSON-RPC 2.0 message
+...              framing, and transport selection logic.
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    ${CURDIR}/helper_a2a_stdio_transport.py
+
+*** Test Cases ***
+A2A Stdio Transport Initial State Is Disconnected
+    [Documentation]    Verify A2aStdioTransport starts in disconnected state
+    ...                with no subprocess handle.
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    initial-state    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    a2a-stdio-initial-state-ok
+
+A2A Stdio Transport Connect Validates Agent Path
+    [Documentation]    Verify A2aStdioTransport.connect() raises ValueError
+    ...                for empty or non-string agent_path.
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    connect-validation    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    a2a-stdio-connect-validation-ok
+
+A2A Stdio Transport Send Requires Connection
+    [Documentation]    Verify A2aStdioTransport.send() raises RuntimeError
+    ...                when not connected to a subprocess.
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    send-not-connected    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    a2a-stdio-send-not-connected-ok
+
+A2A Stdio Transport Send Validates Request Type
+    [Documentation]    Verify A2aStdioTransport.send() raises TypeError
+    ...                for non-A2aRequest arguments.
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    send-type-validation    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    a2a-stdio-send-type-validation-ok
+
+A2A Stdio Transport Disconnect Is Safe When Not Connected
+    [Documentation]    Verify A2aStdioTransport.disconnect() is a no-op
+    ...                when not connected.
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    disconnect-safe    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    a2a-stdio-disconnect-safe-ok
+
+A2A Stdio Transport Subprocess Lifecycle
+    [Documentation]    Verify A2aStdioTransport connects to a subprocess,
+    ...                sends a JSON-RPC 2.0 request, receives a response,
+    ...                and disconnects cleanly.
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    subprocess-lifecycle    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    a2a-stdio-subprocess-lifecycle-ok
+
+A2A Transport Selector Selects Stdio For Local Mode
+    [Documentation]    Verify TransportSelector.select() returns A2aStdioTransport
+    ...                when no server URL is configured.
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    selector-stdio    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    a2a-selector-stdio-ok
+
+A2A Transport Selector Selects HTTP For Server Mode
+    [Documentation]    Verify TransportSelector.select() returns A2aHttpTransport
+    ...                when a server URL is configured.
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    selector-http    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    a2a-selector-http-ok
+
+A2A Transport Selector Returns Fresh Instances
+    [Documentation]    Verify TransportSelector.select() returns a new transport
+    ...                instance on each call.
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    selector-fresh-instances    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    a2a-selector-fresh-instances-ok

robot/helper_a2a_stdio_transport.py

@@ -0,0 +1,223 @@
+"""Helper script for a2a_stdio_transport.robot integration tests.
+
+Each subcommand is a self-contained check that prints a sentinel on success.
+Uses real subprocess communication for lifecycle tests.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+# Ensure local source tree is importable
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+# Path to mock agents
+_MOCKS_DIR = Path(__file__).resolve().parents[1] / "features" / "mocks"
+_ECHO_AGENT = str(_MOCKS_DIR / "a2a_echo_agent.py")
+
+from cleveragents.a2a.models import A2aRequest  # noqa: E402, I001
+from cleveragents.a2a.stdio_transport import A2aStdioTransport  # noqa: E402
+from cleveragents.a2a.transport import A2aHttpTransport  # noqa: E402
+from cleveragents.a2a.transport_selector import TransportSelector  # noqa: E402
+
+
+# ---------------------------------------------------------------------------
+# Subcommands
+# ---------------------------------------------------------------------------
+
+
+def initial_state() -> None:
+    """Verify A2aStdioTransport starts disconnected with no process."""
+    transport = A2aStdioTransport()
+    if transport.is_connected():
+        print("FAIL: transport should start disconnected", file=sys.stderr)
+        sys.exit(1)
+    if transport.get_process() is not None:
+        print("FAIL: process should be None initially", file=sys.stderr)
+        sys.exit(1)
+    print("a2a-stdio-initial-state-ok")
+
+
+def connect_validation() -> None:
+    """Verify connect() raises ValueError for empty/non-string agent_path."""
+    transport = A2aStdioTransport()
+
+    # Test empty string
+    try:
+        transport.connect("")
+        print("FAIL: should have raised ValueError for empty path", file=sys.stderr)
+        sys.exit(1)
+    except ValueError:
+        pass
+
+    # Test None
+    try:
+        transport.connect(None)  # type: ignore[arg-type]
+        print("FAIL: should have raised ValueError for None path", file=sys.stderr)
+        sys.exit(1)
+    except (ValueError, TypeError):
+        pass
+
+    print("a2a-stdio-connect-validation-ok")
+
+
+def send_not_connected() -> None:
+    """Verify send() raises RuntimeError when not connected."""
+    transport = A2aStdioTransport()
+    try:
+        request = A2aRequest(method="test/ping")
+        transport.send(request)
+        print("FAIL: should have raised RuntimeError", file=sys.stderr)
+        sys.exit(1)
+    except RuntimeError:
+        pass
+    print("a2a-stdio-send-not-connected-ok")
+
+
+def send_type_validation() -> None:
+    """Verify send() raises TypeError for non-A2aRequest."""
+    transport = A2aStdioTransport()
+    try:
+        transport.send("not-a-request")  # type: ignore[arg-type]
+        print("FAIL: should have raised TypeError", file=sys.stderr)
+        sys.exit(1)
+    except TypeError:
+        pass
+    print("a2a-stdio-send-type-validation-ok")
+
+
+def disconnect_safe() -> None:
+    """Verify disconnect() is safe when not connected."""
+    transport = A2aStdioTransport()
+    transport.disconnect()  # Should not raise
+    if transport.is_connected():
+        print("FAIL: transport should remain disconnected", file=sys.stderr)
+        sys.exit(1)
+    print("a2a-stdio-disconnect-safe-ok")
+
+
+def subprocess_lifecycle() -> None:
+    """Verify full subprocess lifecycle: connect, send, receive, disconnect."""
+    transport = A2aStdioTransport()
+
+    # Connect to echo agent
+    transport.connect(sys.executable, _ECHO_AGENT)
+    if not transport.is_connected():
+        print("FAIL: transport should be connected after connect()", file=sys.stderr)
+        sys.exit(1)
+    if transport.get_process() is None:
+        print("FAIL: process should not be None after connect()", file=sys.stderr)
+        sys.exit(1)
+
+    # Send a request and receive a response
+    request = A2aRequest(method="test/ping", params={"key": "value"})
+    response = transport.send(request)
+
+    if response.id != request.id:
+        print(
+            f"FAIL: response id '{response.id}' != request id '{request.id}'",
+            file=sys.stderr,
+        )
+        transport.disconnect()
+        sys.exit(1)
+
+    if response.result is None:
+        print(
+            f"FAIL: response result should not be None, error: {response.error}",
+            file=sys.stderr,
+        )
+        transport.disconnect()
+        sys.exit(1)
+
+    # Disconnect
+    transport.disconnect()
+    if transport.is_connected():
+        print(
+            "FAIL: transport should be disconnected after disconnect()",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+
+    print("a2a-stdio-subprocess-lifecycle-ok")
+
+
+def selector_stdio() -> None:
+    """Verify TransportSelector returns A2aStdioTransport for local mode."""
+    transport = TransportSelector.select()
+    if not isinstance(transport, A2aStdioTransport):
+        print(
+            f"FAIL: expected A2aStdioTransport, got {type(transport)}",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+
+    transport_none = TransportSelector.select(server_url=None)
+    if not isinstance(transport_none, A2aStdioTransport):
+        print(
+            f"FAIL: expected A2aStdioTransport for None URL, "
+            f"got {type(transport_none)}",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+
+    print("a2a-selector-stdio-ok")
+
+
+def selector_http() -> None:
+    """Verify TransportSelector returns A2aHttpTransport for server mode."""
+    transport = TransportSelector.select(server_url="http://localhost:8080")
+    if not isinstance(transport, A2aHttpTransport):
+        print(
+            f"FAIL: expected A2aHttpTransport, got {type(transport)}",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+    print("a2a-selector-http-ok")
+
+
+def selector_fresh_instances() -> None:
+    """Verify TransportSelector returns fresh instances on each call."""
+    t1 = TransportSelector.select()
+    t2 = TransportSelector.select()
+    if t1 is t2:
+        print("FAIL: expected different instances for stdio", file=sys.stderr)
+        sys.exit(1)
+
+    h1 = TransportSelector.select(server_url="http://localhost:8080")
+    h2 = TransportSelector.select(server_url="http://localhost:8080")
+    if h1 is h2:
+        print("FAIL: expected different instances for HTTP", file=sys.stderr)
+        sys.exit(1)
+
+    print("a2a-selector-fresh-instances-ok")
+
+
+# ---------------------------------------------------------------------------
+# Dispatch
+# ---------------------------------------------------------------------------
+
+_COMMANDS = {
+    "initial-state": initial_state,
+    "connect-validation": connect_validation,
+    "send-not-connected": send_not_connected,
+    "send-type-validation": send_type_validation,
+    "disconnect-safe": disconnect_safe,
+    "subprocess-lifecycle": subprocess_lifecycle,
+    "selector-stdio": selector_stdio,
+    "selector-http": selector_http,
+    "selector-fresh-instances": selector_fresh_instances,
+}
+
+
+def main() -> None:
+    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
+        print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>", file=sys.stderr)
+        sys.exit(2)
+    _COMMANDS[sys.argv[1]]()
+
+
+if __name__ == "__main__":
+    main()