feat: add initial acp streable http rfc.

[Leoyzen]

No description provided.

[gemini-code-assist]

Code Review

This pull request introduces RFC-0030, which proposes a new StreamableHTTPTransport for ACP agents to align with the Streamable HTTP WebSocket Transport profile using Starlette and uvicorn. The feedback identifies several areas for refinement, including correcting the technical distinction between ASGI servers and frameworks, addressing port inconsistencies with existing transports, and adjusting JSON-RPC error codes to follow idiomatic standards. Additionally, the reviewer requested more detail on how the initialization guard will intercept asynchronous responses to manage connection state.

[Leoyzen]

@gemini-code-assist All four review comments have been addressed in commit ec743ed:

1. Hypercorn terminology ✅ — Changed "another server framework" to "another ASGI server" in Option 1 Disadvantages.

2. Port default inconsistency ✅ — Added documentation in ACPWebSocketTransport docstring explaining that 8080 (standard HTTP port) differs from the legacy WebSocketTransport default 8765, including a migration note for --ws-port users.

3. JSON-RPC error code ✅ — Changed all occurrences of -32600 to -32002 (Success Criteria, Initialize Guard description, Initialize Lifecycle Flow diagram). Added a Rejected Options table in the Decision Record per repository RFC rules, documenting why -32600 was rejected.

4. Initialize Guard implementation ✅ — Added Implementation notes under Initialize Guard clarifying that the guard operates at the protocol layer (on parsed JSON-RPC request/response objects, not raw bytes), avoiding stream adapter complexity. The guard wraps the agent's initialize() handler and flips state on successful InitializeResponse return.

All changes are pushed to this branch. Please re-review when convenient.

[gemini-code-assist]

Code Review

This pull request introduces RFC-0030, which proposes a new ACP Streamable HTTP WebSocket Transport based on Starlette and uvicorn to ensure compliance with the ACP specification. The RFC covers connection identification, lifecycle enforcement via an initialization guard, and a phased implementation plan. Review feedback highlights the need to clarify the handling of JSON-RPC notifications before initialization and to resolve contradictions regarding batch request support. Additionally, a technical improvement was suggested for the shutdown sequence to avoid potential concurrency issues when iterating over active connections.

[gemini-code-assist]

The RFC defines the behavior for requests received before initialization but does not specify how notifications should be handled. Since JSON-RPC notifications do not have an id and do not receive a response, the server cannot return the -32002 error code. The RFC should clarify whether such notifications are silently ignored, logged, or if they trigger a connection closure due to a protocol violation.

[gemini-code-assist]

The description of batch request handling is contradictory. If Phase 1 does not support JSON-RPC batching, the RFC should explicitly state that any batch request will be rejected (e.g., with an Invalid Request or Method not found error). The phrase "unless all elements are initialize" implies a level of batch support that conflicts with the statement that batching is not supported. If this design option is being rejected, it should be explicitly marked as 'Rejected' with a reference to the decision to preserve historical context, as per repository rules.

References

1. When a design option discussed in an RFC is rejected, explicitly mark it as 'Rejected' with a reference to the decision, rather than removing it, to preserve historical context for future readers.

[gemini-code-assist]

Iterating directly over _active_connections while awaiting conn.close() is unsafe if the set can be modified concurrently. Since the WebSocket endpoint handlers (which are responsible for removing connections from this set) run in separate tasks that may be finishing during uvicorn's shutdown, this loop could encounter a RuntimeError: Set size changed during iteration. It is safer to iterate over a copy of the set.

Suggested change

	for conn in _active_connections:
	for conn in list(_active_connections):

[Leoyzen]

This is already addressed in the implementation. The cleanup loop uses list(active_connections) to iterate over a snapshot:

for conn in list(active_connections):
    await conn.close()

This prevents RuntimeError: Set size changed during iteration by copying the set before iteration. The finally block in handle_acp uses active_connections.discard(conn) which is safe even if the connection was never added.

[Leoyzen]

/gemini review

[gemini-code-assist]

Code Review

