feat(sdk): add true streaming to Node.js and Python SDKs

Node.js:
- Replace broken Vec<AgentEvent> stream() with EventStream struct
- Add EventStream.next() → Promise<{ value, done }>
- Add Symbol.asyncIterator in index.js for `for await...of` support
- Export EventStream from module

Python:
- Change done: bool → Arc<AtomicBool> in PyEventStream
- Add __aiter__ and __anext__ via run_in_executor bridge
- Add BlockingRecv callable for thread-pool async integration
- Both sync `for event in stream` and async `async for event in stream` work

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>

Author: ZhiXiao-Lin

sdk/node/index.js

@@ -310,8 +310,22 @@ if (!nativeBinding) {
   throw new Error(`Failed to load native binding`)
 }
 
-const { Agent, Session, builtinSkills } = nativeBinding
+const { Agent, Session, EventStream, builtinSkills } = nativeBinding
+
+// Add Symbol.asyncIterator to EventStream for `for await...of` support.
+// Each iteration calls .next() until done === true.
+EventStream.prototype[Symbol.asyncIterator] = async function* () {
+  while (true) {
+    const { value, done } = await this.next()
+    if (done) {
+      if (value != null) yield value
+      return
+    }
+    yield value
+  }
+}
 
 module.exports.Agent = Agent
 module.exports.Session = Session
+module.exports.EventStream = EventStream
 module.exports.builtinSkills = builtinSkills

sdk/node/src/lib.rs

