Grayscale/nanobot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: guard tool execution against non-compliant API gateway injection

Browse Source
This commit is contained in:
Subal
2026-04-17 20:39:46 +08:00
committed by Xubin Ren
parent aabc3d5017
commit 322da6ca06
4 changed files with 218 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nanobot/agent/runner.py
+110 -53
View File
@@ -40,14 +40,20 @@ _MAX_INJECTION_CYCLES = 5
_SNIP_SAFETY_BUFFER = 1024
_MICROCOMPACT_KEEP_RECENT = 10
_MICROCOMPACT_MIN_CHARS = 500
_COMPACTABLE_TOOLS = frozenset({
"read_file", "exec", "grep", "glob",
"web_search", "web_fetch", "list_dir",
})
_COMPACTABLE_TOOLS = frozenset(
{
"read_file",
"exec",
"grep",
"glob",
"web_search",
"web_fetch",
"list_dir",
}
)
_BACKFILL_CONTENT = "[Tool result unavailable — call was interrupted or lost]"
@dataclass(slots=True)
class AgentRunSpec:
"""Configuration for a single agent execution."""
@@ -120,11 +126,7 @@ class AgentRunner:
) -> None:
"""Append injected user messages while preserving role alternation."""
for injection in injections:
if (
messages
and injection.get("role") == "user"
and messages[-1].get("role") == "user"
):
if messages and injection.get("role") == "user" and messages[-1].get("role") == "user":
merged = dict(messages[-1])
merged["content"] = cls._merge_message_content(
merged.get("content"),
@@ -174,7 +176,10 @@ class AgentRunner:
self._append_injected_messages(messages, injections)
logger.info(
"Injected {} follow-up message(s) {} ({}/{})",
len(injections), phase, injection_cycles, _MAX_INJECTION_CYCLES,
len(injections),
phase,
injection_cycles,
_MAX_INJECTION_CYCLES,
)
return True, injection_cycles
@@ -190,12 +195,9 @@ class AgentRunner:
return []
try:
signature = inspect.signature(spec.injection_callback)
accepts_limit = (
"limit" in signature.parameters
or any(
parameter.kind is inspect.Parameter.VAR_KEYWORD
for parameter in signature.parameters.values()
)
accepts_limit = "limit" in signature.parameters or any(
parameter.kind is inspect.Parameter.VAR_KEYWORD
for parameter in signature.parameters.values()
)
if accepts_limit:
items = await spec.injection_callback(limit=_MAX_INJECTIONS_PER_TURN)
@@ -218,7 +220,9 @@ class AgentRunner:
dropped = len(injected_messages) - _MAX_INJECTIONS_PER_TURN
logger.warning(
"Injection callback returned {} messages, capping to {} ({} dropped)",
len(injected_messages), _MAX_INJECTIONS_PER_TURN, dropped,
len(injected_messages),
_MAX_INJECTIONS_PER_TURN,
dropped,
)
injected_messages = injected_messages[:_MAX_INJECTIONS_PER_TURN]
return injected_messages
@@ -273,7 +277,7 @@ class AgentRunner:
context.tool_calls = list(response.tool_calls)
self._accumulate_usage(usage, raw_usage)
if response.has_tool_calls:
if response.should_execute_tools:
if hook.wants_streaming():
await hook.on_stream_end(context, resuming=True)
@@ -293,7 +297,9 @@ class AgentRunner:
"model": spec.model,
"assistant_message": assistant_message,
"completed_tool_results": [],
"pending_tool_calls": [tc.to_openai_tool_call() for tc in response.tool_calls],
"pending_tool_calls": [
tc.to_openai_tool_call() for tc in response.tool_calls
],
},
)
@@ -332,7 +338,10 @@ class AgentRunner:
context.stop_reason = stop_reason
await hook.after_iteration(context)
should_continue, injection_cycles = await self._try_drain_injections(
spec, messages, None, injection_cycles,
spec,
messages,
None,
injection_cycles,
phase="after tool error",
)
if should_continue:
@@ -354,7 +363,10 @@ class AgentRunner:
length_recovery_count = 0
# Checkpoint 1: drain injections after tools, before next LLM call
_drained, injection_cycles = await self._try_drain_injections(
spec, messages, None, injection_cycles,
spec,
messages,
None,
injection_cycles,
phase="after tool execution",
)
if _drained:
@@ -362,6 +374,13 @@ class AgentRunner:
await hook.after_iteration(context)
continue
elif response.has_tool_calls:
logger.warning(
"Ignoring tool calls under finish_reason='%s' for %s",
response.finish_reason,
spec.session_key or "default",
)
clean = hook.finalize_content(context, response.content)
if response.finish_reason != "error" and is_blank_text(clean):
empty_content_retries += 1
@@ -406,11 +425,13 @@ class AgentRunner:
)
if hook.wants_streaming():
await hook.on_stream_end(context, resuming=True)
messages.append(build_assistant_message(
clean,
reasoning_content=response.reasoning_content,
thinking_blocks=response.thinking_blocks,
))
messages.append(
build_assistant_message(
clean,
reasoning_content=response.reasoning_content,
thinking_blocks=response.thinking_blocks,
)
)
messages.append(build_length_recovery_message())
await hook.after_iteration(context)
continue
@@ -427,7 +448,10 @@ class AgentRunner:
# If injections are found we keep the stream alive (resuming=True)
# so streaming channels don't prematurely finalize the card.
should_continue, injection_cycles = await self._try_drain_injections(
spec, messages, assistant_message, injection_cycles,
spec,
messages,
assistant_message,
injection_cycles,
phase="after final response",
iteration=iteration,
)
@@ -451,7 +475,10 @@ class AgentRunner:
context.stop_reason = stop_reason
await hook.after_iteration(context)
should_continue, injection_cycles = await self._try_drain_injections(
spec, messages, None, injection_cycles,
spec,
messages,
None,
injection_cycles,
phase="after LLM error",
)
if should_continue:
@@ -468,7 +495,10 @@ class AgentRunner:
context.stop_reason = stop_reason
await hook.after_iteration(context)
should_continue, injection_cycles = await self._try_drain_injections(
spec, messages, None, injection_cycles,
spec,
messages,
None,
injection_cycles,
phase="after empty response",
)
if should_continue:
@@ -476,11 +506,14 @@ class AgentRunner:
continue
break
messages.append(assistant_message or build_assistant_message(
clean,
reasoning_content=response.reasoning_content,
thinking_blocks=response.thinking_blocks,
))
messages.append(
assistant_message
or build_assistant_message(
clean,
reasoning_content=response.reasoning_content,
thinking_blocks=response.thinking_blocks,
)
)
await self._emit_checkpoint(
spec,
{
@@ -516,7 +549,10 @@ class AgentRunner:
# We ignore should_continue here because the for-loop has already
# exhausted all iterations.
drained_after_max_iterations, injection_cycles = await self._try_drain_injections(
spec, messages, None, injection_cycles,
spec,
messages,
None,
injection_cycles,
phase="after max_iterations",
)
if drained_after_max_iterations:
@@ -568,6 +604,7 @@ class AgentRunner:
tools=spec.tools.get_definitions(),
)
if hook.wants_streaming():
async def _stream(delta: str) -> None:
await hook.on_stream(context, delta)
@@ -621,13 +658,19 @@ class AgentRunner:
tool_results: list[tuple[Any, dict[str, str], BaseException | None]] = []
for batch in batches:
if spec.concurrent_tools and len(batch) > 1:
tool_results.extend(await asyncio.gather(*(
self._run_tool(spec, tool_call, external_lookup_counts)
for tool_call in batch
)))
tool_results.extend(
await asyncio.gather(
*(
self._run_tool(spec, tool_call, external_lookup_counts)
for tool_call in batch
)
)
)
else:
for tool_call in batch:
tool_results.append(await self._run_tool(spec, tool_call, external_lookup_counts))
tool_results.append(
await self._run_tool(spec, tool_call, external_lookup_counts)
)
results: list[Any] = []
events: list[dict[str, str]] = []
@@ -675,7 +718,11 @@ class AgentRunner:
"status": "error",
"detail": prep_error.split(": ", 1)[-1][:120],
}
return prep_error + _HINT, event, RuntimeError(prep_error) if spec.fail_on_tool_error else None
return (
prep_error + _HINT,
event,
RuntimeError(prep_error) if spec.fail_on_tool_error else None,
)
try:
if tool is not None:
result = await tool.execute(**params)
@@ -737,7 +784,11 @@ class AgentRunner:
@staticmethod
def _append_model_error_placeholder(messages: list[dict[str, Any]]) -> None:
if messages and messages[-1].get("role") == "assistant" and not messages[-1].get("tool_calls"):
if (
messages
and messages[-1].get("role") == "assistant"
and not messages[-1].get("tool_calls")
):
return
messages.append(build_assistant_message(_PERSISTED_MODEL_ERROR_PLACEHOLDER))
@@ -827,12 +878,15 @@ class AgentRunner:
insert_at = assistant_idx + 1 + offset
while insert_at < len(updated) and updated[insert_at].get("role") == "tool":
insert_at += 1
updated.insert(insert_at, {
"role": "tool",
"tool_call_id": call_id,
"name": name,
"content": _BACKFILL_CONTENT,
})
updated.insert(
insert_at,
{
"role": "tool",
"tool_call_id": call_id,
"name": name,
"content": _BACKFILL_CONTENT,
},
)
offset += 1
return updated
@@ -891,9 +945,13 @@ class AgentRunner:
if not messages or not spec.context_window_tokens:
return messages
provider_max_tokens = getattr(getattr(self.provider, "generation", None), "max_tokens", 4096)
max_output = spec.max_tokens if isinstance(spec.max_tokens, int) else (
provider_max_tokens if isinstance(provider_max_tokens, int) else 4096
provider_max_tokens = getattr(
getattr(self.provider, "generation", None), "max_tokens", 4096
)
max_output = (
spec.max_tokens
if isinstance(spec.max_tokens, int)
else (provider_max_tokens if isinstance(provider_max_tokens, int) else 4096)
)
budget = spec.context_block_limit or (
spec.context_window_tokens - max_output - _SNIP_SAFETY_BUFFER
@@ -976,4 +1034,3 @@ class AgentRunner:
if current:
batches.append(current)
return batches