Competitor-Informed Bug Fixes Implementation Plan¶

# Append to tests/agent/test_regression.py (BUG-01)
"""BUG-01: Streaming run()/arun() silently drops ToolCall objects.

Source: Agno #6757 pattern — competitor bug where tool function names
become empty strings in streaming responses.

Selectools variant: _streaming_call and _astreaming_call filter chunks
with `isinstance(chunk, str)` which drops ToolCall objects entirely.
Tools are never executed when AgentConfig(stream=True).
"""
from __future__ import annotations

from typing import Any, Iterator

import pytest

from selectools import Agent, AgentConfig, Tool, ToolParameter
from selectools.providers.stubs import LocalProvider
from selectools.types import Message, Role, ToolCall


class StreamingToolProvider(LocalProvider):
    """Provider that yields a ToolCall during streaming."""

    name = "streaming_tool_stub"
    supports_streaming = True
    supports_async = False

    def __init__(self) -> None:
        super().__init__()
        self.call_count = 0

    def stream(self, **kwargs: Any) -> Iterator[Any]:
        self.call_count += 1
        if self.call_count == 1:
            yield "I will call a tool. "
            yield ToolCall(tool_name="echo", parameters={"text": "hello"})
        else:
            yield "Done. Got: hello"


def _echo_fn(text: str) -> str:
    return text


def test_streaming_preserves_tool_calls():
    """When stream=True, ToolCall objects from the provider must be executed."""
    echo_tool = Tool(
        name="echo",
        description="Echo text",
        parameters=[ToolParameter(name="text", param_type=str, description="t", required=True)],
        function=_echo_fn,
    )
    provider = StreamingToolProvider()
    agent = Agent(
        tools=[echo_tool],
        provider=provider,
        config=AgentConfig(stream=True, max_iterations=3),
    )
    result = agent.run([Message(role=Role.USER, content="echo hello")])
    assert "Done" in result.content, f"Expected tool to execute; got: {result.content!r}"
    assert provider.call_count >= 2, "Agent should have looped after tool execution"

    def _streaming_call(
        self, stream_handler: Optional[Callable[[str], None]] = None
    ) -> Tuple[str, List["ToolCall"]]:
        if not getattr(self.provider, "supports_streaming", False):
            raise ProviderError(f"Provider {self.provider.name} does not support streaming.")

        aggregated: List[str] = []
        tool_calls: List["ToolCall"] = []
        for chunk in self.provider.stream(
            model=self._effective_model,
            system_prompt=self._system_prompt,
            messages=self._history,
            tools=self.tools,
            temperature=self.config.temperature,
            max_tokens=self.config.max_tokens,
            timeout=self.config.request_timeout,
        ):
            if isinstance(chunk, str):
                if chunk:
                    aggregated.append(chunk)
                    if stream_handler:
                        stream_handler(chunk)
            elif isinstance(chunk, ToolCall):
                tool_calls.append(chunk)

        return "".join(aggregated), tool_calls

response_text, streamed_tool_calls = self._streaming_call(stream_handler)
return Message(
    role=Role.ASSISTANT,
    content=response_text,
    tool_calls=streamed_tool_calls or None,
)

# Append to tests/agent/test_regression.py (BUG-02)
"""BUG-02: typing.Literal crashes @tool() creation.

Source: Agno #6720 — get_json_schema_for_arg() does not handle
typing.Literal, producing {"type": "object"} instead of
{"type": "string", "enum": [...]}.

Selectools variant: _unwrap_type() returns Literal unchanged, then
_validate_tool_definition() rejects it as an unsupported type.
"""
from __future__ import annotations

from typing import Literal, Optional

from selectools.tools import tool


def test_literal_str_produces_enum():
    @tool()
    def set_mode(mode: Literal["fast", "slow", "auto"]) -> str:
        return f"mode={mode}"

    assert set_mode.name == "set_mode"
    params = {p.name: p for p in set_mode.parameters}
    assert "mode" in params
    assert params["mode"].enum == ["fast", "slow", "auto"]
    assert params["mode"].param_type is str


def test_literal_int_produces_enum():
    @tool()
    def set_level(level: Literal[1, 2, 3]) -> str:
        return f"level={level}"

    params = {p.name: p for p in set_level.parameters}
    assert params["level"].enum == [1, 2, 3]
    assert params["level"].param_type is int


def test_optional_literal_works():
    @tool()
    def filter_by(tag: Optional[Literal["red", "blue"]] = None) -> str:
        return f"tag={tag}"

    params = {p.name: p for p in filter_by.parameters}
    assert params["tag"].enum == ["red", "blue"]
    assert params["tag"].required is False

