diff --git a/packages/code-sdk-python/README.md b/packages/code-sdk-python/README.md index 3bad9cbe1..f1aa0a366 100644 --- a/packages/code-sdk-python/README.md +++ b/packages/code-sdk-python/README.md @@ -68,8 +68,10 @@ later = client.resume_thread(thread.id, ThreadOptions(cwd="/path/to/project")) ```python streamed = await thread.run_streamed("Fix the failing test.") async for event in streamed: - if event["type"] == "item.updated" and event["item"]["type"] == "agent_message": - print(event["item"].get("contentDelta", ""), end="") + item = event.get("item") + if item and item["type"] == "agent_message": + # Text deltas arrive on item.started as well as item.updated. + print(item.get("contentDelta", ""), end="") elif event["type"] == "turn.failed": print("failed:", event.get("error")) ``` diff --git a/packages/code-sdk-python/src/minimax_code/errors.py b/packages/code-sdk-python/src/minimax_code/errors.py index 7b034dd0d..d6b27f618 100644 --- a/packages/code-sdk-python/src/minimax_code/errors.py +++ b/packages/code-sdk-python/src/minimax_code/errors.py @@ -42,7 +42,7 @@ def __init__( self.stderr = stderr self.session_id = session_id self.result = result - self.items = list(items) + self.items: list[ThreadItem] = list(items) def __reduce__(self) -> tuple[Any, ...]: # The default reduce calls ``cls(*self.args)``, which breaks subclasses whose diff --git a/packages/code-sdk-python/src/minimax_code/sync.py b/packages/code-sdk-python/src/minimax_code/sync.py index e67adacdb..1ccebe1b4 100644 --- a/packages/code-sdk-python/src/minimax_code/sync.py +++ b/packages/code-sdk-python/src/minimax_code/sync.py @@ -8,13 +8,37 @@ from __future__ import annotations import asyncio -from collections.abc import Iterator -from typing import Any +from collections.abc import Awaitable, Iterator +from typing import Any, TypeVar from .client import MCode from .thread import Thread from .types import MCodeOptions, RunOptions, ThreadEvent, ThreadOptions, Turn +_T = TypeVar("_T") + + +def _in_running_loop() -> bool: + try: + asyncio.get_running_loop() + except RuntimeError: + return False + return True + + +def _run(loop: asyncio.AbstractEventLoop, coro: Awaitable[_T]) -> _T: + """``loop.run_until_complete`` that closes ``coro`` when it cannot run, so a call from inside a + running event loop raises one RuntimeError and leaves no "never awaited" warning.""" + if _in_running_loop(): + close = getattr(coro, "close", None) + if close is not None: + close() + raise RuntimeError( + "The blocking minimax_code API cannot run inside a running event loop; " + "use the async API (MCode) with await." + ) + return loop.run_until_complete(coro) + class SyncThread: def __init__(self, loop: asyncio.AbstractEventLoop, thread: Thread) -> None: @@ -26,20 +50,20 @@ def id(self) -> str | None: return self._thread.id def run(self, input: str, options: RunOptions | None = None) -> Turn: - return self._loop.run_until_complete(self._thread.run(input, options)) + return _run(self._loop, self._thread.run(input, options)) def run_streamed(self, input: str, options: RunOptions | None = None) -> Iterator[ThreadEvent]: """Yields events; closing the iterator early cancels the Turn.""" - streamed = self._loop.run_until_complete(self._thread.run_streamed(input, options)) + streamed = _run(self._loop, self._thread.run_streamed(input, options)) events = streamed.events try: while True: try: - yield self._loop.run_until_complete(events.__anext__()) + yield _run(self._loop, events.__anext__()) except StopAsyncIteration: return finally: - self._loop.run_until_complete(streamed.aclose()) + _run(self._loop, streamed.aclose()) class SyncMCode: @@ -58,11 +82,17 @@ def resume_thread(self, thread_id: str, options: ThreadOptions | None = None) -> return SyncThread(self._loop, self._client.resume_thread(thread_id, options)) def close(self) -> None: - if not self._loop.is_closed(): - # Let subprocess transports finish closing before the loop goes away. - self._loop.run_until_complete(asyncio.sleep(0.05)) - self._loop.run_until_complete(self._loop.shutdown_asyncgens()) + if self._loop.is_closed(): + return + if _in_running_loop(): + # Called from async code (an unsupported use that already raised): the private loop + # never ran a Turn to completion here, so close it without running it. self._loop.close() + return + # Let subprocess transports finish closing before the loop goes away. + self._loop.run_until_complete(asyncio.sleep(0.05)) + self._loop.run_until_complete(self._loop.shutdown_asyncgens()) + self._loop.close() def __enter__(self) -> SyncMCode: return self diff --git a/packages/code-sdk-python/tests/test_sdk.py b/packages/code-sdk-python/tests/test_sdk.py index 6395bba62..e19093046 100644 --- a/packages/code-sdk-python/tests/test_sdk.py +++ b/packages/code-sdk-python/tests/test_sdk.py @@ -363,3 +363,29 @@ async def test_thread_cwd_is_pinned_when_the_thread_is_created( await default.run("a") await relative.run("b") assert [c["args"][4] for c in calls(fake_env)] == [f"--cwd={first}", f"--cwd={first / 'sub'}"] + + +async def test_readme_streaming_loop_keeps_delta_on_item_started(fake_env: dict[str, str]) -> None: + # The CLI can put the first (or only) text delta on item.started; the README loop must keep it. + thread = MCode(MCodeOptions(env=fake_env)).start_thread() + streamed = await thread.run_streamed("x") + text = "" + async for event in streamed: + item = event.get("item") + if item and item["type"] == "agent_message": + text += item.get("contentDelta", "") + assert text == "Hello" + + +async def test_sync_client_inside_running_loop_raises_once(fake_env: dict[str, str]) -> None: + import warnings + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + with pytest.raises(RuntimeError, match="running event loop"): # noqa: SIM117 + with SyncMCode(MCodeOptions(env=fake_env)) as client: + client.start_thread().run("x") + import gc + + gc.collect() + assert not [w for w in caught if issubclass(w.category, RuntimeWarning)]