This pull request implements Phase 1 of the ACP Streamable HTTP WebSocket Transport (RFC-0030), introducing a Starlette-based WebSocket server at /acp. Key changes include the addition of an initialization guard to enforce the protocol handshake, CLI and YAML configuration updates, and the deprecation of the legacy WebSocket transport. Feedback focused on a violation of the ByteReceiveStream contract in the new stream adapters regarding max_bytes handling and buffering. Additionally, recommendations were made to improve type safety, respect encapsulation by avoiding private internal state, and refine message decoding to prevent aggressive whitespace stripping.

[gemini-code-assist]

The _StarletteWebSocketReadStream.receive implementation ignores the max_bytes parameter and lacks buffering. This violates the ByteReceiveStream interface contract, which requires receiving at most max_bytes. If the WebSocket frame contains more data than requested, the excess must be buffered for subsequent calls. This inconsistency could lead to buffer overflows or logic errors in protocol parsers that expect the limit to be respected.

class _StarletteWebSocketReadStream(ByteReceiveStream):
    """Adapter to read from Starlette WebSocket as a ByteReceiveStream."""

    def __init__(self, websocket: Any) -> None:
        self._websocket = websocket
        self._buffer = b""

    async def receive(self, max_bytes: int = 65536) -> bytes:
        if self._buffer:
            data = self._buffer[:max_bytes]
            self._buffer = self._buffer[max_bytes:]
            return data

        try:
            message: str = await self._websocket.receive_text()
        except Exception as e:
            raise anyio.EndOfStream from e

        data = message.encode()
        if not data.endswith(b"\n"):
            data += b"\n"

        if len(data) > max_bytes:
            self._buffer = data[max_bytes:]
            return data[:max_bytes]

        return data

    async def aclose(self) -> None:
        pass

[Leoyzen]

Fixed in commit f2feb4a.

Changes made:

• Added _buffer attribute to _StarletteWebSocketReadStream to cache excess bytes
• receive() now checks buffer first before calling receive_text()
• Excess data beyond max_bytes is stored in buffer for subsequent calls
• Added unit test test_starlette_read_stream_buffers_across_receives to verify behavior

This respects the ByteReceiveStream interface contract while maintaining compatibility with the JSON-RPC line protocol.

[gemini-code-assist]

The websocket parameter should be type-hinted as starlette.websockets.WebSocket instead of Any to improve type safety and enable better IDE support.

Suggested change

	async def handle_acp(websocket: Any) -> None:
	async def handle_acp(websocket: starlette.websockets.WebSocket) -> None:

[Leoyzen]

This is an intentional design choice due to optional dependency constraints.

Constraint: starlette is an optional dependency (only required when using --transport streamable-http). If we import WebSocket at the module level for type hints, users running stdio-only transports would receive ModuleNotFoundError on import.

Current approach: The websocket parameter is typed as Any with the actual type enforced at runtime via duck-typing (Starlette's WebSocket API). Starlette imports are deferred to function scope inside _serve_streamable_http().

Alternative considered: Using TYPE_CHECKING guard:

if TYPE_CHECKING:
    from starlette.websockets import WebSocket

This would improve IDE support without runtime impact. However, since handle_acp is a nested closure function inside _serve_streamable_http(), the type annotation would need to reference the import in the outer scope, which adds complexity for minimal gain.

Recommendation: Keep Any for now. If Starlette becomes a required dependency in the future (e.g., when Streamable HTTP Phase 2 lands), we can migrate to proper type annotations.

[gemini-code-assist]

Accessing conn._conn._recv_task violates encapsulation by reaching into private implementation details of the acp library's Connection class. This makes the transport implementation brittle and dependent on internal state that might change in future versions. Consider adding a public method to AgentSideConnection or Connection to wait for the connection lifecycle to end.

[Leoyzen]

Acknowledged. This is a technical debt item constrained by the acp library architecture.

Constraint: Connection (from acp.connection) is an external library class, not under our control. It does not expose a public API to wait for the receive loop lifecycle. The _recv_task attribute is the only mechanism available to detect when the connection's read loop has terminated.

Mitigation: The access is isolated to a single line in _serve_streamable_http() and is wrapped with isinstance(recv_task, asyncio.Task) and not recv_task.done() guards. If the acp library changes this internal structure in a future version, the isinstance check will fail gracefully (the connection will still close via the finally block).

Long-term: This should be addressed upstream in the acp library by adding a public wait_closed() or lifecycle_task property to Connection. I've noted this as a follow-up item for the acp library maintainers.