from typing import Any, Callable, Dict, List, Literal, Optional, Tuple, Union, get_args, get_origin, get_type_hints

def _literal_info(type_hint: Any) -> Optional[Tuple[Any, List[Any]]]:
    """Return (base_type, enum_values) for Literal[...] hints, else None.

    Unwraps Optional[Literal[...]] as well. Base type is inferred from the
    first literal value (e.g. Literal["a", "b"] → str).
    """
    origin = get_origin(type_hint)
    if origin is Union:
        args = get_args(type_hint)
        non_none = [a for a in args if a is not type(None)]
        if len(non_none) == 1:
            return _literal_info(non_none[0])
    if sys.version_info >= (3, 10):
        import types as _types  # noqa: PLC0415
        if isinstance(type_hint, _types.UnionType):
            args = get_args(type_hint)
            non_none = [a for a in args if a is not type(None)]
            if len(non_none) == 1:
                return _literal_info(non_none[0])
    if origin is Literal:
        values = list(get_args(type_hint))
        if not values:
            return None
        base_type = type(values[0])
        return base_type, values
    return None

        meta = param_metadata.get(name, {})
        description = meta.get("description", f"Parameter {name}")
        enum_values: Optional[List[Any]] = meta.get("enum")

        raw_type = type_hints.get(name, str)
        lit = _literal_info(raw_type)
        if lit is not None:
            param_type, literal_values = lit
            if enum_values is None:
                enum_values = literal_values
        else:
            param_type = _unwrap_type(raw_type)

        is_optional = param.default != inspect.Parameter.empty

# Append to tests/agent/test_regression.py (BUG-03)
"""BUG-03: asyncio.run() in sync wrappers crashes inside running event loops.

Source: PraisonAI #1165 — asyncio.run() called from sync context reachable
by async callers crashes with "cannot call asyncio.run() while an event
loop is running". Reachable from Jupyter, FastAPI handlers, async tests.
"""
from __future__ import annotations

import asyncio

import pytest

from selectools._async_utils import run_sync


def test_run_sync_outside_event_loop():
    async def coro():
        return 42

    assert run_sync(coro()) == 42


def test_run_sync_inside_running_loop():
    """Key test — calling run_sync from within an async function."""
    async def outer():
        async def inner():
            return "hello"
        return run_sync(inner())

    result = asyncio.run(outer())
    assert result == "hello"

# src/selectools/_async_utils.py
"""Safe synchronous-wrapper utilities for async code.

Calling asyncio.run() from a sync function that is itself reachable
from an async caller raises RuntimeError: asyncio.run() cannot be called
when another event loop is running. This module provides a helper that
detects the surrounding event loop and executes the coroutine on a fresh
loop in a dedicated thread when one is already running.
"""

from __future__ import annotations

import asyncio
import concurrent.futures
from typing import Any, Awaitable, TypeVar

T = TypeVar("T")


def run_sync(coro: Awaitable[T]) -> T:
    """Run a coroutine to completion from sync code.

    If no event loop is running in the current thread, uses asyncio.run.
    If one is running, spawns a worker thread, creates a fresh event loop
    there, and waits for the result. Safe to call from Jupyter notebooks,
    FastAPI handlers, async tests, and nested orchestration.
    """
    try:
        asyncio.get_running_loop()
    except RuntimeError:
        return asyncio.run(coro)  # type: ignore[arg-type]

    def _runner() -> T:
        return asyncio.run(coro)  # type: ignore[arg-type]

    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
        future = executor.submit(_runner)
        return future.result()

        return asyncio.run(
            self.arun(
                prompt_or_state,
                checkpoint_store=checkpoint_store,
                checkpoint_id=checkpoint_id,
            )
        )

        from .._async_utils import run_sync

        return run_sync(
            self.arun(
                prompt_or_state,
                checkpoint_store=checkpoint_store,
                checkpoint_id=checkpoint_id,
            )
        )

# Append to tests/agent/test_regression.py (BUG-04)
"""BUG-04: InterruptRequest from a child node in a parallel group is silently
dropped. The parent graph treats the child as completed.

Source: Agno #4921 — NoneType does not have run_id error when running HITL
within agent tools which are part of team.
"""
from __future__ import annotations

from selectools import AgentGraph
from selectools.orchestration import (
    GraphNode,
    GraphState,
    InMemoryCheckpointStore,
    InterruptRequest,
    ParallelGroupNode,
)
from selectools.types import AgentResult, Message, Role


