Changes
diff --git a/CMakeLists.txt b/CMakeLists.txt
index ed3ce40..89cde78 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -73,6 +73,7 @@ add_library(agent_smith ${AGENT_SMITH_LIBRARY_TYPE}
src/agent.cpp
src/detail/blocking_pool.cpp
src/detail/timer_service.cpp
+ src/detail/tool_dispatch.cpp
src/llm_client.cpp
src/message.cpp
src/runtime.cpp
diff --git a/designs/design-0-framework.md b/designs/design-0-framework.md
index 956968a..37fc812 100644
--- a/designs/design-0-framework.md
+++ b/designs/design-0-framework.md
@@ -448,7 +448,7 @@ types in the Agent Smith public API.
Retry backoff and timeout races use `Runtime::sleepFor`; they must not
block a coroutine thread with `std::this_thread::sleep_for`.
-An internal helper in `src/detail/timeout.hpp` provides the common race
+An internal helper in `include/agent_smith/detail/timeout.hpp` provides the common race
logic:
```cpp
@@ -1234,6 +1234,14 @@ tool call still abort the loop: LLM transport/API errors, malformed LLM
responses, memory snapshot errors, whole-agent cancellation/deadline,
and internal invariant violations.
+Milestone 8 exposes this batch operation internally as `dispatchTools`
+for the later agent loop. A timed-out invocation retains its tool and
+coroutine state until it finishes. The runtime drains these operations
+before destroying HTTP, worker, or timer services; non-cooperative tools
+can therefore delay runtime shutdown. Timers are cancelled when the tool
+finishes first. Result truncation keeps a UTF-8 boundary and appends the
+marker after the configured content-byte allowance.
+
`ToolMessage::is_error` is internal. When serializing to the
OpenAI-compatible wire format, emit only standard fields:
diff --git a/include/agent_smith/agent.hpp b/include/agent_smith/agent.hpp
index bfeb347..8ed8d91 100644
--- a/include/agent_smith/agent.hpp
+++ b/include/agent_smith/agent.hpp
@@ -1,8 +1,20 @@
#pragma once
+#include <chrono>
+#include <cstddef>
+
namespace agent_smith
{
+/// Agent execution limits used by tool dispatch.
+struct AgentConfig
+{
+ /// Default per-call budget when a tool has no override.
+ std::chrono::milliseconds default_tool_timeout{30'000};
+ /// Maximum retained content bytes, excluding the truncation marker.
+ std::size_t tool_result_max_bytes = 64 * 1024;
+};
+
/// Placeholder for the agent loop, implemented in later milestones.
class Agent
{
diff --git a/include/agent_smith/detail/background_tasks.hpp b/include/agent_smith/detail/background_tasks.hpp
new file mode 100644
index 0000000..d176543
--- /dev/null
+++ b/include/agent_smith/detail/background_tasks.hpp
@@ -0,0 +1,18 @@
+#pragma once
+
+#include "../runtime.hpp"
+
+namespace agent_smith::internal
+{
+
+/// Keeps timeout race participants alive until they finish.
+struct BackgroundTasks
+{
+ /// Start an operation and transfer its lifetime to the runtime.
+ static void start(Runtime& runtime, mw::Task<mw::E<void>> task)
+ {
+ runtime.retain(StartedTask<void>::start(std::move(task)));
+ }
+};
+
+} // namespace agent_smith::internal
diff --git a/include/agent_smith/detail/started_task.hpp b/include/agent_smith/detail/started_task.hpp
index cf98a77..4cc281a 100644
--- a/include/agent_smith/detail/started_task.hpp
+++ b/include/agent_smith/detail/started_task.hpp
@@ -116,6 +116,13 @@ public:
co_return co_await std::move(task);
}
+ /// Check completion without consuming the result.
+ bool complete() const
+ {
+ std::lock_guard lock(handle_.promise().mutex);
+ return handle_.promise().complete;
+ }
+
auto operator co_await() &&
{
struct Awaiter
@@ -168,7 +175,7 @@ private:
return;
}
- if(!handle_.done())
+ if(!complete())
{
std::terminate();
}
diff --git a/include/agent_smith/detail/timeout.hpp b/include/agent_smith/detail/timeout.hpp
new file mode 100644
index 0000000..34d298d
--- /dev/null
+++ b/include/agent_smith/detail/timeout.hpp
@@ -0,0 +1,93 @@
+#pragma once
+
+#include <atomic>
+#include <memory>
+#include <stop_token>
+
+#include "async_queue.hpp"
+#include "background_tasks.hpp"
+#include "../cancellation.hpp"
+#include "../error.hpp"
+
+namespace agent_smith::internal
+{
+
+template <typename T> struct TimeoutState
+{
+ std::atomic<bool> settled{false};
+ AsyncQueue<mw::E<T>> result;
+ std::stop_source timer_stop;
+ CancellationSource cancel;
+};
+
+template <typename T>
+mw::Task<mw::E<void>> finishOperation(
+ mw::Task<mw::E<T>> task, std::shared_ptr<TimeoutState<T>> state)
+{
+ auto result = co_await std::move(task);
+ if(!state->settled.exchange(true))
+ {
+ state->timer_stop.request_stop();
+ state->result.push(std::move(result));
+ }
+ co_return mw::E<void>{};
+}
+
+template <typename T>
+mw::Task<mw::E<void>> finishTimeout(
+ Runtime& runtime, std::chrono::milliseconds timeout,
+ std::shared_ptr<TimeoutState<T>> state)
+{
+ auto slept = co_await runtime.sleepFor(
+ timeout, state->timer_stop.get_token());
+ if(!state->settled.exchange(true))
+ {
+ state->cancel.cancel();
+ if(!slept)
+ {
+ state->result.push(std::unexpected(slept.error()));
+ }
+ else
+ {
+ state->result.push(std::unexpected(mw::Error(TimeoutError{
+ .msg = "tool execution timed out",
+ .operation = "tool invocation",
+ .budget = timeout,
+ })));
+ }
+ }
+ co_return mw::E<void>{};
+}
+
+/// Race an operation against a timer, retaining the loser in the runtime.
+/// The task must own everything it uses after a timeout. The runtime must
+/// outlive this call; shutdown drains non-cooperative operations.
+template <typename T>
+mw::Task<mw::E<T>> withTimeout(
+ Runtime& runtime, mw::Task<mw::E<T>> task,
+ std::chrono::milliseconds timeout, CancellationSource* cancel = nullptr)
+{
+ if(timeout <= std::chrono::milliseconds::zero())
+ {
+ if(cancel != nullptr)
+ {
+ cancel->cancel();
+ }
+ co_return std::unexpected(mw::Error(TimeoutError{
+ .msg = "tool execution timed out",
+ .operation = "tool invocation",
+ .budget = timeout,
+ }));
+ }
+ auto state = std::make_shared<TimeoutState<T>>();
+ if(cancel != nullptr)
+ {
+ state->cancel = *cancel;
+ }
+ BackgroundTasks::start(runtime, finishOperation(std::move(task), state));
+ BackgroundTasks::start(runtime, finishTimeout(runtime, timeout, state));
+ auto result = co_await state->result.pop();
+ co_return std::move(*result);
+}
+
+} // namespace agent_smith::internal
diff --git a/include/agent_smith/detail/tool_dispatch.hpp b/include/agent_smith/detail/tool_dispatch.hpp
new file mode 100644
index 0000000..4b0c278
--- /dev/null
+++ b/include/agent_smith/detail/tool_dispatch.hpp
@@ -0,0 +1,16 @@
+#pragma once
+
+#include "../agent.hpp"
+#include "../message.hpp"
+#include "../tool.hpp"
+
+namespace agent_smith::internal
+{
+
+/// Run every call concurrently and collect tool messages in input order.
+/// Timed-out calls retain their tool instance until execution completes.
+mw::Task<mw::E<std::vector<ToolMessage>>> dispatchTools(
+ Runtime& runtime, ToolRegistry& registry, std::vector<ToolCall> calls,
+ AgentConfig config = {});
+
+} // namespace agent_smith::internal
diff --git a/include/agent_smith/error.hpp b/include/agent_smith/error.hpp
index 6cea54f..6540139 100644
--- a/include/agent_smith/error.hpp
+++ b/include/agent_smith/error.hpp
@@ -1,6 +1,7 @@
#pragma once
#include <cstddef>
+#include <chrono>
#include <string>
#include <string_view>
@@ -27,6 +28,17 @@ struct LlmApiError
std::string body_excerpt;
};
+/// An operation exceeded its execution-time budget.
+struct TimeoutError
+{
+ /// Human-readable failure description.
+ std::string msg;
+ /// Name of the operation that timed out.
+ std::string operation;
+ /// Maximum duration allowed for the operation.
+ std::chrono::milliseconds budget;
+};
+
namespace error
{
diff --git a/include/agent_smith/runtime.hpp b/include/agent_smith/runtime.hpp
index b687f8c..74d2e31 100644
--- a/include/agent_smith/runtime.hpp
+++ b/include/agent_smith/runtime.hpp
@@ -3,15 +3,20 @@
#include <chrono>
#include <cstddef>
#include <memory>
+#include <mutex>
+#include <stop_token>
+#include <vector>
#include <type_traits>
#include <mw/http_client_async.hpp>
#include "detail/blocking_pool.hpp"
+#include "detail/started_task.hpp"
namespace agent_smith::internal
{
class TimerService;
+struct BackgroundTasks;
}
namespace agent_smith
@@ -37,7 +42,8 @@ public:
/// Construct a runtime using the supplied shared-service configuration.
explicit Runtime(RuntimeConfig cfg = {});
- /// Stop the HTTP session and release runtime resources.
+ /// Drain retained tool work, then release runtime resources.
+ /// A non-cooperative timed-out tool can delay shutdown until it finishes.
~Runtime();
Runtime(const Runtime&) = delete;
@@ -56,11 +62,21 @@ public:
/// Suspend without blocking the caller's coroutine thread.
mw::Task<mw::E<void>> sleepFor(std::chrono::milliseconds duration);
+ /// Suspend until the duration elapses or stop is requested.
+ /// An early wake is successful; inspect the token to distinguish it.
+ mw::Task<mw::E<void>> sleepFor(std::chrono::milliseconds duration,
+ std::stop_token stop);
+
private:
+ friend struct internal::BackgroundTasks;
+ void retain(internal::StartedTask<void> task);
+
RuntimeConfig cfg_;
std::unique_ptr<mw::HTTPSessionAsync> http_;
std::unique_ptr<internal::BlockingPool> blocking_pool_;
std::unique_ptr<internal::TimerService> timer_service_;
+ std::mutex background_mutex_;
+ std::vector<internal::StartedTask<void>> background_tasks_;
};
template <typename F>
diff --git a/include/agent_smith/tool.hpp b/include/agent_smith/tool.hpp
index 1c79a05..a996dd9 100644
--- a/include/agent_smith/tool.hpp
+++ b/include/agent_smith/tool.hpp
@@ -20,6 +20,11 @@
namespace agent_smith
{
+namespace internal
+{
+struct ToolAccess;
+}
+
/// OpenAI-compatible schema and presentation metadata for one tool.
struct ToolSchema
{
@@ -253,6 +258,7 @@ std::unique_ptr<Tool> makeTool(ToolSchema schema, F function)
}
/// Owns named tools and their options in deterministic insertion order.
+/// Registry access is not thread-safe. Running calls keep replaced tools alive.
class ToolRegistry
{
public:
@@ -275,9 +281,11 @@ public:
std::size_t size() const;
private:
+ friend struct internal::ToolAccess;
struct Entry
{
- std::unique_ptr<Tool> tool;
+ // In-flight invocations retain ownership after timeout or replacement.
+ std::shared_ptr<Tool> tool;
ToolOptions options;
};
diff --git a/src/detail/timer_service.cpp b/src/detail/timer_service.cpp
index 021f166..5f69bf1 100644
--- a/src/detail/timer_service.cpp
+++ b/src/detail/timer_service.cpp
@@ -1,4 +1,5 @@
#include "timer_service.hpp"
+#include <algorithm>
namespace agent_smith::internal
{
@@ -10,19 +11,31 @@ TimerService::TimerService()
TimerService::~TimerService()
{
- worker_.request_stop();
+ {
+ std::lock_guard lock(mutex_);
+ worker_.request_stop();
+ }
condition_.notify_all();
}
TimerService::SleepAwaiter::SleepAwaiter(TimerService& service,
- std::chrono::milliseconds duration)
+ std::chrono::milliseconds duration,
+ std::stop_token stop)
: service_(&service), duration_(duration), entry_(std::make_shared<Entry>())
{
+ entry_->stop = stop;
+ callback_ = std::make_unique<std::stop_callback<std::function<void()>>>(
+ stop, [&service]()
+ {
+ std::lock_guard lock(service.mutex_);
+ service.condition_.notify_one();
+ });
}
bool TimerService::SleepAwaiter::await_ready() const
{
- return duration_ <= std::chrono::milliseconds::zero();
+ return duration_ <= std::chrono::milliseconds::zero() ||
+ entry_->stop.stop_requested();
}
bool TimerService::SleepAwaiter::await_suspend(
@@ -34,9 +47,10 @@ bool TimerService::SleepAwaiter::await_suspend(
}
TimerService::SleepAwaiter
-TimerService::sleepFor(std::chrono::milliseconds duration)
+TimerService::sleepFor(std::chrono::milliseconds duration,
+ std::stop_token stop)
{
- return SleepAwaiter(*this, duration);
+ return SleepAwaiter(*this, duration, stop);
}
void TimerService::schedule(const std::shared_ptr<Entry>& entry,
@@ -45,7 +59,7 @@ void TimerService::schedule(const std::shared_ptr<Entry>& entry,
{
std::lock_guard lock(mutex_);
entry->continuation = continuation;
- entries_.push(entry);
+ entries_.push_back(entry);
}
condition_.notify_one();
}
@@ -63,20 +77,24 @@ void TimerService::worker(std::stop_token stop)
continue;
}
- const auto deadline = entries_.top()->deadline;
- if(condition_.wait_until(lock, deadline,
- [&]()
- {
- return stop.stop_requested() ||
- entries_.empty() ||
- entries_.top()->deadline < deadline;
- }))
+ const auto now = std::chrono::steady_clock::now();
+ auto ready = std::find_if(entries_.begin(), entries_.end(),
+ [now](const auto& entry)
+ {
+ return entry->stop.stop_requested() || entry->deadline <= now;
+ });
+ if(ready == entries_.end())
{
+ const auto earliest = std::min_element(
+ entries_.begin(), entries_.end(),
+ [](const auto& lhs, const auto& rhs)
+ { return lhs->deadline < rhs->deadline; });
+ condition_.wait_until(lock, (*earliest)->deadline);
continue;
}
- auto entry = entries_.top();
- entries_.pop();
+ auto entry = *ready;
+ entries_.erase(ready);
auto continuation = entry->continuation;
lock.unlock();
if(continuation != nullptr)
diff --git a/src/detail/timer_service.hpp b/src/detail/timer_service.hpp
index 12bd2c2..24a8301 100644
--- a/src/detail/timer_service.hpp
+++ b/src/detail/timer_service.hpp
@@ -6,7 +6,8 @@
#include <cstddef>
#include <memory>
#include <mutex>
-#include <queue>
+#include <functional>
+#include <stop_token>
#include <thread>
#include <vector>
@@ -28,7 +29,8 @@ public:
class SleepAwaiter
{
public:
- SleepAwaiter(TimerService& service, std::chrono::milliseconds duration);
+ SleepAwaiter(TimerService& service, std::chrono::milliseconds duration,
+ std::stop_token stop);
bool await_ready() const;
bool await_suspend(std::coroutine_handle<> continuation);
@@ -38,24 +40,18 @@ public:
TimerService* service_;
std::chrono::milliseconds duration_;
std::shared_ptr<Entry> entry_;
+ std::unique_ptr<std::stop_callback<std::function<void()>>> callback_;
};
- SleepAwaiter sleepFor(std::chrono::milliseconds duration);
+ SleepAwaiter sleepFor(std::chrono::milliseconds duration,
+ std::stop_token stop = {});
private:
struct Entry
{
std::chrono::steady_clock::time_point deadline;
std::coroutine_handle<> continuation = nullptr;
- };
-
- struct EntryLater
- {
- bool operator()(const std::shared_ptr<Entry>& lhs,
- const std::shared_ptr<Entry>& rhs) const
- {
- return lhs->deadline > rhs->deadline;
- }
+ std::stop_token stop;
};
void schedule(const std::shared_ptr<Entry>& entry,
@@ -64,10 +60,7 @@ private:
std::mutex mutex_;
std::condition_variable condition_;
- std::priority_queue<std::shared_ptr<Entry>,
- std::vector<std::shared_ptr<Entry>>,
- EntryLater>
- entries_;
+ std::vector<std::shared_ptr<Entry>> entries_;
std::jthread worker_;
};
diff --git a/src/detail/tool_dispatch.cpp b/src/detail/tool_dispatch.cpp
new file mode 100644
index 0000000..fd17f71
--- /dev/null
+++ b/src/detail/tool_dispatch.cpp
@@ -0,0 +1,106 @@
+#include "agent_smith/detail/tool_dispatch.hpp"
+
+#include "agent_smith/detail/gather_all.hpp"
+#include "agent_smith/detail/timeout.hpp"
+
+namespace agent_smith::internal
+{
+
+struct ToolAccess
+{
+ static std::shared_ptr<Tool> retain(ToolRegistry& registry,
+ const std::string& name)
+ {
+ const auto found = registry.by_name_.find(name);
+ return found == registry.by_name_.end() ? nullptr : found->second.tool;
+ }
+};
+
+namespace
+{
+
+mw::Task<mw::E<ToolResult>> invokeOwned(std::shared_ptr<Tool> tool,
+ nlohmann::json arguments,
+ ToolContext context)
+{
+ co_return co_await tool->invoke(std::move(arguments), std::move(context));
+}
+
+mw::Task<mw::E<ToolResult>> unknownTool(std::string name)
+{
+ co_return std::unexpected(mw::runtimeError("unknown tool: " + name));
+}
+
+std::string errorContent(const std::string& message)
+{
+ return nlohmann::json({{"error", true}, {"message", message}}).dump();
+}
+
+} // namespace
+
+mw::Task<mw::E<std::vector<ToolMessage>>> dispatchTools(
+ Runtime& runtime, ToolRegistry& registry, std::vector<ToolCall> calls,
+ AgentConfig config)
+{
+ std::vector<StartedTask<ToolResult>> started;
+ started.reserve(calls.size());
+ for(const auto& call : calls)
+ {
+ auto tool = ToolAccess::retain(registry, call.name);
+ if(tool == nullptr)
+ {
+ started.push_back(StartedTask<ToolResult>::start(
+ unknownTool(call.name)));
+ continue;
+ }
+ const auto budget = registry.options(call.name).timeout.value_or(
+ config.default_tool_timeout);
+ CancellationSource cancel;
+ ToolContext context{
+ .cancel = cancel.token(),
+ .budget = budget,
+ .call_id = call.id,
+ .logger = spdlog::default_logger_raw(),
+ .runtime = &runtime,
+ };
+ started.push_back(StartedTask<ToolResult>::start(withTimeout(
+ runtime, invokeOwned(std::move(tool), call.arguments,
+ std::move(context)),
+ budget, &cancel)));
+ }
+
+ auto results = co_await gatherAllStarted(std::move(started));
+ std::vector<ToolMessage> messages;
+ messages.reserve(calls.size());
+ for(std::size_t index = 0; index < calls.size(); ++index)
+ {
+ auto& result = results[index];
+ const bool is_error = !result || result->is_error;
+ std::string content = !result ? errorContent(result.error().msg())
+ : result->is_error ? errorContent(result->content)
+ : std::move(result->content);
+ if(content.size() > config.tool_result_max_bytes)
+ {
+ const auto original_size = content.size();
+ auto end = config.tool_result_max_bytes;
+ // Do not split a UTF-8 code point in a serializable message.
+ while(end > 0 &&
+ (static_cast<unsigned char>(content[end]) & 0xc0) == 0x80)
+ {
+ --end;
+ }
+ content.resize(end);
+ content += "\n…[truncated, original size " +
+ std::to_string(original_size) + " bytes]";
+ }
+ messages.push_back(ToolMessage{
+ .tool_call_id = calls[index].id,
+ .name = calls[index].name,
+ .content = std::move(content),
+ .is_error = is_error,
+ });
+ }
+ co_return messages;
+}
+
+} // namespace agent_smith::internal
diff --git a/src/runtime.cpp b/src/runtime.cpp
index 4310ca4..5db25ab 100644
--- a/src/runtime.cpp
+++ b/src/runtime.cpp
@@ -1,6 +1,9 @@
#include "agent_smith/runtime.hpp"
#include "detail/timer_service.hpp"
+#include "agent_smith/detail/gather_all.hpp"
+#include "agent_smith/sync_wait.hpp"
+#include <algorithm>
namespace agent_smith
{
@@ -20,15 +23,45 @@ Runtime::Runtime(RuntimeConfig cfg)
Runtime::~Runtime()
{
+ // Keep HTTP, workers, and timers alive until timed-out work finishes.
+ // A background operation may itself register additional work.
+ for(;;)
+ {
+ std::vector<internal::StartedTask<void>> pending;
+ {
+ std::lock_guard lock(background_mutex_);
+ pending.swap(background_tasks_);
+ }
+ if(pending.empty())
+ {
+ break;
+ }
+ [[maybe_unused]] auto results =
+ syncWait(internal::gatherAllStarted(std::move(pending)));
+ }
http_.reset();
}
+void Runtime::retain(internal::StartedTask<void> task)
+{
+ std::lock_guard lock(background_mutex_);
+ std::erase_if(background_tasks_,
+ [](const auto& item) { return item.complete(); });
+ background_tasks_.push_back(std::move(task));
+}
+
mw::HTTPSessionAsync& Runtime::http()
{
return *http_;
}
mw::Task<mw::E<void>> Runtime::sleepFor(std::chrono::milliseconds duration)
+{
+ co_return co_await sleepFor(duration, {});
+}
+
+mw::Task<mw::E<void>> Runtime::sleepFor(std::chrono::milliseconds duration,
+ std::stop_token stop)
{
if(duration < std::chrono::milliseconds::zero())
{
@@ -36,7 +69,7 @@ mw::Task<mw::E<void>> Runtime::sleepFor(std::chrono::milliseconds duration)
mw::runtimeError("sleep duration cannot be negative"));
}
- co_await timer_service_->sleepFor(duration);
+ co_await timer_service_->sleepFor(duration, stop);
co_return mw::E<void>{};
}
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index d9df8a4..7a52c86 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -4,6 +4,7 @@ add_executable(agent_smith_test_skeleton
test_sse_parser.cpp
test_skeleton.cpp
test_tool.cpp
+ test_tool_dispatch.cpp
)
target_link_libraries(agent_smith_test_skeleton
PRIVATE
diff --git a/tests/test_tool_dispatch.cpp b/tests/test_tool_dispatch.cpp
new file mode 100644
index 0000000..bdbeef1
--- /dev/null
+++ b/tests/test_tool_dispatch.cpp
@@ -0,0 +1,275 @@
+#include <atomic>
+#include <chrono>
+#include <thread>
+
+#include <agent_smith/detail/tool_dispatch.hpp>
+#include <agent_smith/detail/timeout.hpp>
+#include <agent_smith/sync_wait.hpp>
+#include <gtest/gtest.h>
+
+namespace agent_smith
+{
+namespace
+{
+using namespace std::chrono_literals;
+
+ToolSchema testSchema(std::string name)
+{
+ return {std::move(name), "dispatch test",
+ {{"type", "object"}, {"properties", nlohmann::json::object()}}};
+}
+
+mw::Task<mw::E<ToolResult>> rendezvous(
+ std::atomic<int>& started, nlohmann::json args, ToolContext context)
+{
+ ++started;
+ while(started.load() < 2 && !context.cancel.cancelled())
+ {
+ co_await context.runtime->sleepFor(1ms);
+ }
+ if(context.cancel.cancelled())
+ {
+ co_return std::unexpected(mw::runtimeError("not concurrent"));
+ }
+ co_await context.runtime->sleepFor(
+ std::chrono::milliseconds(args.at("delay").get<int>()));
+ co_return ToolResult{.content = context.call_id};
+}
+
+TEST(ToolDispatchTest, StartsAllCallsAndPreservesInputOrder)
+{
+ Runtime runtime;
+ ToolRegistry registry;
+ std::atomic<int> started{0};
+ registry.add(makeTool(testSchema("parallel"),
+ [&started](nlohmann::json args, ToolContext context)
+ { return rendezvous(started, std::move(args), std::move(context)); }));
+ auto result = syncWait(internal::dispatchTools(runtime, registry,
+ {{"first", "parallel", {{"delay", 30}}},
+ {"second", "parallel", {{"delay", 0}}}},
+ {.default_tool_timeout = 1s}));
+ ASSERT_TRUE(result);
+ ASSERT_EQ(result->size(), 2U);
+ EXPECT_FALSE((*result)[0].is_error);
+ EXPECT_FALSE((*result)[1].is_error);
+ EXPECT_EQ((*result)[0].content, "first");
+ EXPECT_EQ((*result)[1].content, "second");
+}
+
+TEST(ToolDispatchTest, ConvertsFailuresAndTruncatesResults)
+{
+ Runtime runtime;
+ ToolRegistry registry;
+ registry.add(makeTool(testSchema("failure"),
+ [](nlohmann::json, ToolContext) -> mw::E<ToolResult>
+ { return std::unexpected(mw::runtimeError("failed")); }));
+ registry.add(makeTool(testSchema("reported"),
+ [](nlohmann::json, ToolContext)
+ { return ToolResult{.content = "reported failure", .is_error = true}; }));
+ registry.add(makeTool(testSchema("large"),
+ [](nlohmann::json, ToolContext) { return std::string(100, 'x'); }));
+ auto result = syncWait(internal::dispatchTools(runtime, registry,
+ {{"a", "missing", {}}, {"b", "failure", {}},
+ {"c", "reported", {}}, {"d", "large", {}}},
+ {.tool_result_max_bytes = 64}));
+ ASSERT_TRUE(result);
+ ASSERT_EQ(result->size(), 4U);
+ for(int index = 0; index < 3; ++index)
+ {
+ EXPECT_TRUE((*result)[index].is_error);
+ EXPECT_TRUE(nlohmann::json::parse((*result)[index].content).at("error"));
+ }
+ EXPECT_EQ(nlohmann::json::parse((*result)[0].content).at("message"),
+ "unknown tool: missing");
+ EXPECT_EQ((*result)[3].content,
+ std::string(64, 'x') + "\n…[truncated, original size 100 bytes]");
+}
+
+mw::Task<mw::E<ToolResult>> waitForCancellation(
+ std::atomic<bool>& cancelled, nlohmann::json, ToolContext context)
+{
+ while(!context.cancel.cancelled())
+ {
+ co_await context.runtime->sleepFor(1ms);
+ }
+ cancelled = true;
+ co_return ToolResult{.content = "late"};
+}
+
+TEST(ToolDispatchTest, AppliesOverrideAndSignalsCancellation)
+{
+ std::atomic<bool> cancelled{false};
+ {
+ Runtime runtime;
+ ToolRegistry registry;
+ registry.add(makeTool(testSchema("slow"),
+ [&cancelled](nlohmann::json args, ToolContext context)
+ { return waitForCancellation(cancelled, std::move(args),
+ std::move(context)); }),
+ {.timeout = 20ms});
+ const auto start = std::chrono::steady_clock::now();
+ auto result = syncWait(internal::dispatchTools(runtime, registry,
+ {{"call", "slow", {}}}, {.default_tool_timeout = 5s}));
+ ASSERT_TRUE(result);
+ ASSERT_EQ(result->size(), 1U);
+ EXPECT_TRUE(result->front().is_error);
+ EXPECT_EQ(result->front().tool_call_id, "call");
+ EXPECT_LT(std::chrono::steady_clock::now() - start, 1s);
+ }
+ EXPECT_TRUE(cancelled);
+}
+
+mw::Task<mw::E<int>> delayedValue(Runtime& runtime)
+{
+ co_await runtime.sleepFor(40ms);
+ co_return 42;
+}
+
+TEST(TimeoutTest, ReturnsTypedErrorAndRetainsLateOperation)
+{
+ Runtime runtime;
+ CancellationSource cancel;
+ auto result = syncWait(internal::withTimeout(
+ runtime, delayedValue(runtime), 5ms, &cancel));
+ ASSERT_FALSE(result);
+ ASSERT_NE(result.error().as<TimeoutError>(), nullptr);
+ EXPECT_EQ(result.error().as<TimeoutError>()->budget, 5ms);
+ EXPECT_TRUE(cancel.token().cancelled());
+}
+
+TEST(TimeoutTest, CancelsLosingTimerPromptly)
+{
+ const auto start = std::chrono::steady_clock::now();
+ {
+ Runtime runtime;
+ auto result = syncWait(internal::withTimeout(
+ runtime, delayedValue(runtime), 30s));
+ ASSERT_TRUE(result);
+ EXPECT_EQ(*result, 42);
+ }
+ EXPECT_LT(std::chrono::steady_clock::now() - start, 2s);
+}
+
+TEST(ToolDispatchTest, SyncCallsRunConcurrentlyAndUseDefaultBudget)
+{
+ Runtime runtime;
+ ToolRegistry registry;
+ std::atomic<int> started{0};
+ registry.add(makeTool(testSchema("sync"),
+ [&started](nlohmann::json, ToolContext context) -> ToolResult
+ {
+ ++started;
+ while(started.load() < 2 && !context.cancel.cancelled())
+ {
+ std::this_thread::yield();
+ }
+ return {.content = std::to_string(context.budget.count()),
+ .is_error = context.cancel.cancelled()};
+ }));
+ auto result = syncWait(internal::dispatchTools(runtime, registry,
+ {{"a", "sync", {}}, {"b", "sync", {}}},
+ {.default_tool_timeout = 500ms}));
+ ASSERT_TRUE(result);
+ ASSERT_EQ(result->size(), 2U);
+ EXPECT_FALSE((*result)[0].is_error);
+ EXPECT_FALSE((*result)[1].is_error);
+ EXPECT_EQ((*result)[0].content, "500");
+}
+
+class LateTool final : public Tool
+{
+public:
+ LateTool(std::atomic<bool>& destroyed, std::atomic<bool>& finished)
+ : destroyed_(destroyed), finished_(finished), schema_(testSchema("late"))
+ {}
+ ~LateTool() override { destroyed_ = true; }
+ const ToolSchema& schema() const override { return schema_; }
+ mw::Task<mw::E<ToolResult>> invoke(nlohmann::json,
+ ToolContext context) override
+ {
+ co_await context.runtime->sleepFor(80ms);
+ finished_ = true;
+ co_return ToolResult{.content = schema_.name};
+ }
+private:
+ std::atomic<bool>& destroyed_;
+ std::atomic<bool>& finished_;
+ ToolSchema schema_;
+};
+
+TEST(ToolDispatchTest, RetainsTimedOutToolAfterRegistryDestruction)
+{
+ std::atomic<bool> destroyed{false};
+ std::atomic<bool> finished{false};
+ {
+ Runtime runtime;
+ {
+ ToolRegistry registry;
+ registry.add(std::make_unique<LateTool>(destroyed, finished));
+ auto result = syncWait(internal::dispatchTools(runtime, registry,
+ {{"a", "late", {}}}, {.default_tool_timeout = 5ms}));
+ ASSERT_TRUE(result);
+ EXPECT_TRUE(result->front().is_error);
+ }
+ EXPECT_FALSE(destroyed);
+ }
+ EXPECT_TRUE(finished);
+ EXPECT_TRUE(destroyed);
+}
+
+TEST(ToolDispatchTest, HandlesEmptyBatchAndZeroBudget)
+{
+ Runtime runtime;
+ ToolRegistry registry;
+ auto empty = syncWait(internal::dispatchTools(runtime, registry, {}));
+ ASSERT_TRUE(empty);
+ EXPECT_TRUE(empty->empty());
+ std::atomic<bool> invoked{false};
+ registry.add(makeTool(testSchema("zero"),
+ [&invoked](nlohmann::json, ToolContext)
+ { invoked = true; return "unexpected"; }));
+ auto result = syncWait(internal::dispatchTools(runtime, registry,
+ {{"a", "zero", {}}}, {.default_tool_timeout = 0ms}));
+ ASSERT_TRUE(result);
+ EXPECT_TRUE(result->front().is_error);
+ EXPECT_FALSE(invoked);
+}
+
+TEST(ToolDispatchTest, TruncationKeepsUtf8Serializable)
+{
+ Runtime runtime;
+ ToolRegistry registry;
+ registry.add(makeTool(testSchema("unicode"),
+ [](nlohmann::json, ToolContext) { return "北京天气"; }));
+ auto result = syncWait(internal::dispatchTools(runtime, registry,
+ {{"a", "unicode", {}}}, {.tool_result_max_bytes = 4}));
+ ASSERT_TRUE(result);
+ EXPECT_EQ(result->front().content,
+ "北\n…[truncated, original size 12 bytes]");
+ EXPECT_NO_THROW(toOpenAIJson(Message(result->front())).dump());
+}
+
+TEST(ToolDispatchTest, SyncTimeoutReturnsBeforeNonCooperativeWorkFinishes)
+{
+ std::atomic<bool> finished{false};
+ {
+ Runtime runtime;
+ ToolRegistry registry;
+ registry.add(makeTool(testSchema("blocking"),
+ [&finished](nlohmann::json, ToolContext)
+ {
+ std::this_thread::sleep_for(200ms);
+ finished = true;
+ return "late";
+ }));
+ auto result = syncWait(internal::dispatchTools(runtime, registry,
+ {{"a", "blocking", {}}}, {.default_tool_timeout = 10ms}));
+ ASSERT_TRUE(result);
+ EXPECT_TRUE(result->front().is_error);
+ EXPECT_FALSE(finished);
+ }
+ EXPECT_TRUE(finished);
+}
+
+} // namespace
+} // namespace agent_smith