Concurrent chat() calls on same Agent crash or silently lose data

[jiridanek]

Calling agent.chat() concurrently on the same Agent instance produces three failure modes depending on timing: a RuntimeError crash, garbled responses, or silent data loss. None of these are documented.

Failure modes

Scenario	Behavior
Sequential (A finishes, then B)	Both correct
Overlapping (B starts before A consumed)	A's response silently lost (empty string), B works
Concurrent (asyncio.gather)	One gets garbled data (merged responses), the other crashes with RuntimeError

Reproduction

import asyncio
from google.antigravity import Agent, LocalAgentConfig, types
from google.antigravity.hooks import hooks, policy

@hooks.pre_tool_call_decide
async def allow_all(data: types.ToolCall) -> types.HookResult:
    return types.HookResult(allow=True)

async def chat_task(agent, label, prompt):
    print(f"  [{label}] sending...")
    response = await agent.chat(prompt)
    text = await response.text()
    print(f"  [{label}] got: {text.strip()!r}")
    return text.strip()

async def main():
    config = LocalAgentConfig(
        model="gemini-2.5-flash",
        policies=[policy.allow_all()],
        hooks=[allow_all],
    )

    async with Agent(config) as agent:
        # Test 1: Sequential — works
        print("\n=== Test 1: Sequential ===")
        r1 = await chat_task(agent, "A", "What is 2+2? Reply with just the number.")
        r2 = await chat_task(agent, "B", "What is 3+3? Reply with just the number.")
        print(f"  Results: A={r1!r}, B={r2!r}")

        # Test 2: Overlapping — silent data loss
        print("\n=== Test 2: Overlapping ===")
        resp_a = await agent.chat("What is 4+4? Reply with just the number.")
        resp_b = await agent.chat("What is 5+5? Reply with just the number.")
        text_b = await resp_b.text()
        text_a = await resp_a.text()
        print(f"  Results: A={text_a.strip()!r}, B={text_b.strip()!r}")
        # A is empty, B is correct

        # Test 3: Concurrent — crash
        print("\n=== Test 3: Concurrent ===")
        results = await asyncio.gather(
            chat_task(agent, "X", "What is 6+6? Reply with just the number."),
            chat_task(agent, "Y", "What is 7+7? Reply with just the number."),
            return_exceptions=True,
        )
        for i, r in enumerate(results):
            label = ["X", "Y"][i]
            if isinstance(r, Exception):
                print(f"  [{label}] FAILED: {type(r).__name__}: {r}")
            else:
                print(f"  [{label}] result: {r!r}")

asyncio.run(main())

Output:

=== Test 1: Sequential ===
  [A] got: '4'
  [B] got: '6'
  Results: A='4', B='6'

=== Test 2: Overlapping ===
  Results: A='', B='10'

=== Test 3: Concurrent ===
  [X] got: '1214'
  [X] result: '1214'
  [Y] FAILED: RuntimeError: Concurrent receive_steps() calls are not supported on this connection.

Root cause

LocalConnection.receive_steps() (local_connection.py:574) sets an _is_receiving flag and raises RuntimeError if a second caller enters while the first is active:

if self._is_receiving:
    raise RuntimeError("Concurrent receive_steps() calls are not supported on this connection.")

Conversation.send() (conversation.py:135) tries to handle this — when the connection isn't idle, it attempts to drain remaining steps, catching RuntimeError and falling back to wait_for_idle(). But this only helps in the sequential case. When two coroutines hold separate ChatResponse objects that are actively iterating chunks, both are calling receive_steps() through their own receive_chunks() generators. The send() drain can't help because neither coroutine has exited its iterator.

The overlapping case (test 2) silently drains A's remaining steps during B's send(), so A's response.text() returns empty. The concurrent case (test 3) hits the _is_receiving guard directly.

Use case

The ACP protocol supports multiple concurrent sessions on the same agent server. IDE clients like Zed allow users to open multiple chat sessions simultaneously. Each session calls prompt() which calls agent.chat(). With a single Agent instance, these overlap.

Currently there's no documented way to create multiple independent conversations on one agent, nor a way to create multiple Agent instances sharing the same harness process. Each Agent() spawns its own Go harness subprocess.

Request

Either:

1. Document the limitation — "only one chat() call may be active at a time; fully consume or cancel the response before starting another"
2. Support multiple conversations — allow creating separate Conversation instances on the same connection, each with their own step stream
3. Raise earlier and louder — the overlapping case (test 2) silently loses data with no error. At minimum, chat() should raise if a previous response hasn't been consumed.

Environment

• SDK: google-antigravity (latest from PyPI as of 2026-06-15)
• Model: gemini-2.5-flash
• Platform: macOS 26, Python 3.14