def _normal_child(state: GraphState):
    state.data["normal"] = "done"
    return AgentResult(message=Message(role=Role.ASSISTANT, content="normal")), state


def _hitl_child_generator(state: GraphState):
    response = yield InterruptRequest(key="approval", prompt="approve?")
    state.data["approval"] = response
    yield AgentResult(message=Message(role=Role.ASSISTANT, content="hitl")), state


def test_parallel_group_propagates_hitl():
    """When a child in a parallel group interrupts, the parent graph must pause."""
    graph = AgentGraph(name="test_parallel_hitl")
    normal_node = GraphNode(name="normal", agent=None, callable_fn=_normal_child)
    hitl_node = GraphNode(name="hitl", agent=None, generator_fn=_hitl_child_generator)
    parallel = ParallelGroupNode(name="group", child_node_names=["normal", "hitl"])
    graph.add_node(normal_node)
    graph.add_node(hitl_node)
    graph.add_node(parallel)
    graph.set_entry("group")
    graph.add_edge("group", "__end__")

    store = InMemoryCheckpointStore()
    result = graph.run("start", checkpoint_store=store)

    assert result.interrupted, f"Expected graph to pause; got: {result}"
    assert result.interrupt_key == "approval"

            async def run_child(
                child_name: str, branch_state: GraphState
            ) -> Tuple[str, AgentResult, GraphState, bool]:
                child_node = self._nodes.get(child_name)
                if child_node is None:
                    raise GraphExecutionError(
                        self.name, child_name, KeyError(f"Child node {child_name!r} not found"), 0
                    )
                if isinstance(child_node, GraphNode):
                    result, new_state, interrupted = await self._aexecute_node(
                        child_node, branch_state, trace, run_id
                    )
                else:
                    result = _make_synthetic_result(branch_state)
                    new_state = branch_state
                    interrupted = False
                return child_name, result, new_state, interrupted

            child_results: Dict[str, List[AgentResult]] = {}
            branch_final_states: List[GraphState] = []
            interrupted_child: Optional[str] = None
            for i, output in enumerate(child_outputs):
                if isinstance(output, BaseException):
                    child_name = node.child_node_names[i]
                    state.errors.append(
                        {"node": child_name, "error": str(output), "type": type(output).__name__}
                    )
                    if self.error_policy == ErrorPolicy.ABORT:
                        exc = output if isinstance(output, Exception) else Exception(str(output))
                        raise GraphExecutionError(self.name, child_name, exc, 0) from output
                    continue
                child_name, result, new_state, child_interrupted = output
                child_results.setdefault(child_name, []).append(result)
                branch_final_states.append(new_state)
                if child_interrupted and interrupted_child is None:
                    interrupted_child = child_name

            # Propagate interrupt metadata to parent state
            if interrupted_child is not None:
                merged_interrupt_marker = {"__parallel_interrupt__": interrupted_child}
            else:
                merged_interrupt_marker = {}

            if merged_interrupt_marker:
                merged.data.update(merged_interrupt_marker)

# Append to tests/agent/test_regression.py (BUG-05)
"""BUG-05: InterruptRequest raised inside a subgraph is silently dropped
by the parent graph. The subgraph's pause state is lost.

Source: Agno #4921 — HITL inside nested contexts fails.
"""
from __future__ import annotations

from selectools import AgentGraph
from selectools.orchestration import (
    GraphNode,
    GraphState,
    InMemoryCheckpointStore,
    InterruptRequest,
    SubgraphNode,
)
from selectools.types import AgentResult, Message, Role


def _hitl_generator(state: GraphState):
    response = yield InterruptRequest(key="approval", prompt="ok?")
    state.data["approval"] = response
    yield AgentResult(message=Message(role=Role.ASSISTANT, content="done")), state


