feat(python-sdk): expose subagent task query API

sdk/python/src/lib.rs

@@ -1535,6 +1535,40 @@ impl PySession {
         json_string_to_py(py, &json)
     }
 
+    /// Look up a delegated subagent task by id. Returns None when no such
+    /// task has been observed in this session.
+    fn subagent_task(&self, py: Python<'_>, task_id: String) -> PyResult<PyObject> {
+        let session = self.inner.clone();
+        let snapshot =
+            py.allow_threads(move || get_runtime().block_on(session.subagent_task(&task_id)));
+        let json = serde_json::to_string(&snapshot).map_err(|e| {
+            PyRuntimeError::new_err(format!("Failed to serialize subagent task: {e}"))
+        })?;
+        json_string_to_py(py, &json)
+    }
+
+    /// Return snapshots of every delegated subagent task observed in this
+    /// session (including completed and failed ones), oldest first.
+    fn subagent_tasks(&self, py: Python<'_>) -> PyResult<PyObject> {
+        let session = self.inner.clone();
+        let tasks = py.allow_threads(move || get_runtime().block_on(session.subagent_tasks()));
+        let json = serde_json::to_string(&tasks).map_err(|e| {
+            PyRuntimeError::new_err(format!("Failed to serialize subagent tasks: {e}"))
+        })?;
+        json_string_to_py(py, &json)
+    }
+
+    /// Return snapshots of subagent tasks still in `running` state.
+    fn pending_subagent_tasks(&self, py: Python<'_>) -> PyResult<PyObject> {
+        let session = self.inner.clone();
+        let tasks =
+            py.allow_threads(move || get_runtime().block_on(session.pending_subagent_tasks()));
+        let json = serde_json::to_string(&tasks).map_err(|e| {
+            PyRuntimeError::new_err(format!("Failed to serialize pending subagent tasks: {e}"))
+        })?;
+        json_string_to_py(py, &json)
+    }
+
     /// Cancel a specific run only if it is still the active run.
     fn cancel_run(&self, py: Python<'_>, run_id: String) -> bool {
         let session = self.inner.clone();

sdk/python/tests/test_subagent_query_api.py

@@ -0,0 +1,59 @@
+"""Smoke test for the subagent task query API exposed in PR #4.
+
+Verifies the three new Session methods are reachable from Python and
+return the expected empty-state shapes for a fresh session. Mirrors the
+Node SDK smoke test in sdk/node/test.mjs.
+
+Run with: A3S_CONFIG_FILE not needed — uses inline ACL.
+"""
+
+from __future__ import annotations
+
+import tempfile
+
+from a3s_code import Agent, LocalWorkspaceBackend, PermissionPolicy, SessionOptions
+
+
+INLINE_CONFIG = """
+default_model = "anthropic/claude-sonnet-4-20250514"
+
+providers "anthropic" {
+  api_key = "test-key"
+  models "claude-sonnet-4-20250514" {
+    name = "Claude Sonnet 4"
+  }
+}
+""".strip()
+
+
+def main() -> None:
+    workspace = tempfile.mkdtemp(prefix="a3s-code-python-subagent-")
+    agent = Agent.create(INLINE_CONFIG)
+
+    opts = SessionOptions()
+    opts.permission_policy = PermissionPolicy(default_decision="allow")
+    opts.workspace_backend = LocalWorkspaceBackend(workspace)
+
+    session = agent.session(workspace, opts)
+
+    tasks = session.subagent_tasks()
+    assert isinstance(tasks, list), f"subagent_tasks() should return list, got {type(tasks)!r}"
+    assert tasks == [], f"fresh session should have no subagent tasks, got {tasks!r}"
+
+    pending = session.pending_subagent_tasks()
+    assert isinstance(
+        pending, list
+    ), f"pending_subagent_tasks() should return list, got {type(pending)!r}"
+    assert (
+        pending == []
+    ), f"fresh session should have no pending subagent tasks, got {pending!r}"
+
+    missing = session.subagent_task("task-does-not-exist")
+    assert missing is None, f"unknown subagent task id should return None, got {missing!r}"
+
+    session.close()
+    print("python sdk subagent query api ok")
+
+
+if __name__ == "__main__":
+    main()