@@ -34,7 +34,10 @@ use a3s_code_core::{
 use a3s_code_core::{
     Agent as RustAgent, AgentSession as RustAgentSession, SessionOptions as RustSessionOptions,
 };
-use std::sync::Arc;
+use std::sync::{
+    atomic::{AtomicBool, Ordering},
+    Arc,
+};
 use tokio::runtime::Runtime;
 
 // ============================================================================
@@ -179,6 +182,65 @@ pub struct ToolResult {
     pub exit_code: i32,
 }
 
+// ============================================================================
+// EventStream
+// ============================================================================
+
+/// Result of a single `EventStream.next()` call.
+#[napi(object)]
+#[derive(Clone)]
+pub struct NextResult {
+    pub value: Option<AgentEvent>,
+    pub done: bool,
+}
+
+/// Streaming event iterator. Use `for await (const event of stream)` or call `.next()` manually.
+#[napi]
+pub struct EventStream {
+    rx: Arc<tokio::sync::Mutex<tokio::sync::mpsc::Receiver<RustAgentEvent>>>,
+    done: Arc<AtomicBool>,
+}
+
+#[napi]
+impl EventStream {
+    /// Get the next event from the stream.
+    ///
+    /// Returns `{ value: AgentEvent | null, done: boolean }`.
+    /// When `done` is true, the stream is exhausted.
+    #[napi]
+    pub async fn next(&self) -> napi::Result<NextResult> {
+        if self.done.load(Ordering::Relaxed) {
+            return Ok(NextResult { value: None, done: true });
+        }
+        let rx = self.rx.clone();
+        let done_flag = self.done.clone();
+        let result = get_runtime()
+            .spawn(async move {
+                let mut guard = rx.lock().await;
+                guard.recv().await
+            })
+            .await
+            .map_err(|e| napi::Error::from_reason(format!("Task join error: {e}")))?;
+        match result {
+            Some(event) => {
+                let is_end = matches!(event, RustAgentEvent::End { .. });
+                let is_error = matches!(event, RustAgentEvent::Error { .. });
+                let js_event = AgentEvent::from(event);
+                if is_end || is_error {
+                    done_flag.store(true, Ordering::Relaxed);
+                    Ok(NextResult { value: Some(js_event), done: true })
+                } else {
+                    Ok(NextResult { value: Some(js_event), done: false })
+                }
+            }
+            None => {
+                done_flag.store(true, Ordering::Relaxed);
+                Ok(NextResult { value: None, done: true })
+            }
+        }
+    }
+}
+
 // ============================================================================
 // SessionOptions
 // ============================================================================
@@ -447,7 +509,9 @@ impl Session {
         Ok(AgentResult::from(result))
     }
 
-    /// Send a prompt and get a stream of events.
+    /// Send a prompt and get a streaming event iterator.
+    ///
+    /// Returns an `EventStream`. Use `for await (const event of stream)` or call `.next()` manually.
     ///
     /// @param prompt - The prompt to send
     /// @param history - Optional conversation history
@@ -456,7 +520,7 @@ impl Session {
         &self,
         prompt: String,
         history: Option<Vec<MessageObject>>,
-    ) -> napi::Result<Vec<AgentEvent>> {
+    ) -> napi::Result<EventStream> {
         let rust_history = history
             .map(|h| js_messages_to_rust(&h))
             .transpose()?;
@@ -469,27 +533,10 @@ impl Session {
             .map_err(|e| napi::Error::from_reason(format!("Task join error: {e}")))?
             .map_err(|e| napi::Error::from_reason(format!("Failed to start stream: {e}")))?;
 
-        let rx_arc: Arc<tokio::sync::Mutex<tokio::sync::mpsc::Receiver<RustAgentEvent>>> =
-            Arc::new(tokio::sync::Mutex::new(rx));
-
-        let collected = get_runtime()
-            .spawn(async move {
-                let mut guard = rx_arc.lock().await;
-                let mut events: Vec<AgentEvent> = Vec::new();
-                while let Some(event) = guard.recv().await {
-                    let is_end = matches!(event, RustAgentEvent::End { .. });
-                    let is_error = matches!(event, RustAgentEvent::Error { .. });
-                    events.push(AgentEvent::from(event));
-                    if is_end || is_error {
-                        break;
-                    }
-                }
-                events
-            })
-            .await
-            .map_err(|e| napi::Error::from_reason(format!("Task join error: {e}")))?;
-
-        Ok(collected)
+        Ok(EventStream {
+            rx: Arc::new(tokio::sync::Mutex::new(rx)),
+            done: Arc::new(AtomicBool::new(false)),
+        })
     }
 
     /// Return the session's conversation history.

sdk/python/src/lib.rs

@@ -31,10 +31,13 @@ use a3s_code_core::{
 use a3s_code_core::{
     Agent as RustAgent, AgentSession as RustAgentSession, SessionOptions as RustSessionOptions,
 };
-use pyo3::exceptions::{PyRuntimeError, PyStopIteration, PyTypeError, PyValueError};
+use pyo3::exceptions::{PyRuntimeError, PyStopAsyncIteration, PyStopIteration, PyTypeError, PyValueError};
 use pyo3::prelude::*;
 use pyo3::types::{PyDict, PyList};
-use std::sync::Arc;
+use std::sync::{
+    atomic::{AtomicBool, Ordering},
+    Arc,
+};
 use tokio::runtime::Runtime;
 use tokio::sync::Mutex;
 
@@ -261,32 +264,79 @@ impl PyToolResult {
 }
 
 // ============================================================================
-// EventStream (Python Iterator)
+// EventStream (Python Iterator + Async Iterator)
 // ============================================================================
 
-/// Iterator that yields AgentEvents from a streaming execution.
+/// One-shot callable used by `run_in_executor` for async iteration.
+///
+/// Each `__anext__` call creates a new instance; `__call__` blocks on the
+/// next channel receive and raises `StopAsyncIteration` when done.
+#[pyclass]
+struct BlockingRecv {
+    rx: Arc<Mutex<tokio::sync::mpsc::Receiver<RustAgentEvent>>>,
+    done: Arc<AtomicBool>,
+}
+
+#[pymethods]
+impl BlockingRecv {
+    fn __call__(&self, py: Python<'_>) -> PyResult<PyAgentEvent> {
+        let rx = self.rx.clone();
+        let done_flag = self.done.clone();
+        let result = py.allow_threads(|| {
+            get_runtime().block_on(async {
+                let mut guard = rx.lock().await;
+                guard.recv().await
+            })
+        });
+        match result {
+            Some(event) => {
+                let is_end = matches!(event, RustAgentEvent::End { .. });
+                let is_error = matches!(event, RustAgentEvent::Error { .. });
+                let py_event = PyAgentEvent::from(event);
+                if is_end || is_error {
+                    done_flag.store(true, Ordering::Relaxed);
+                }
+                Ok(py_event)
+            }
+            None => {
+                done_flag.store(true, Ordering::Relaxed);
+                Err(PyStopAsyncIteration::new_err("stream exhausted"))
+            }
+        }
+    }
+}
+
+/// Iterator / async-iterator that yields AgentEvents from a streaming execution.
+///
+/// Sync usage:  `for event in session.stream(prompt):`
+/// Async usage: `async for event in session.stream(prompt):`
 #[pyclass(name = "EventStream")]
 struct PyEventStream {
     rx: Arc<Mutex<tokio::sync::mpsc::Receiver<RustAgentEvent>>>,
-    done: bool,
+    done: Arc<AtomicBool>,
 }
 
 #[pymethods]
 impl PyEventStream {
+    // ------------------------------------------------------------------
+    // Sync iterator protocol
+    // ------------------------------------------------------------------
+
     fn __iter__(slf: PyRef<'_, Self>) -> PyRef<'_, Self> {
         slf
     }
 
     fn __next__(&mut self, py: Python<'_>) -> PyResult<Option<PyAgentEvent>> {
-        if self.done {
+        if self.done.load(Ordering::Relaxed) {
             return Err(PyStopIteration::new_err("stream exhausted"));
         }
 
         let rx = self.rx.clone();
+        let done_flag = self.done.clone();
         let result = py.allow_threads(|| {
             get_runtime().block_on(async {
-                let mut rx = rx.lock().await;
-                rx.recv().await
+                let mut guard = rx.lock().await;
+                guard.recv().await
             })
         });
 
@@ -296,16 +346,47 @@ impl PyEventStream {
                 let is_error = matches!(event, RustAgentEvent::Error { .. });
                 let py_event = PyAgentEvent::from(event);
                 if is_end || is_error {
-                    self.done = true;
+                    done_flag.store(true, Ordering::Relaxed);
                 }
                 Ok(Some(py_event))
             }
             None => {
-                self.done = true;
+                done_flag.store(true, Ordering::Relaxed);
                 Err(PyStopIteration::new_err("stream exhausted"))
             }
         }
     }
+
+    // ------------------------------------------------------------------
+    // Async iterator protocol
+    // ------------------------------------------------------------------
+
+    fn __aiter__(slf: PyRef<'_, Self>) -> PyRef<'_, Self> {
+        slf
+    }
+
+    /// Returns an `asyncio.Future` that resolves to the next `AgentEvent`.
+    ///
+    /// Uses `run_in_executor` to bridge the blocking channel recv into an
+    /// asyncio-compatible awaitable without requiring `pyo3-async`.
+    fn __anext__<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
+        if self.done.load(Ordering::Relaxed) {
+            return Err(PyStopAsyncIteration::new_err("stream exhausted"));
+        }
+
+        let callable = Bound::new(
+            py,
+            BlockingRecv {
+                rx: self.rx.clone(),
+                done: self.done.clone(),
+            },
+        )?;
+
+        let asyncio = py.import("asyncio")?;
+        let loop_ = asyncio.call_method0("get_running_loop")?;
+        let future = loop_.call_method1("run_in_executor", (py.None(), callable))?;
+        Ok(future)
+    }
 }
 
 // ============================================================================
@@ -491,7 +572,7 @@ impl PySession {
 
         Ok(PyEventStream {
             rx: Arc::new(Mutex::new(rx)),
-            done: false,
+            done: Arc::new(AtomicBool::new(false)),
         })
     }