def test_subgraph_propagates_hitl_interrupt():
    inner = AgentGraph(name="inner")
    inner_node = GraphNode(name="gate", agent=None, generator_fn=_hitl_generator)
    inner.add_node(inner_node)
    inner.set_entry("gate")
    inner.add_edge("gate", "__end__")

    outer = AgentGraph(name="outer")
    sub = SubgraphNode(name="nested", graph=inner)
    outer.add_node(sub)
    outer.set_entry("nested")
    outer.add_edge("nested", "__end__")

    store = InMemoryCheckpointStore()
    result = outer.run("start", checkpoint_store=store)

    assert result.interrupted, "Subgraph interrupt must propagate to parent"
    assert result.interrupt_key == "approval"

    async def _aexecute_subgraph(
        self,
        node: SubgraphNode,
        state: GraphState,
        trace: AgentTrace,
        run_id: str,
    ) -> Tuple[AgentResult, GraphState, bool]:
        """Execute a nested AgentGraph as a node."""
        sub_state = GraphState.from_prompt(
            state.data.get(STATE_KEY_LAST_OUTPUT, "")
            or (state.messages[-1].content if state.messages else "")
        )

        for parent_key, sub_key in node.input_map.items():
            if parent_key in state.data:
                sub_state.data[sub_key] = state.data[parent_key]

        sub_result = await node.graph.arun(sub_state, _interrupt_response=None)

        if sub_result.interrupted:
            state.data["__subgraph_interrupt__"] = {
                "key": sub_result.interrupt_key,
                "prompt": sub_result.interrupt_prompt,
                "subgraph": node.name,
            }
            synthetic = AgentResult(
                message=Message(role=Role.ASSISTANT, content=""),
                iterations=sub_result.steps,
                usage=sub_result.total_usage,
            )
            return synthetic, state, True

        for sub_key, parent_key in node.output_map.items():
            if sub_key in sub_result.state.data:
                state.data[parent_key] = sub_result.state.data[sub_key]

        state.data[STATE_KEY_LAST_OUTPUT] = sub_result.content
        state.messages.extend(sub_result.state.messages[-2:])
        state.history.extend(sub_result.state.history)

        synthetic = AgentResult(
            message=Message(role=Role.ASSISTANT, content=sub_result.content),
            iterations=sub_result.steps,
            usage=sub_result.total_usage,
        )
        return synthetic, state, False

# Append to tests/agent/test_regression.py (BUG-06)
"""BUG-06: ConversationMemory has no threading.Lock. Concurrent mutation
from multiple threads races on the _messages list.

Source: PraisonAI #1164, #1260 — thread-unsafe shared mutable state.
"""
from __future__ import annotations

import threading

from selectools.memory import ConversationMemory
from selectools.types import Message, Role


