refactor(agent): remove duplicate progress streaming path

This commit is contained in:
chengyongru
2026-08-27 10:56:07 +08:00
committed by chengyongru
parent 2113870e27
commit 55f85b3c1f
10 changed files with 49 additions and 236 deletions
+1 -49
View File
@@ -46,13 +46,11 @@ from nanobot.runtime_context import (
from nanobot.session.history_visibility import is_hidden_history_message from nanobot.session.history_visibility import is_hidden_history_message
from nanobot.session.recovery import PENDING_FOLLOWUP_ID_KEY from nanobot.session.recovery import PENDING_FOLLOWUP_ID_KEY
from nanobot.utils.helpers import ( from nanobot.utils.helpers import (
IncrementalThinkExtractor,
build_assistant_message, build_assistant_message,
estimate_message_tokens, estimate_message_tokens,
estimate_prompt_tokens_chain, estimate_prompt_tokens_chain,
extract_reasoning, extract_reasoning,
strip_reasoning_tags, strip_reasoning_tags,
strip_think,
) )
from nanobot.utils.llm_runtime import LLMRuntime from nanobot.utils.llm_runtime import LLMRuntime
from nanobot.utils.prompt_templates import render_template from nanobot.utils.prompt_templates import render_template
@@ -67,7 +65,6 @@ from nanobot.utils.runtime import (
) )
ContinuationCallback = Callable[[], str | None] ContinuationCallback = Callable[[], str | None]
ProgressCallback = Callable[[str], Awaitable[None]]
RetryWaitCallback = Callable[[str], Awaitable[None]] RetryWaitCallback = Callable[[str], Awaitable[None]]
CheckpointCallback = Callable[[dict[str, Any]], Awaitable[None]] CheckpointCallback = Callable[[dict[str, Any]], Awaitable[None]]
InjectionCallback = Callable[..., Awaitable[Iterable[Any] | None]] InjectionCallback = Callable[..., Awaitable[Iterable[Any] | None]]
@@ -112,7 +109,6 @@ class AgentRunSpec:
session_key: str | None = None session_key: str | None = None
context_block_limit: int | None = None context_block_limit: int | None = None
provider_retry_mode: str = "standard" provider_retry_mode: str = "standard"
progress_callback: ProgressCallback | None = None
retry_wait_callback: RetryWaitCallback | None = None retry_wait_callback: RetryWaitCallback | None = None
checkpoint_callback: CheckpointCallback | None = None checkpoint_callback: CheckpointCallback | None = None
injection_callback: InjectionCallback | None = None injection_callback: InjectionCallback | None = None
@@ -951,14 +947,7 @@ class AgentRunner:
tools=spec.tools.get_definitions(), tools=spec.tools.get_definitions(),
) )
wants_streaming = hook.wants_streaming() wants_streaming = hook.wants_streaming()
progress_callback = spec.progress_callback
wants_progress_streaming = (
not wants_streaming
and progress_callback is not None
and getattr(spec.runtime.provider, "supports_progress_deltas", False) is True
)
progress_state: dict[str, bool] | None = None
active_hosted_tools: dict[str, dict[str, Any]] = {} active_hosted_tools: dict[str, dict[str, Any]] = {}
request_started_at = 0.0 request_started_at = 0.0
first_output_at: float | None = None first_output_at: float | None = None
@@ -1029,40 +1018,6 @@ class AgentRunner:
on_tool_call_delta=_provider_tool_event, on_tool_call_delta=_provider_tool_event,
on_stream_recover=_stream_recover, on_stream_recover=_stream_recover,
) )
elif wants_progress_streaming:
stream_buf = ""
think_extractor = IncrementalThinkExtractor()
progress_state = {"reasoning_open": False}
async def _stream_progress(delta: str) -> None:
nonlocal stream_buf
if not delta:
return
_generation_delta(delta)
prev_clean = strip_think(stream_buf)
stream_buf += delta
new_clean = strip_think(stream_buf)
incremental = new_clean[len(prev_clean):]
if await think_extractor.feed(stream_buf, hook.emit_reasoning):
context.streamed_reasoning = True
progress_state["reasoning_open"] = True
if incremental:
if progress_state["reasoning_open"]:
await hook.emit_reasoning_end()
progress_state["reasoning_open"] = False
context.streamed_content = True
callback = progress_callback
if callback is not None:
await callback(incremental)
coro = spec.runtime.provider.chat_stream_with_retry(
**kwargs,
provider_context=provider_context,
on_content_delta=_stream_progress,
on_tool_call_delta=_provider_tool_event,
)
else: else:
coro = spec.runtime.provider.chat_with_retry( coro = spec.runtime.provider.chat_with_retry(
**kwargs, **kwargs,
@@ -1074,10 +1029,9 @@ class AgentRunner:
# very slow deltas can still run forever. Use a more generous wall-clock # very slow deltas can still run forever. Use a more generous wall-clock
# timeout for streaming while preserving NANOBOT_LLM_TIMEOUT_S=0 as an # timeout for streaming while preserving NANOBOT_LLM_TIMEOUT_S=0 as an
# opt-out for all LLM wall-clock timeouts. # opt-out for all LLM wall-clock timeouts.
is_streaming_request = wants_streaming or wants_progress_streaming
outer_timeout_s = ( outer_timeout_s = (
max(300.0, timeout_s * 2) max(300.0, timeout_s * 2)
if is_streaming_request and timeout_s is not None if wants_streaming and timeout_s is not None
else timeout_s else timeout_s
) )
request_started_at = time.perf_counter() request_started_at = time.perf_counter()
@@ -1115,8 +1069,6 @@ class AgentRunner:
"error": response.content "error": response.content
or "Model request failed before the provider-hosted tool completed.", or "Model request failed before the provider-hosted tool completed.",
}) })
if progress_state and progress_state.get("reasoning_open"):
await hook.emit_reasoning_end()
dropped, all_dropped, original_finish_reason = ( dropped, all_dropped, original_finish_reason = (
self._drop_malformed_tool_calls(response) self._drop_malformed_tool_calls(response)
) )
-2
View File
@@ -603,8 +603,6 @@ _SYNTHETIC_USER_CONTENT = "(conversation continued)"
class LLMProvider(ABC): class LLMProvider(ABC):
"""Base class for LLM providers.""" """Base class for LLM providers."""
supports_progress_deltas = False
_CHAT_RETRY_DELAYS = (1, 2, 4) _CHAT_RETRY_DELAYS = (1, 2, 4)
_PERSISTENT_MAX_DELAY = 60 _PERSISTENT_MAX_DELAY = 60
_PERSISTENT_IDENTICAL_ERROR_LIMIT = 10 _PERSISTENT_IDENTICAL_ERROR_LIMIT = 10
-4
View File
@@ -157,10 +157,6 @@ class FallbackProvider(LLMProvider):
super().set_llm_call_observer(observer) super().set_llm_call_observer(observer)
self._primary.set_llm_call_observer(observer) self._primary.set_llm_call_observer(observer)
@property
def supports_progress_deltas(self) -> bool:
return bool(getattr(self._primary, "supports_progress_deltas", False))
def can_resume_conversation_state( def can_resume_conversation_state(
self, self,
state: ProviderConversationState, state: ProviderConversationState,
@@ -44,8 +44,6 @@ _COMPACTION_RETAINED_CHAR_BUDGET = 256_000
class OpenAICodexProvider(LLMProvider): class OpenAICodexProvider(LLMProvider):
"""Use Codex OAuth to call the Responses API.""" """Use Codex OAuth to call the Responses API."""
supports_progress_deltas = True
def __init__( def __init__(
self, self,
default_model: str = "openai-codex/gpt-5.6-sol", default_model: str = "openai-codex/gpt-5.6-sol",
-2
View File
@@ -63,8 +63,6 @@ def _is_named_x_search_tool(value: object) -> bool:
class XAIGrokProvider(LLMProvider): class XAIGrokProvider(LLMProvider):
"""Call xAI's subscription proxy and expose supported hosted tools.""" """Call xAI's subscription proxy and expose supported hosted tools."""
supports_progress_deltas = True
def __init__( def __init__(
self, self,
default_model: str = DEFAULT_XAI_GROK_MODEL, default_model: str = DEFAULT_XAI_GROK_MODEL,
+2 -12
View File
@@ -373,7 +373,6 @@ class TestToolEventProgress:
"""The /goal command rewrites the prompt but must not bypass WebUI file-edit progress.""" """The /goal command rewrites the prompt but must not bypass WebUI file-edit progress."""
bus = MessageBus() bus = MessageBus()
provider = MagicMock() provider = MagicMock()
provider.supports_progress_deltas = True
provider.get_default_model.return_value = "test-model" provider.get_default_model.return_value = "test-model"
call_count = 0 call_count = 0
@@ -460,7 +459,6 @@ class TestToolEventProgress:
"""Non-streaming channels should get one final reply, not token progress spam.""" """Non-streaming channels should get one final reply, not token progress spam."""
bus = MessageBus() bus = MessageBus()
provider = MagicMock() provider = MagicMock()
provider.supports_progress_deltas = True
provider.get_default_model.return_value = "openai-codex/gpt-5.5" provider.get_default_model.return_value = "openai-codex/gpt-5.5"
provider.chat_with_retry = AsyncMock(return_value=LLMResponse(content="Hello", tool_calls=[])) provider.chat_with_retry = AsyncMock(return_value=LLMResponse(content="Hello", tool_calls=[]))
provider.chat_stream_with_retry = AsyncMock() provider.chat_stream_with_retry = AsyncMock()
@@ -493,7 +491,6 @@ class TestToolEventProgress:
"""Streaming channels still receive provider deltas through stream events.""" """Streaming channels still receive provider deltas through stream events."""
bus = MessageBus() bus = MessageBus()
provider = MagicMock() provider = MagicMock()
provider.supports_progress_deltas = True
provider.get_default_model.return_value = "openai-codex/gpt-5.5" provider.get_default_model.return_value = "openai-codex/gpt-5.5"
async def chat_stream_with_retry(*, on_content_delta, **kwargs): async def chat_stream_with_retry(*, on_content_delta, **kwargs):
@@ -544,7 +541,6 @@ class TestToolEventProgress:
) -> None: ) -> None:
bus = MessageBus() bus = MessageBus()
provider = MagicMock() provider = MagicMock()
provider.supports_progress_deltas = True
provider.get_default_model.return_value = "test-model" provider.get_default_model.return_value = "test-model"
responses = iter([ responses = iter([
LLMResponse(content="first-", finish_reason="length"), LLMResponse(content="first-", finish_reason="length"),
@@ -590,7 +586,6 @@ class TestToolEventProgress:
) -> None: ) -> None:
bus = MessageBus() bus = MessageBus()
provider = MagicMock() provider = MagicMock()
provider.supports_progress_deltas = True
provider.get_default_model.return_value = "test-model" provider.get_default_model.return_value = "test-model"
call_count = 0 call_count = 0
@@ -637,7 +632,6 @@ class TestToolEventProgress:
) -> None: ) -> None:
bus = MessageBus() bus = MessageBus()
provider = MagicMock() provider = MagicMock()
provider.supports_progress_deltas = True
provider.get_default_model.return_value = "test-model" provider.get_default_model.return_value = "test-model"
async def chat_stream_with_retry(*, on_content_delta, **kwargs): async def chat_stream_with_retry(*, on_content_delta, **kwargs):
@@ -728,7 +722,6 @@ class TestToolEventProgress:
"""A no-tools finalization must not be dropped after empty stream retries.""" """A no-tools finalization must not be dropped after empty stream retries."""
bus = MessageBus() bus = MessageBus()
provider = MagicMock() provider = MagicMock()
provider.supports_progress_deltas = True
provider.get_default_model.return_value = "openai-codex/gpt-5.5" provider.get_default_model.return_value = "openai-codex/gpt-5.5"
provider.chat_stream_with_retry = AsyncMock(side_effect=[ provider.chat_stream_with_retry = AsyncMock(side_effect=[
LLMResponse(content=None, tool_calls=[]), LLMResponse(content=None, tool_calls=[]),
@@ -776,7 +769,6 @@ class TestToolEventProgress:
) -> None: ) -> None:
bus = MessageBus() bus = MessageBus()
provider = MagicMock() provider = MagicMock()
provider.supports_progress_deltas = True
provider.get_default_model.return_value = "openai-codex/gpt-5.5" provider.get_default_model.return_value = "openai-codex/gpt-5.5"
first_request_started = asyncio.Event() first_request_started = asyncio.Event()
release_first_request = asyncio.Event() release_first_request = asyncio.Event()
@@ -935,7 +927,6 @@ class TestToolEventProgress:
"""Recovered streaming output should use a new stream segment.""" """Recovered streaming output should use a new stream segment."""
bus = MessageBus() bus = MessageBus()
provider = MagicMock() provider = MagicMock()
provider.supports_progress_deltas = True
provider.get_default_model.return_value = "openai-codex/gpt-5.5" provider.get_default_model.return_value = "openai-codex/gpt-5.5"
async def chat_stream_with_retry(*, on_content_delta, on_stream_recover, **kwargs): async def chat_stream_with_retry(*, on_content_delta, on_stream_recover, **kwargs):
@@ -988,13 +979,12 @@ class TestToolEventProgress:
provider.chat_with_retry.assert_not_awaited() provider.chat_with_retry.assert_not_awaited()
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_streamed_progress_is_not_repeated_before_tool_execution( async def test_streamed_content_is_not_repeated_before_tool_execution(
self, self,
tmp_path: Path, tmp_path: Path,
) -> None: ) -> None:
"""If content was already streamed as progress, tool setup should not repeat it.""" """If content was already streamed, tool setup should not repeat it."""
loop = _make_loop(tmp_path) loop = _make_loop(tmp_path)
loop.provider.supports_progress_deltas = True
tool_call = ToolCallRequest(id="call1", name="custom_tool", arguments={"path": "foo.txt"}) tool_call = ToolCallRequest(id="call1", name="custom_tool", arguments={"path": "foo.txt"})
calls = iter([ calls = iter([
LLMResponse(content="I will inspect it.", tool_calls=[tool_call]), LLMResponse(content="I will inspect it.", tool_calls=[tool_call]),
+2 -68
View File
@@ -798,64 +798,6 @@ async def test_runner_times_out_never_ending_streaming_request():
provider.chat_with_retry.assert_not_awaited() provider.chat_with_retry.assert_not_awaited()
@pytest.mark.asyncio
async def test_runner_closes_progress_reasoning_on_streaming_wall_timeout():
from nanobot.agent.hook import AgentHook
from nanobot.agent.runner import AgentRunner
provider = MagicMock(spec=LLMProvider)
provider.supports_progress_deltas = True
events: list[tuple[str, str | None]] = []
async def chat_stream_with_retry(*, on_content_delta, **kwargs):
try:
await on_content_delta("<think>working...</think>")
await asyncio.sleep(3600)
finally:
events.append(("provider_cancelled", None))
provider.chat_stream_with_retry = chat_stream_with_retry
provider.chat_with_retry = AsyncMock()
tools = MagicMock()
tools.get_definitions.return_value = []
class ProgressReasoningHook(AgentHook):
async def emit_reasoning(self, reasoning_content: str | None) -> None:
if reasoning_content:
events.append(("reasoning", reasoning_content))
async def emit_reasoning_end(self) -> None:
events.append(("reasoning_end", None))
real_wait_for = asyncio.wait_for
async def fake_wait_for(coro, *, timeout):
assert timeout == 300.0
return await real_wait_for(coro, timeout=0.01)
runner = AgentRunner()
with patch("nanobot.agent.runner.asyncio.wait_for", fake_wait_for):
result = await runner.run(make_run_spec(provider,
initial_messages=[{"role": "user", "content": "think forever"}],
tools=tools,
model="test-model",
max_iterations=1,
max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
hook=ProgressReasoningHook(),
progress_callback=AsyncMock(),
llm_timeout_s=1,
))
assert result.stop_reason == "error"
assert result.final_content == "Error calling LLM: timed out after 300s"
assert events == [
("reasoning", "working..."),
("provider_cancelled", None),
("reasoning_end", None),
]
provider.chat_with_retry.assert_not_awaited()
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_runner_replaces_empty_tool_result_with_marker(): async def test_runner_replaces_empty_tool_result_with_marker():
from nanobot.agent.runner import AgentRunner from nanobot.agent.runner import AgentRunner
@@ -1285,13 +1227,8 @@ async def test_runner_accumulates_usage_and_preserves_cache_reads():
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_runner_binds_on_retry_wait_to_retry_callback_not_progress(): async def test_runner_binds_on_retry_wait_callback():
"""Regression: provider retry heartbeats must route through """Provider retry heartbeats use the explicitly supplied callback."""
``retry_wait_callback``, not ``progress_callback``. Binding them to
the progress callback (as an earlier runtime refactor did) caused
internal retry diagnostics like "Model request failed, retry in 1s"
to leak to end-user channels as normal progress updates.
"""
from nanobot.agent.runner import AgentRunner from nanobot.agent.runner import AgentRunner
captured: dict = {} captured: dict = {}
@@ -1305,7 +1242,6 @@ async def test_runner_binds_on_retry_wait_to_retry_callback_not_progress():
tools = MagicMock() tools = MagicMock()
tools.get_definitions.return_value = [] tools.get_definitions.return_value = []
progress_cb = AsyncMock()
retry_wait_cb = AsyncMock() retry_wait_cb = AsyncMock()
runner = AgentRunner() runner = AgentRunner()
@@ -1318,12 +1254,10 @@ async def test_runner_binds_on_retry_wait_to_retry_callback_not_progress():
model="test-model", model="test-model",
max_iterations=1, max_iterations=1,
max_tool_result_chars=_MAX_TOOL_RESULT_CHARS, max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
progress_callback=progress_cb,
retry_wait_callback=retry_wait_cb, retry_wait_callback=retry_wait_cb,
)) ))
assert captured["on_retry_wait"] is retry_wait_cb assert captured["on_retry_wait"] is retry_wait_cb
assert captured["on_retry_wait"] is not progress_cb
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
+20 -68
View File
@@ -1,4 +1,4 @@
"""Tests for provider progress delta routing in the shared runner.""" """Tests for runner progress hooks and provider event routing."""
import asyncio import asyncio
from unittest.mock import AsyncMock, MagicMock from unittest.mock import AsyncMock, MagicMock
@@ -6,7 +6,6 @@ from unittest.mock import AsyncMock, MagicMock
import pytest import pytest
from agent.runner_helpers import make_run_spec from agent.runner_helpers import make_run_spec
from nanobot.agent.hook import CompositeHook
from nanobot.agent.hooks import FileEditActivityHook from nanobot.agent.hooks import FileEditActivityHook
from nanobot.agent.progress_hook import AgentProgressHook from nanobot.agent.progress_hook import AgentProgressHook
from nanobot.agent.runner import AgentRunner from nanobot.agent.runner import AgentRunner
@@ -17,45 +16,9 @@ from nanobot.providers.base import LLMResponse, ToolCallRequest
_MAX_TOOL_RESULT_CHARS = AgentDefaults().max_tool_result_chars _MAX_TOOL_RESULT_CHARS = AgentDefaults().max_tool_result_chars
@pytest.mark.asyncio
async def test_runner_streams_provider_progress_deltas_by_default():
"""Direct runner users keep the existing opt-in provider progress behavior."""
provider = MagicMock()
provider.supports_progress_deltas = True
async def chat_stream_with_retry(*, on_content_delta, **kwargs):
await on_content_delta("he")
await on_content_delta("llo")
return LLMResponse(content="hello", tool_calls=[], usage=None)
provider.chat_stream_with_retry = chat_stream_with_retry
provider.chat_with_retry = AsyncMock()
tools = MagicMock()
tools.get_definitions.return_value = []
progress_cb = AsyncMock()
runner = AgentRunner()
result = await runner.run(make_run_spec(provider,
initial_messages=[
{"role": "system", "content": "system"},
{"role": "user", "content": "hi"},
],
tools=tools,
model="test-model",
max_iterations=1,
max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
progress_callback=progress_cb,
))
assert result.final_content == "hello"
assert [call.args[0] for call in progress_cb.await_args_list] == ["he", "llo"]
provider.chat_with_retry.assert_not_awaited()
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_runner_routes_hosted_tool_events_to_structured_progress(): async def test_runner_routes_hosted_tool_events_to_structured_progress():
provider = MagicMock() provider = MagicMock()
provider.supports_progress_deltas = True
async def chat_stream_with_retry(*, on_content_delta, on_tool_call_delta, **kwargs): async def chat_stream_with_retry(*, on_content_delta, on_tool_call_delta, **kwargs):
await on_tool_call_delta({ await on_tool_call_delta({
@@ -88,13 +51,17 @@ async def test_runner_routes_hosted_tool_events_to_structured_progress():
tools.get_definitions.return_value = [] tools.get_definitions.return_value = []
progress_events: list[dict] = [] progress_events: list[dict] = []
progress_text: list[str] = [] progress_text: list[str] = []
streamed_text: list[str] = []
async def progress_cb(content, *, tool_events=None, **kwargs): async def progress_cb(content, *, tool_events=None, **kwargs):
progress_text.append(content) progress_text.append(content)
if tool_events: if tool_events:
progress_events.extend(tool_events) progress_events.extend(tool_events)
hook = CompositeHook([AgentProgressHook(on_progress=progress_cb)]) async def stream_cb(content: str) -> None:
streamed_text.append(content)
hook = AgentProgressHook(on_progress=progress_cb, on_stream=stream_cb)
result = await AgentRunner().run(make_run_spec( result = await AgentRunner().run(make_run_spec(
provider, provider,
initial_messages=[{"role": "user", "content": "search X"}], initial_messages=[{"role": "user", "content": "search X"}],
@@ -102,7 +69,6 @@ async def test_runner_routes_hosted_tool_events_to_structured_progress():
model="test-model", model="test-model",
max_iterations=1, max_iterations=1,
max_tool_result_chars=_MAX_TOOL_RESULT_CHARS, max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
progress_callback=progress_cb,
hook=hook, hook=hook,
)) ))
@@ -133,14 +99,14 @@ async def test_runner_routes_hosted_tool_events_to_structured_progress():
"embeds": [], "embeds": [],
}, },
] ]
assert progress_text == ['search X "nanobot oauth"', "", "done"] assert progress_text == ['search X "nanobot oauth"', ""]
assert streamed_text == ["done"]
provider.chat_with_retry.assert_not_awaited() provider.chat_with_retry.assert_not_awaited()
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_runner_fails_pending_hosted_tool_when_model_request_fails(): async def test_runner_fails_pending_hosted_tool_when_model_request_fails():
provider = MagicMock() provider = MagicMock()
provider.supports_progress_deltas = True
async def chat_stream_with_retry(*, on_tool_call_delta, **kwargs): async def chat_stream_with_retry(*, on_tool_call_delta, **kwargs):
await on_tool_call_delta({ await on_tool_call_delta({
@@ -166,7 +132,10 @@ async def test_runner_fails_pending_hosted_tool_when_model_request_fails():
if tool_events: if tool_events:
progress_events.extend(tool_events) progress_events.extend(tool_events)
hook = CompositeHook([AgentProgressHook(on_progress=progress_cb)]) async def stream_cb(_content: str) -> None:
pass
hook = AgentProgressHook(on_progress=progress_cb, on_stream=stream_cb)
result = await AgentRunner().run(make_run_spec( result = await AgentRunner().run(make_run_spec(
provider, provider,
initial_messages=[{"role": "user", "content": "search X"}], initial_messages=[{"role": "user", "content": "search X"}],
@@ -174,7 +143,6 @@ async def test_runner_fails_pending_hosted_tool_when_model_request_fails():
model="test-model", model="test-model",
max_iterations=1, max_iterations=1,
max_tool_result_chars=_MAX_TOOL_RESULT_CHARS, max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
progress_callback=progress_cb,
hook=hook, hook=hook,
)) ))
@@ -200,7 +168,6 @@ async def test_runner_fails_pending_hosted_tool_when_model_request_fails():
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_runner_emits_write_file_diff_from_tool_execution_snapshots(tmp_path): async def test_runner_emits_write_file_diff_from_tool_execution_snapshots(tmp_path):
provider = MagicMock() provider = MagicMock()
provider.supports_progress_deltas = True
call_count = 0 call_count = 0
progress_events: list[dict] = [] progress_events: list[dict] = []
(tmp_path / "big.txt").write_text("old\n", encoding="utf-8") (tmp_path / "big.txt").write_text("old\n", encoding="utf-8")
@@ -218,7 +185,7 @@ async def test_runner_emits_write_file_diff_from_tool_execution_snapshots(tmp_pa
def prepare_call(self, name, params): def prepare_call(self, name, params):
return tool, params, None return tool, params, None
async def chat_stream_with_retry(**kwargs): async def chat_with_retry(**kwargs):
nonlocal call_count nonlocal call_count
call_count += 1 call_count += 1
if call_count == 1: if call_count == 1:
@@ -235,8 +202,7 @@ async def test_runner_emits_write_file_diff_from_tool_execution_snapshots(tmp_pa
) )
return LLMResponse(content="done", tool_calls=[], usage=None) return LLMResponse(content="done", tool_calls=[], usage=None)
provider.chat_stream_with_retry = chat_stream_with_retry provider.chat_with_retry = chat_with_retry
provider.chat_with_retry = AsyncMock()
tools = Tools() tools = Tools()
runner = AgentRunner() runner = AgentRunner()
@@ -246,7 +212,6 @@ async def test_runner_emits_write_file_diff_from_tool_execution_snapshots(tmp_pa
model="test-model", model="test-model",
max_iterations=2, max_iterations=2,
max_tool_result_chars=_MAX_TOOL_RESULT_CHARS, max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
progress_callback=progress_cb,
workspace=tmp_path, workspace=tmp_path,
hook=FileEditActivityHook(on_progress=progress_cb, workspace=tmp_path), hook=FileEditActivityHook(on_progress=progress_cb, workspace=tmp_path),
)) ))
@@ -263,13 +228,11 @@ async def test_runner_emits_write_file_diff_from_tool_execution_snapshots(tmp_pa
and event["diff"]["format"] == "unified" and event["diff"]["format"] == "unified"
for event in progress_events for event in progress_events
) )
provider.chat_with_retry.assert_not_awaited()
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_runner_emits_edit_file_diff_from_tool_execution_snapshots(tmp_path): async def test_runner_emits_edit_file_diff_from_tool_execution_snapshots(tmp_path):
provider = MagicMock() provider = MagicMock()
provider.supports_progress_deltas = True
call_count = 0 call_count = 0
progress_events: list[dict] = [] progress_events: list[dict] = []
target = tmp_path / "notes.txt" target = tmp_path / "notes.txt"
@@ -288,7 +251,7 @@ async def test_runner_emits_edit_file_diff_from_tool_execution_snapshots(tmp_pat
def prepare_call(self, name, params): def prepare_call(self, name, params):
return tool, params, None return tool, params, None
async def chat_stream_with_retry(**kwargs): async def chat_with_retry(**kwargs):
nonlocal call_count nonlocal call_count
call_count += 1 call_count += 1
if call_count == 1: if call_count == 1:
@@ -309,8 +272,7 @@ async def test_runner_emits_edit_file_diff_from_tool_execution_snapshots(tmp_pat
) )
return LLMResponse(content="done", tool_calls=[], usage=None) return LLMResponse(content="done", tool_calls=[], usage=None)
provider.chat_stream_with_retry = chat_stream_with_retry provider.chat_with_retry = chat_with_retry
provider.chat_with_retry = AsyncMock()
tools = Tools() tools = Tools()
runner = AgentRunner() runner = AgentRunner()
@@ -320,7 +282,6 @@ async def test_runner_emits_edit_file_diff_from_tool_execution_snapshots(tmp_pat
model="test-model", model="test-model",
max_iterations=2, max_iterations=2,
max_tool_result_chars=_MAX_TOOL_RESULT_CHARS, max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
progress_callback=progress_cb,
workspace=tmp_path, workspace=tmp_path,
hook=FileEditActivityHook(on_progress=progress_cb, workspace=tmp_path), hook=FileEditActivityHook(on_progress=progress_cb, workspace=tmp_path),
)) ))
@@ -335,13 +296,11 @@ async def test_runner_emits_edit_file_diff_from_tool_execution_snapshots(tmp_pat
and event["diff"]["format"] == "unified" and event["diff"]["format"] == "unified"
for event in progress_events for event in progress_events
) )
provider.chat_with_retry.assert_not_awaited()
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_runner_marks_file_edit_activity_failed_when_tool_errors(tmp_path): async def test_runner_marks_file_edit_activity_failed_when_tool_errors(tmp_path):
provider = MagicMock() provider = MagicMock()
provider.supports_progress_deltas = True
call_count = 0 call_count = 0
progress_events: list[dict] = [] progress_events: list[dict] = []
@@ -358,7 +317,7 @@ async def test_runner_marks_file_edit_activity_failed_when_tool_errors(tmp_path)
def prepare_call(self, name, params): def prepare_call(self, name, params):
return tool, params, None return tool, params, None
async def chat_stream_with_retry(**kwargs): async def chat_with_retry(**kwargs):
nonlocal call_count nonlocal call_count
call_count += 1 call_count += 1
if call_count == 1: if call_count == 1:
@@ -375,8 +334,7 @@ async def test_runner_marks_file_edit_activity_failed_when_tool_errors(tmp_path)
) )
return LLMResponse(content="done", tool_calls=[], usage=None) return LLMResponse(content="done", tool_calls=[], usage=None)
provider.chat_stream_with_retry = chat_stream_with_retry provider.chat_with_retry = chat_with_retry
provider.chat_with_retry = AsyncMock()
tools = Tools() tools = Tools()
runner = AgentRunner() runner = AgentRunner()
@@ -386,7 +344,6 @@ async def test_runner_marks_file_edit_activity_failed_when_tool_errors(tmp_path)
model="test-model", model="test-model",
max_iterations=2, max_iterations=2,
max_tool_result_chars=_MAX_TOOL_RESULT_CHARS, max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
progress_callback=progress_cb,
workspace=tmp_path, workspace=tmp_path,
hook=FileEditActivityHook(on_progress=progress_cb, workspace=tmp_path), hook=FileEditActivityHook(on_progress=progress_cb, workspace=tmp_path),
)) ))
@@ -395,13 +352,11 @@ async def test_runner_marks_file_edit_activity_failed_when_tool_errors(tmp_path)
assert progress_events[-1]["path"] == "aborted.txt" assert progress_events[-1]["path"] == "aborted.txt"
assert progress_events[-1]["phase"] == "error" assert progress_events[-1]["phase"] == "error"
assert progress_events[-1]["status"] == "error" assert progress_events[-1]["status"] == "error"
provider.chat_with_retry.assert_not_awaited()
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_runner_marks_file_edit_activity_failed_when_cancelled(tmp_path): async def test_runner_marks_file_edit_activity_failed_when_cancelled(tmp_path):
provider = MagicMock() provider = MagicMock()
provider.supports_progress_deltas = True
progress_events: list[dict] = [] progress_events: list[dict] = []
executing = asyncio.Event() executing = asyncio.Event()
target = tmp_path / "cancelled.txt" target = tmp_path / "cancelled.txt"
@@ -426,7 +381,7 @@ async def test_runner_marks_file_edit_activity_failed_when_cancelled(tmp_path):
def prepare_call(self, name, params): def prepare_call(self, name, params):
return tool, params, None return tool, params, None
async def chat_stream_with_retry(**kwargs): async def chat_with_retry(**kwargs):
return LLMResponse( return LLMResponse(
content=None, content=None,
tool_calls=[ tool_calls=[
@@ -439,8 +394,7 @@ async def test_runner_marks_file_edit_activity_failed_when_cancelled(tmp_path):
usage=None, usage=None,
) )
provider.chat_stream_with_retry = chat_stream_with_retry provider.chat_with_retry = chat_with_retry
provider.chat_with_retry = AsyncMock()
tools = Tools() tools = Tools()
runner = AgentRunner() runner = AgentRunner()
@@ -450,7 +404,6 @@ async def test_runner_marks_file_edit_activity_failed_when_cancelled(tmp_path):
model="test-model", model="test-model",
max_iterations=2, max_iterations=2,
max_tool_result_chars=_MAX_TOOL_RESULT_CHARS, max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
progress_callback=progress_cb,
workspace=tmp_path, workspace=tmp_path,
hook=FileEditActivityHook(on_progress=progress_cb, workspace=tmp_path), hook=FileEditActivityHook(on_progress=progress_cb, workspace=tmp_path),
))) )))
@@ -464,4 +417,3 @@ async def test_runner_marks_file_edit_activity_failed_when_cancelled(tmp_path):
assert progress_events[-1]["path"] == "cancelled.txt" assert progress_events[-1]["path"] == "cancelled.txt"
assert progress_events[-1]["status"] == "error" assert progress_events[-1]["status"] == "error"
assert progress_events[-1]["error"] == "Task interrupted before this tool finished." assert progress_events[-1]["error"] == "Task interrupted before this tool finished."
provider.chat_with_retry.assert_not_awaited()
+24 -23
View File
@@ -15,6 +15,7 @@ import pytest
from agent.runner_helpers import make_run_spec from agent.runner_helpers import make_run_spec
from nanobot.agent.hook import AgentHook, AgentHookContext from nanobot.agent.hook import AgentHook, AgentHookContext
from nanobot.agent.progress_hook import AgentProgressHook
from nanobot.config.schema import AgentDefaults from nanobot.config.schema import AgentDefaults
from nanobot.providers.base import LLMResponse, LLMUsage, ToolCallRequest from nanobot.providers.base import LLMResponse, LLMUsage, ToolCallRequest
@@ -35,6 +36,18 @@ class _RecordingHook(AgentHook):
self.end_calls += 1 self.end_calls += 1
class _StreamRecordingHook(_RecordingHook):
def __init__(self) -> None:
super().__init__()
self.streamed: list[str] = []
def wants_streaming(self) -> bool:
return True
async def on_stream(self, _ctx: AgentHookContext, delta: str) -> None:
self.streamed.append(delta)
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_runner_preserves_reasoning_fields_in_assistant_history(): async def test_runner_preserves_reasoning_fields_in_assistant_history():
"""Reasoning fields ride along on the persisted assistant message so """Reasoning fields ride along on the persisted assistant message so
@@ -201,7 +214,6 @@ async def test_runner_emits_reasoning_content_even_when_answer_was_streamed():
from nanobot.agent.runner import AgentRunner from nanobot.agent.runner import AgentRunner
provider = MagicMock() provider = MagicMock()
provider.supports_progress_deltas = True
async def chat_stream_with_retry(*, on_content_delta=None, **kwargs): async def chat_stream_with_retry(*, on_content_delta=None, **kwargs):
if on_content_delta: if on_content_delta:
@@ -218,12 +230,7 @@ async def test_runner_emits_reasoning_content_even_when_answer_was_streamed():
tools = MagicMock() tools = MagicMock()
tools.get_definitions.return_value = [] tools.get_definitions.return_value = []
progress_calls: list[str] = [] hook = _StreamRecordingHook()
async def _progress(content: str, **_kwargs):
progress_calls.append(content)
hook = _RecordingHook()
runner = AgentRunner() runner = AgentRunner()
result = await runner.run(make_run_spec(provider, result = await runner.run(make_run_spec(provider,
initial_messages=[{"role": "user", "content": "question"}], initial_messages=[{"role": "user", "content": "question"}],
@@ -232,11 +239,10 @@ async def test_runner_emits_reasoning_content_even_when_answer_was_streamed():
max_iterations=3, max_iterations=3,
max_tool_result_chars=_MAX_TOOL_RESULT_CHARS, max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
hook=hook, hook=hook,
progress_callback=_progress,
)) ))
assert result.final_content == "The answer." assert result.final_content == "The answer."
assert progress_calls, "answer should have streamed via progress callback" assert hook.streamed == ["The ", "answer."]
assert hook.emitted == ["step-by-step deduction"] assert hook.emitted == ["step-by-step deduction"]
@@ -247,7 +253,6 @@ async def test_runner_does_not_double_emit_when_inline_think_already_streamed():
from nanobot.agent.runner import AgentRunner from nanobot.agent.runner import AgentRunner
provider = MagicMock() provider = MagicMock()
provider.supports_progress_deltas = True
async def chat_stream_with_retry(*, on_content_delta=None, **kwargs): async def chat_stream_with_retry(*, on_content_delta=None, **kwargs):
if on_content_delta: if on_content_delta:
@@ -263,10 +268,16 @@ async def test_runner_does_not_double_emit_when_inline_think_already_streamed():
tools = MagicMock() tools = MagicMock()
tools.get_definitions.return_value = [] tools.get_definitions.return_value = []
async def _progress(content: str, **_kwargs): reasoning_events: list[str] = []
async def _progress(content: str, *, reasoning: bool = False, **_kwargs):
if reasoning:
reasoning_events.append(content)
async def _stream(_content: str) -> None:
pass pass
hook = _RecordingHook() hook = AgentProgressHook(on_progress=_progress, on_stream=_stream)
runner = AgentRunner() runner = AgentRunner()
result = await runner.run(make_run_spec(provider, result = await runner.run(make_run_spec(provider,
initial_messages=[{"role": "user", "content": "question"}], initial_messages=[{"role": "user", "content": "question"}],
@@ -275,12 +286,10 @@ async def test_runner_does_not_double_emit_when_inline_think_already_streamed():
max_iterations=3, max_iterations=3,
max_tool_result_chars=_MAX_TOOL_RESULT_CHARS, max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
hook=hook, hook=hook,
progress_callback=_progress,
)) ))
assert result.final_content == "The answer." assert result.final_content == "The answer."
assert hook.emitted == ["working..."] assert reasoning_events == ["working..."]
assert hook.end_calls >= 1, "reasoning stream must be closed once the answer starts"
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -320,14 +329,6 @@ async def test_runner_closes_reasoning_stream_after_one_shot_response():
assert hook.end_calls == 1 assert hook.end_calls == 1
class _StreamRecordingHook(_RecordingHook):
def wants_streaming(self) -> bool:
return True
async def on_stream(self, _ctx: AgentHookContext, delta: str) -> None:
pass
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_runner_streams_native_thinking_deltas_without_post_hoc_dup(): async def test_runner_streams_native_thinking_deltas_without_post_hoc_dup():
"""Anthropic-style ``on_thinking_delta`` should fan out to ``emit_reasoning``; """Anthropic-style ``on_thinking_delta`` should fan out to ``emit_reasoning``;
-6
View File
@@ -63,9 +63,3 @@ def test_explicit_provider_import_still_works(monkeypatch) -> None:
finally: finally:
monkeypatch.undo() monkeypatch.undo()
setattr(sys.modules["nanobot"], "providers", original_package) setattr(sys.modules["nanobot"], "providers", original_package)
def test_openai_codex_supports_progress_deltas() -> None:
from nanobot.providers.openai_codex_provider import OpenAICodexProvider
assert OpenAICodexProvider.supports_progress_deltas is True