def test_concurrent_add_preserves_all_messages():
    """10 threads × 100 adds = 1000 messages should all be preserved."""
    memory = ConversationMemory(max_messages=10000)
    n_threads = 10
    n_adds = 100
    errors = []

    def worker(thread_id: int):
        try:
            for i in range(n_adds):
                memory.add(Message(role=Role.USER, content=f"t{thread_id}-m{i}"))
        except Exception as e:
            errors.append(e)

    threads = [threading.Thread(target=worker, args=(i,)) for i in range(n_threads)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    assert not errors, f"Worker errors: {errors}"
    history = memory.get_history()
    assert len(history) == n_threads * n_adds, (
        f"Expected {n_threads * n_adds} messages, got {len(history)}"
    )


def test_concurrent_add_with_trim_no_crash():
    """Low max_messages triggers _enforce_limits concurrently — must not crash."""
    memory = ConversationMemory(max_messages=50)
    errors = []

    def worker(thread_id: int):
        try:
            for i in range(200):
                memory.add(Message(role=Role.USER, content=f"t{thread_id}-m{i}"))
        except Exception as e:
            errors.append(e)

    threads = [threading.Thread(target=worker, args=(i,)) for i in range(5)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    assert not errors, f"Worker errors: {errors}"
    assert len(memory.get_history()) <= 50

import threading

        self.max_messages = max_messages
        self.max_tokens = max_tokens
        self._messages: List[Message] = []
        self._summary: Optional[str] = None
        self._last_trimmed: List[Message] = []
        self._lock = threading.RLock()

    def add(self, message: Message) -> None:
        with self._lock:
            self._messages.append(message)
            self._enforce_limits()

    def __getstate__(self) -> Dict[str, Any]:
        state = self.__dict__.copy()
        state.pop("_lock", None)
        return state

    def __setstate__(self, state: Dict[str, Any]) -> None:
        self.__dict__.update(state)
        self._lock = threading.RLock()

# Append to tests/agent/test_regression.py (BUG-07)
"""BUG-07: <think>...</think> content leaks into conversation history.

Source: Agno #6878.
"""
from selectools.providers.anthropic_provider import _strip_reasoning_tags


def test_strip_simple_think_tags():
    text = "<think>This is my reasoning.</think>The answer is 42."
    assert _strip_reasoning_tags(text) == "The answer is 42."


def test_strip_multiline_think_tags():
    text = "<think>\nLine 1\nLine 2\n</think>\nFinal answer."
    assert _strip_reasoning_tags(text).strip() == "Final answer."


def test_strip_multiple_think_blocks():
    text = "<think>first</think>Hello<think>second</think> world"
    assert _strip_reasoning_tags(text) == "Hello world"


def test_no_think_tags_unchanged():
    text = "Plain text with no tags"
    assert _strip_reasoning_tags(text) == text

import re as _re

_THINK_TAG_RE = _re.compile(r"<think>.*?</think>", _re.DOTALL)


def _strip_reasoning_tags(text: str) -> str:
    """Remove <think>...</think> blocks from model output."""
    if not text or "<think>" not in text:
        return text
    return _THINK_TAG_RE.sub("", text)

# Append to tests/agent/test_regression.py (BUG-08)
from unittest.mock import MagicMock


def test_chroma_batches_large_upsert():
    from selectools.rag.stores.chroma import ChromaVectorStore
    from selectools.rag.types import Document

    store = ChromaVectorStore.__new__(ChromaVectorStore)
    store.collection = MagicMock()
    store._batch_size = 100
    store.embedder = MagicMock()
    store.embedder.embed_batch.return_value = [[0.1] * 16 for _ in range(250)]

    docs = [Document(text=f"doc {i}", metadata={}) for i in range(250)]
    store.add_documents(docs)
    assert store.collection.upsert.call_count == 3

        self._batch_size = 5000

        for start in range(0, len(ids), self._batch_size):
            end = start + self._batch_size
            self.collection.upsert(
                ids=ids[start:end],
                embeddings=embeddings[start:end],
                documents=texts[start:end],
                metadatas=metadatas[start:end],
            )

        if isinstance(value, param_type):
            return value
        if isinstance(value, str) and param_type in (int, float, bool):
            try:
                if param_type is bool:
                    lowered = value.strip().lower()
                    if lowered in ("true", "1", "yes", "on"):
                        return True
                    if lowered in ("false", "0", "no", "off"):
                        return False
                    raise ValueError(f"Cannot coerce {value!r} to bool")
                return param_type(value)
            except (ValueError, TypeError) as exc:
                raise ToolValidationError(
                    f"Invalid {name!r}: cannot coerce {value!r} to {param_type.__name__}: {exc}"
                )
        raise ToolValidationError(
            f"Invalid {name!r}: expected {param_type.__name__}, got {type(value).__name__}"
        )

    if origin is Union:
        args = get_args(type_hint)
        non_none_args = [a for a in args if a is not type(None)]
        if len(non_none_args) == 1:
            return _unwrap_type(non_none_args[0])
        if len(non_none_args) > 1:
            return str

    def to_dict(self) -> Dict[str, Any]:
        import json
        try:
            serialized_data = json.loads(json.dumps(self.data))
        except (TypeError, ValueError) as exc:
            raise ValueError(
                f"GraphState.data contains non-serializable values: {exc}. "
                f"All values in state.data must be JSON-compatible for checkpointing."
            )
        return {
            "messages": [m.to_dict() for m in self.messages],
            "history": list(self.history),
            "data": serialized_data,
            "errors": list(self.errors),
            "turn_count": self.turn_count,
        }

def test_different_namespaces_isolated():
    with tempfile.TemporaryDirectory() as tmpdir:
        store = JsonFileSessionStore(directory=tmpdir)
        mem_a = ConversationMemory()
        mem_a.add(Message(role=Role.USER, content="A"))
        store.save("shared_id", mem_a, namespace="agent_a")
        mem_b = ConversationMemory()
        mem_b.add(Message(role=Role.USER, content="B"))
        store.save("shared_id", mem_b, namespace="agent_b")
        assert store.load("shared_id", namespace="agent_a").get_history()[0].content == "A"
        assert store.load("shared_id", namespace="agent_b").get_history()[0].content == "B"

_MAX_SUMMARY_CHARS = 4000


def _append_summary(existing: Optional[str], new_chunk: str) -> str:
    if not existing:
        combined = new_chunk
    else:
        combined = f"{existing} {new_chunk}"
    if len(combined) > _MAX_SUMMARY_CHARS:
        combined = combined[-_MAX_SUMMARY_CHARS:]
    return combined

task = asyncio.ensure_future(handler(*args))
def _log_task_exception(t: "asyncio.Task[Any]") -> None:
    if t.cancelled():
        return
    exc = t.exception()
    if exc is not None:
        logger.warning("Async observer raised: %s", exc, exc_info=exc)
task.add_done_callback(_log_task_exception)