Changes
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 5e84c06..5c3b6c9 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -10,6 +10,35 @@ set(CMAKE_CXX_EXTENSIONS OFF)
option(AGENT_SMITH_BUILD_SHARED "Build agent_smith as a shared library" OFF)
option(AGENT_SMITH_BUILD_DEMO "Build the Agent Smith demo" OFF)
option(AGENT_SMITH_BUILD_TESTS "Build Agent Smith tests" OFF)
+option(AGENT_SMITH_TEST_ASAN "Enable AddressSanitizer for project tests" OFF)
+option(AGENT_SMITH_TEST_TSAN "Enable ThreadSanitizer for project tests" OFF)
+
+if(AGENT_SMITH_TEST_ASAN AND AGENT_SMITH_TEST_TSAN)
+ message(FATAL_ERROR
+ "AddressSanitizer and ThreadSanitizer must be enabled separately.")
+endif()
+
+if((AGENT_SMITH_TEST_ASAN OR AGENT_SMITH_TEST_TSAN) AND
+ NOT AGENT_SMITH_BUILD_TESTS)
+ message(FATAL_ERROR
+ "Sanitizers require AGENT_SMITH_BUILD_TESTS=ON.")
+endif()
+
+function(agent_smith_enable_test_sanitizer target)
+ if(AGENT_SMITH_TEST_ASAN)
+ target_compile_options(${target} PRIVATE
+ -fsanitize=address
+ -fno-omit-frame-pointer
+ )
+ target_link_options(${target} PRIVATE -fsanitize=address)
+ elseif(AGENT_SMITH_TEST_TSAN)
+ target_compile_options(${target} PRIVATE
+ -fsanitize=thread
+ -fno-omit-frame-pointer
+ )
+ target_link_options(${target} PRIVATE -fsanitize=thread)
+ endif()
+endfunction()
include(FetchContent)
@@ -42,6 +71,8 @@ endif()
add_library(agent_smith ${AGENT_SMITH_LIBRARY_TYPE}
src/agent.cpp
+ src/detail/blocking_pool.cpp
+ src/detail/timer_service.cpp
src/runtime.cpp
)
add_library(agent_smith::agent_smith ALIAS agent_smith)
@@ -62,9 +93,11 @@ target_link_libraries(agent_smith
target_compile_features(agent_smith PUBLIC cxx_std_23)
target_compile_options(agent_smith PRIVATE -Wall -Wextra -Wpedantic)
+agent_smith_enable_test_sanitizer(agent_smith)
if(AGENT_SMITH_BUILD_DEMO)
add_subdirectory(demo)
+ agent_smith_enable_test_sanitizer(agent_smith_demo)
endif()
if(AGENT_SMITH_BUILD_TESTS)
diff --git a/designs/design-0-framework.md b/designs/design-0-framework.md
index 9145a50..74769ca 100644
--- a/designs/design-0-framework.md
+++ b/designs/design-0-framework.md
@@ -361,7 +361,7 @@ not flattened into a single "run this closure on a worker" mechanism.
HTTP uses `mw::HTTPSessionAsync` directly because it has streaming,
connection, timeout, and cancellation semantics that are better handled
by libmw's curl driver. Only operations that are inherently blocking at
-the framework layer are lifted into coroutines with `Runtime::runBlocking`.
+the framework layer are lifted into coroutines with `Runtime::runOnWorker`.
### 5.2 The `Runtime` type
@@ -394,7 +394,7 @@ public:
// an awaitable result. Do not use for HTTP.
template <typename F>
mw::Task<detail::ExpectedResult<std::invoke_result_t<F>>>
- runBlocking(F&& f);
+ runOnWorker(F&& f);
// Coroutine-friendly timer used for retry backoff and internal
// timeout races. Does not expose asio or scheduler-specific types.
@@ -414,12 +414,12 @@ private:
`detail::BlockingPool` is a private utility implemented with
`std::jthread`, `std::mutex`, `std::condition_variable`, and a FIFO work
queue. Subsystems do not access it directly; they call
-`Runtime::runBlocking`:
+`Runtime::runOnWorker`:
```cpp
template <typename F>
mw::Task<detail::ExpectedResult<std::invoke_result_t<F>>>
-Runtime::runBlocking(F&& f);
+Runtime::runOnWorker(F&& f);
```
The submitted callable must not throw as part of normal failure. It
@@ -1146,14 +1146,14 @@ makeTool(ToolSchema schema, AsyncToolCallable auto fn);
Implementation: each overload returns a small derived class that
holds `schema`/`fn`. The async overload awaits the callable directly.
The sync overload returns a `SyncToolAdapter` whose `invoke` submits
-the callable to `ctx.runtime->runBlocking()` and awaits that task, so
+the callable to `ctx.runtime->runOnWorker()` and awaits that task, so
blocking user code never runs on the agent coroutine:
```cpp
mw::Task<mw::E<ToolResult>> invoke(nlohmann::json args,
ToolContext ctx) override
{
- co_return co_await ctx.runtime->runBlocking(
+ co_return co_await ctx.runtime->runOnWorker(
[fn = fn_, args = std::move(args), ctx]() mutable
{
return normalizeToolResult(fn(std::move(args), ctx));
@@ -1221,7 +1221,7 @@ In each loop iteration, after the LLM response contains
Async tool callables run as coroutines inside their started task. Sync
tools created through `makeTool` submit their blocking callable to
-`Runtime::runBlocking()` from inside their adapter. `StartedTask`
+`Runtime::runOnWorker()` from inside their adapter. `StartedTask`
remains an internal helper; the public API still exposes only
`mw::Task<mw::E<T>>`.
@@ -1466,12 +1466,12 @@ Implementation outline:
1. `connect()` calls `mw::Process::exec(input_pipe, argv,
output_pipe)` where both pipes are owned by the transport.
-2. A reader task submits blocking reads to `Runtime::runBlocking()`.
+2. A reader task submits blocking reads to `Runtime::runOnWorker()`.
It uses `read(pipe.outputFD(), ...)` in a loop, splits on `\n`
(newline-delimited JSON per the MCP stdio profile), and pushes
parsed `nlohmann::json` frames into an `AsyncQueue<json>`.
3. `send(frame)` serializes the JSON, appends `\n`, and writes via
- `Runtime::runBlocking()`.
+ `Runtime::runOnWorker()`.
4. `close()` closes the pipes and `wait()`s the child.
Caveat: the libmw `Pipe::read()` reads until EOF, which is
@@ -2011,7 +2011,7 @@ public:
* One coroutine drives one `Agent` instance at a time.
* `complete*` and MCP Streamable HTTP use `Runtime::http()` directly.
* Synchronous native tools and MCP stdio operations use
- `Runtime::runBlocking()`.
+ `Runtime::runOnWorker()`.
* Hooks fire synchronously from the agent coroutine. They are
best-effort observers and are expected not to throw. All hook
invocations go through a single internal helper that catches
@@ -2120,7 +2120,7 @@ independent from any particular coroutine scheduler.
* Before each loop iteration (cheap, deterministic).
* Before and after each awaited framework operation.
* Before submitting synchronous native tools or stdio work through
- `Runtime::runBlocking()`.
+ `Runtime::runOnWorker()`.
* Streaming: the HTTP chunk callback returns `false` if cancellation
is observed. The LLM client maps the resulting HTTP abort to
`CancelledError`.
@@ -2222,12 +2222,12 @@ separate one-off runners for each subsystem.
| # | Milestone | Deliverable |
| - | ------------------------------------ | -------------------------------------------- |
| 1 | Skeleton + CMake + libmw integration | `Runtime` builds, empty `Agent` constructs. |
-| 2 | Internal concurrency core | `Runtime::runBlocking`, awaitable `StartedTask`, `AsyncQueue`, timer/sleep, and `gatherAll` tested. |
+| 2 | Internal concurrency core | `Runtime::runOnWorker`, awaitable `StartedTask`, `AsyncQueue`, timer/sleep, and `gatherAll` tested. |
| 3 | Message model + JSON serialization | `tests/test_message.cpp` green. |
| 4 | SSE parser (offline) | `tests/test_sse_parser.cpp` green. |
| 5 | LLM client non-streaming | `tests/test_llm_client.cpp` with `mw::HTTPSessionAsyncMock` green. |
| 6 | LLM client streaming | `StartedTask` + `AsyncQueue` stream events while HTTP is active. |
-| 7 | Tool subsystem | Sync adapters use `Runtime::runBlocking`; async adapters are awaited directly. |
+| 7 | Tool subsystem | Sync adapters use `Runtime::runOnWorker`; async adapters are awaited directly. |
| 8 | Parallel tool dispatch + timeouts | Ordered `gatherAll` results and per-tool timeout tests green. |
| 9 | Memory: Full + RollingWindow | `tests/test_memory.cpp` green. |
| 10| Agent loop | Non-streaming and streaming loop tests green. |
diff --git a/include/agent_smith/detail/async_queue.hpp b/include/agent_smith/detail/async_queue.hpp
new file mode 100644
index 0000000..b888371
--- /dev/null
+++ b/include/agent_smith/detail/async_queue.hpp
@@ -0,0 +1,146 @@
+#pragma once
+
+#include <coroutine>
+#include <deque>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <utility>
+
+#include <mw/task.hpp>
+
+namespace agent_smith::detail
+{
+
+/// A closeable queue that resumes consumers when another thread pushes data.
+template <typename T> class AsyncQueue
+{
+public:
+ /// Add an item unless the queue has already been closed.
+ void push(T value)
+ {
+ std::shared_ptr<PopState> waiter;
+ std::coroutine_handle<> continuation;
+ {
+ std::lock_guard lock(mutex_);
+ if(closed_)
+ {
+ return;
+ }
+
+ if(waiters_.empty())
+ {
+ values_.push_back(std::move(value));
+ return;
+ }
+
+ waiter = std::move(waiters_.front());
+ waiters_.pop_front();
+ waiter->value.emplace(std::move(value));
+ continuation = waiter->continuation;
+ }
+
+ if(continuation != nullptr)
+ {
+ continuation.resume();
+ }
+ }
+
+ /// Wait for one item, or nullopt after closure and queue drainage.
+ mw::Task<std::optional<T>> pop()
+ {
+ co_return co_await PopAwaiter(*this);
+ }
+
+ /// Prevent further pushes and wake consumers once buffered items are read.
+ void close()
+ {
+ std::deque<std::shared_ptr<PopState>> waiters;
+ {
+ std::lock_guard lock(mutex_);
+ if(closed_)
+ {
+ return;
+ }
+
+ closed_ = true;
+ if(!values_.empty())
+ {
+ return;
+ }
+
+ waiters.swap(waiters_);
+ }
+
+ for(const auto& waiter : waiters)
+ {
+ if(waiter->continuation != nullptr)
+ {
+ waiter->continuation.resume();
+ }
+ }
+ }
+
+private:
+ struct PopState
+ {
+ std::optional<T> value;
+ std::coroutine_handle<> continuation = nullptr;
+ };
+
+ class PopAwaiter
+ {
+ public:
+ explicit PopAwaiter(AsyncQueue& queue)
+ : queue_(&queue), state_(std::make_shared<PopState>())
+ {
+ }
+
+ bool await_ready()
+ {
+ std::lock_guard lock(queue_->mutex_);
+ return queue_->take(*state_);
+ }
+
+ bool await_suspend(std::coroutine_handle<> continuation)
+ {
+ std::lock_guard lock(queue_->mutex_);
+ if(queue_->take(*state_))
+ {
+ return false;
+ }
+
+ state_->continuation = continuation;
+ queue_->waiters_.push_back(state_);
+ return true;
+ }
+
+ std::optional<T> await_resume()
+ {
+ return std::move(state_->value);
+ }
+
+ private:
+ AsyncQueue* queue_;
+ std::shared_ptr<PopState> state_;
+ };
+
+ bool take(PopState& state)
+ {
+ if(!values_.empty())
+ {
+ state.value.emplace(std::move(values_.front()));
+ values_.pop_front();
+ return true;
+ }
+
+ return closed_;
+ }
+
+ std::mutex mutex_;
+ std::deque<T> values_;
+ std::deque<std::shared_ptr<PopState>> waiters_;
+ bool closed_ = false;
+};
+
+} // namespace agent_smith::detail
diff --git a/include/agent_smith/detail/blocking_pool.hpp b/include/agent_smith/detail/blocking_pool.hpp
new file mode 100644
index 0000000..64b2e9a
--- /dev/null
+++ b/include/agent_smith/detail/blocking_pool.hpp
@@ -0,0 +1,218 @@
+#pragma once
+
+#include <concepts>
+#include <condition_variable>
+#include <coroutine>
+#include <cstddef>
+#include <exception>
+#include <expected>
+#include <functional>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <queue>
+#include <thread>
+#include <type_traits>
+#include <utility>
+#include <vector>
+
+#include <mw/error.hpp>
+
+namespace agent_smith::detail
+{
+
+template <typename T> struct ExpectedResultType
+{
+ using Type = mw::E<T>;
+};
+
+template <typename T> struct ExpectedResultType<mw::E<T>>
+{
+ using Type = mw::E<T>;
+};
+
+/// The flattened expected result produced by a blocking callable.
+template <typename T>
+using ExpectedResult =
+ typename ExpectedResultType<std::remove_cvref_t<T>>::Type;
+
+/// Private FIFO executor for work that cannot run on the agent coroutine.
+class BlockingPool
+{
+public:
+ explicit BlockingPool(std::size_t size);
+ ~BlockingPool();
+
+ BlockingPool(const BlockingPool&) = delete;
+ BlockingPool& operator=(const BlockingPool&) = delete;
+
+ template <typename F> auto run(F&& function);
+
+private:
+ struct WorkItem
+ {
+ std::function<void()> run;
+ std::function<void()> cancel;
+ };
+
+ bool enqueue(WorkItem item);
+ void worker(std::stop_token stop);
+
+ mutable std::mutex mutex_;
+ std::condition_variable condition_;
+ std::queue<WorkItem> pending_;
+ bool stopping_ = false;
+ std::vector<std::jthread> workers_;
+};
+
+template <typename Result, typename Function> class BlockingWork
+{
+public:
+ explicit BlockingWork(Function&& function) : function_(std::move(function))
+ {
+ }
+
+ void run()
+ {
+ Result result = invoke();
+ complete(std::move(result));
+ }
+
+ void cancel()
+ {
+ complete(std::unexpected(mw::runtimeError("blocking pool stopped")));
+ }
+
+ bool complete() const
+ {
+ std::lock_guard lock(mutex_);
+ return complete_;
+ }
+
+ bool suspend(std::coroutine_handle<> continuation)
+ {
+ std::lock_guard lock(mutex_);
+ if(complete_)
+ {
+ return false;
+ }
+
+ continuation_ = continuation;
+ return true;
+ }
+
+ Result result()
+ {
+ std::lock_guard lock(mutex_);
+ return std::move(*result_);
+ }
+
+private:
+ Result invoke()
+ {
+ try
+ {
+ using RawResult = std::invoke_result_t<Function&>;
+ if constexpr(std::is_void_v<RawResult>)
+ {
+ function_();
+ return {};
+ }
+ else if constexpr(std::same_as<std::remove_cvref_t<RawResult>,
+ Result>)
+ {
+ return function_();
+ }
+ else
+ {
+ return Result(function_());
+ }
+ }
+ catch(const std::exception& error)
+ {
+ return std::unexpected(mw::runtimeError(error.what()));
+ }
+ catch(...)
+ {
+ return std::unexpected(mw::runtimeError(
+ "blocking callable threw an unknown exception"));
+ }
+ }
+
+ void complete(Result result)
+ {
+ std::coroutine_handle<> continuation;
+ {
+ std::lock_guard lock(mutex_);
+ if(complete_)
+ {
+ return;
+ }
+
+ result_.emplace(std::move(result));
+ complete_ = true;
+ continuation = continuation_;
+ }
+
+ if(continuation != nullptr)
+ {
+ continuation.resume();
+ }
+ }
+
+ mutable std::mutex mutex_;
+ Function function_;
+ std::optional<Result> result_;
+ std::coroutine_handle<> continuation_ = nullptr;
+ bool complete_ = false;
+};
+
+template <typename Result, typename Function> class BlockingAwaiter
+{
+public:
+ explicit BlockingAwaiter(
+ std::shared_ptr<BlockingWork<Result, Function>> work)
+ : work_(std::move(work))
+ {
+ }
+
+ bool await_ready() const
+ {
+ return work_->complete();
+ }
+
+ bool await_suspend(std::coroutine_handle<> continuation)
+ {
+ return work_->suspend(continuation);
+ }
+
+ Result await_resume()
+ {
+ return work_->result();
+ }
+
+private:
+ std::shared_ptr<BlockingWork<Result, Function>> work_;
+};
+
+template <typename F> auto BlockingPool::run(F&& function)
+{
+ using Function = std::decay_t<F>;
+ using Result = ExpectedResult<std::invoke_result_t<Function&>>;
+ using Work = BlockingWork<Result, Function>;
+
+ auto work = std::make_shared<Work>(std::forward<F>(function));
+ WorkItem item{
+ .run = [work]() { work->run(); },
+ .cancel = [work]() { work->cancel(); },
+ };
+
+ if(!enqueue(std::move(item)))
+ {
+ work->cancel();
+ }
+
+ return BlockingAwaiter<Result, Function>(std::move(work));
+}
+
+} // namespace agent_smith::detail
diff --git a/include/agent_smith/detail/gather_all.hpp b/include/agent_smith/detail/gather_all.hpp
new file mode 100644
index 0000000..edcfced
--- /dev/null
+++ b/include/agent_smith/detail/gather_all.hpp
@@ -0,0 +1,42 @@
+#pragma once
+
+#include <utility>
+#include <vector>
+
+#include <mw/task.hpp>
+
+#include "started_task.hpp"
+
+namespace agent_smith::detail
+{
+
+/// Await started tasks in input order without changing their concurrency.
+template <typename T>
+mw::Task<std::vector<mw::E<T>>>
+gatherAllStarted(std::vector<StartedTask<T>> tasks)
+{
+ std::vector<mw::E<T>> results;
+ results.reserve(tasks.size());
+ for(auto& task : tasks)
+ {
+ results.push_back(co_await std::move(task));
+ }
+
+ co_return results;
+}
+
+/// Start all supplied tasks, then collect their results in input order.
+template <typename T>
+mw::Task<std::vector<mw::E<T>>> gatherAll(std::vector<mw::Task<mw::E<T>>> tasks)
+{
+ std::vector<StartedTask<T>> started;
+ started.reserve(tasks.size());
+ for(auto& task : tasks)
+ {
+ started.push_back(StartedTask<T>::start(std::move(task)));
+ }
+
+ co_return co_await gatherAllStarted(std::move(started));
+}
+
+} // namespace agent_smith::detail
diff --git a/include/agent_smith/detail/started_task.hpp b/include/agent_smith/detail/started_task.hpp
new file mode 100644
index 0000000..d19fe8e
--- /dev/null
+++ b/include/agent_smith/detail/started_task.hpp
@@ -0,0 +1,183 @@
+#pragma once
+
+#include <coroutine>
+#include <exception>
+#include <mutex>
+#include <optional>
+#include <stdexcept>
+#include <utility>
+
+#include <mw/error.hpp>
+#include <mw/task.hpp>
+
+namespace agent_smith::detail
+{
+
+/// An internal eager wrapper around one mw::Task<mw::E<T>> operation.
+template <typename T> class StartedTask
+{
+public:
+ struct promise_type
+ {
+ StartedTask get_return_object()
+ {
+ return StartedTask(
+ std::coroutine_handle<promise_type>::from_promise(*this));
+ }
+
+ std::suspend_never initial_suspend() noexcept
+ {
+ return {};
+ }
+
+ auto final_suspend() noexcept
+ {
+ struct FinalAwaiter
+ {
+ bool await_ready() noexcept
+ {
+ return false;
+ }
+
+ std::coroutine_handle<> await_suspend(
+ std::coroutine_handle<promise_type> handle) noexcept
+ {
+ std::lock_guard lock(handle.promise().mutex);
+ handle.promise().complete = true;
+ if(handle.promise().continuation != nullptr)
+ {
+ return handle.promise().continuation;
+ }
+ return std::noop_coroutine();
+ }
+
+ void await_resume() noexcept {}
+ };
+
+ return FinalAwaiter{};
+ }
+
+ void return_value(mw::E<T> value)
+ {
+ value_.emplace(std::move(value));
+ }
+
+ void unhandled_exception()
+ {
+ try
+ {
+ throw;
+ }
+ catch(const std::exception& error)
+ {
+ value_.emplace(std::unexpected(mw::runtimeError(error.what())));
+ }
+ catch(...)
+ {
+ value_.emplace(std::unexpected(mw::runtimeError(
+ "started task threw an unknown exception")));
+ }
+ }
+
+ std::mutex mutex;
+ std::optional<mw::E<T>> value_;
+ std::coroutine_handle<> continuation = nullptr;
+ bool complete = false;
+ };
+
+ StartedTask(StartedTask&& other) noexcept
+ : handle_(std::exchange(other.handle_, nullptr))
+ {
+ }
+
+ StartedTask& operator=(StartedTask&& other) noexcept
+ {
+ if(this == &other)
+ {
+ return *this;
+ }
+
+ destroy();
+ handle_ = std::exchange(other.handle_, nullptr);
+ return *this;
+ }
+
+ StartedTask(const StartedTask&) = delete;
+ StartedTask& operator=(const StartedTask&) = delete;
+
+ ~StartedTask()
+ {
+ destroy();
+ }
+
+ /// Start a lazy task immediately and retain its result for later awaiting.
+ static StartedTask start(mw::Task<mw::E<T>> task)
+ {
+ co_return co_await std::move(task);
+ }
+
+ auto operator co_await() &&
+ {
+ struct Awaiter
+ {
+ std::coroutine_handle<promise_type> handle;
+
+ bool await_ready()
+ {
+ std::lock_guard lock(handle.promise().mutex);
+ return handle.promise().complete;
+ }
+
+ bool await_suspend(std::coroutine_handle<> continuation)
+ {
+ std::lock_guard lock(handle.promise().mutex);
+ if(handle.promise().complete)
+ {
+ return false;
+ }
+
+ handle.promise().continuation = continuation;
+ return true;
+ }
+
+ mw::E<T> await_resume()
+ {
+ mw::E<T> result = [&]()
+ {
+ std::lock_guard lock(handle.promise().mutex);
+ return std::move(*handle.promise().value_);
+ }();
+ handle.destroy();
+ return result;
+ }
+ };
+
+ return Awaiter{std::exchange(handle_, nullptr)};
+ }
+
+private:
+ explicit StartedTask(std::coroutine_handle<promise_type> handle)
+ : handle_(handle)
+ {
+ }
+
+ void destroy()
+ {
+ if(handle_ == nullptr)
+ {
+ return;
+ }
+
+ if(!handle_.done())
+ {
+ std::terminate();
+ }
+
+ handle_.destroy();
+ handle_ = nullptr;
+ }
+
+ std::coroutine_handle<promise_type> handle_ = nullptr;
+};
+
+} // namespace agent_smith::detail
diff --git a/include/agent_smith/runtime.hpp b/include/agent_smith/runtime.hpp
index c1d67f0..479bf32 100644
--- a/include/agent_smith/runtime.hpp
+++ b/include/agent_smith/runtime.hpp
@@ -2,9 +2,18 @@
#include <chrono>
#include <cstddef>
+#include <memory>
+#include <type_traits>
#include <mw/http_client_async.hpp>
+#include "detail/blocking_pool.hpp"
+
+namespace agent_smith::detail
+{
+class TimerService;
+}
+
namespace agent_smith
{
@@ -39,9 +48,26 @@ public:
/// Return the shared asynchronous HTTP session.
mw::HTTPSessionAsync& http();
+ /// Run non-HTTP work on the internal worker pool.
+ template <typename F>
+ auto runOnWorker(F&& function)
+ -> mw::Task<detail::ExpectedResult<std::invoke_result_t<F&>>>;
+
+ /// Suspend without blocking the caller's coroutine thread.
+ mw::Task<mw::E<void>> sleepFor(std::chrono::milliseconds duration);
+
private:
RuntimeConfig cfg_;
- mw::HTTPSessionAsync http_;
+ std::unique_ptr<mw::HTTPSessionAsync> http_;
+ std::unique_ptr<detail::BlockingPool> blocking_pool_;
+ std::unique_ptr<detail::TimerService> timer_service_;
};
+template <typename F>
+auto Runtime::runOnWorker(F&& function)
+ -> mw::Task<detail::ExpectedResult<std::invoke_result_t<F&>>>
+{
+ co_return co_await blocking_pool_->run(std::forward<F>(function));
+}
+
} // namespace agent_smith
diff --git a/src/detail/blocking_pool.cpp b/src/detail/blocking_pool.cpp
new file mode 100644
index 0000000..71209da
--- /dev/null
+++ b/src/detail/blocking_pool.cpp
@@ -0,0 +1,78 @@
+#include "agent_smith/detail/blocking_pool.hpp"
+
+#include <algorithm>
+
+namespace agent_smith::detail
+{
+
+BlockingPool::BlockingPool(std::size_t size)
+{
+ const std::size_t worker_count = std::max<std::size_t>(size, 1);
+ workers_.reserve(worker_count);
+ for(std::size_t index = 0; index < worker_count; ++index)
+ {
+ workers_.emplace_back([this](std::stop_token stop) { worker(stop); });
+ }
+}
+
+BlockingPool::~BlockingPool()
+{
+ std::queue<WorkItem> pending;
+ {
+ std::lock_guard lock(mutex_);
+ stopping_ = true;
+ pending.swap(pending_);
+ }
+
+ while(!pending.empty())
+ {
+ pending.front().cancel();
+ pending.pop();
+ }
+
+ condition_.notify_all();
+}
+
+bool BlockingPool::enqueue(WorkItem item)
+{
+ {
+ std::lock_guard lock(mutex_);
+ if(stopping_)
+ {
+ return false;
+ }
+ pending_.push(std::move(item));
+ }
+
+ condition_.notify_one();
+ return true;
+}
+
+void BlockingPool::worker(std::stop_token stop)
+{
+ while(!stop.stop_requested())
+ {
+ WorkItem item;
+ {
+ std::unique_lock lock(mutex_);
+ condition_.wait(lock,
+ [&]()
+ {
+ return stop.stop_requested() || stopping_ ||
+ !pending_.empty();
+ });
+
+ if(stop.stop_requested() || stopping_)
+ {
+ return;
+ }
+
+ item = std::move(pending_.front());
+ pending_.pop();
+ }
+
+ item.run();
+ }
+}
+
+} // namespace agent_smith::detail
diff --git a/src/detail/timer_service.cpp b/src/detail/timer_service.cpp
new file mode 100644
index 0000000..cfadf11
--- /dev/null
+++ b/src/detail/timer_service.cpp
@@ -0,0 +1,90 @@
+#include "timer_service.hpp"
+
+namespace agent_smith::detail
+{
+
+TimerService::TimerService()
+ : worker_([this](std::stop_token stop) { worker(stop); })
+{
+}
+
+TimerService::~TimerService()
+{
+ worker_.request_stop();
+ condition_.notify_all();
+}
+
+TimerService::SleepAwaiter::SleepAwaiter(TimerService& service,
+ std::chrono::milliseconds duration)
+ : service_(&service), duration_(duration), entry_(std::make_shared<Entry>())
+{
+}
+
+bool TimerService::SleepAwaiter::await_ready() const
+{
+ return duration_ <= std::chrono::milliseconds::zero();
+}
+
+bool TimerService::SleepAwaiter::await_suspend(
+ std::coroutine_handle<> continuation)
+{
+ entry_->deadline = std::chrono::steady_clock::now() + duration_;
+ service_->schedule(entry_, continuation);
+ return true;
+}
+
+TimerService::SleepAwaiter
+TimerService::sleepFor(std::chrono::milliseconds duration)
+{
+ return SleepAwaiter(*this, duration);
+}
+
+void TimerService::schedule(const std::shared_ptr<Entry>& entry,
+ std::coroutine_handle<> continuation)
+{
+ {
+ std::lock_guard lock(mutex_);
+ entry->continuation = continuation;
+ entries_.push(entry);
+ }
+ condition_.notify_one();
+}
+
+void TimerService::worker(std::stop_token stop)
+{
+ std::unique_lock lock(mutex_);
+ while(!stop.stop_requested())
+ {
+ if(entries_.empty())
+ {
+ condition_.wait(
+ lock,
+ [&]() { return stop.stop_requested() || !entries_.empty(); });
+ continue;
+ }
+
+ const auto deadline = entries_.top()->deadline;
+ if(condition_.wait_until(lock, deadline,
+ [&]()
+ {
+ return stop.stop_requested() ||
+ entries_.empty() ||
+ entries_.top()->deadline < deadline;
+ }))
+ {
+ continue;
+ }
+
+ auto entry = entries_.top();
+ entries_.pop();
+ auto continuation = entry->continuation;
+ lock.unlock();
+ if(continuation != nullptr)
+ {
+ continuation.resume();
+ }
+ lock.lock();
+ }
+}
+
+} // namespace agent_smith::detail
diff --git a/src/detail/timer_service.hpp b/src/detail/timer_service.hpp
new file mode 100644
index 0000000..96f847b
--- /dev/null
+++ b/src/detail/timer_service.hpp
@@ -0,0 +1,74 @@
+#pragma once
+
+#include <chrono>
+#include <condition_variable>
+#include <coroutine>
+#include <cstddef>
+#include <memory>
+#include <mutex>
+#include <queue>
+#include <thread>
+#include <vector>
+
+namespace agent_smith::detail
+{
+
+class TimerService
+{
+private:
+ struct Entry;
+
+public:
+ TimerService();
+ ~TimerService();
+
+ TimerService(const TimerService&) = delete;
+ TimerService& operator=(const TimerService&) = delete;
+
+ class SleepAwaiter
+ {
+ public:
+ SleepAwaiter(TimerService& service, std::chrono::milliseconds duration);
+
+ bool await_ready() const;
+ bool await_suspend(std::coroutine_handle<> continuation);
+ void await_resume() const noexcept {}
+
+ private:
+ TimerService* service_;
+ std::chrono::milliseconds duration_;
+ std::shared_ptr<Entry> entry_;
+ };
+
+ SleepAwaiter sleepFor(std::chrono::milliseconds duration);
+
+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;
+ }
+ };
+
+ void schedule(const std::shared_ptr<Entry>& entry,
+ std::coroutine_handle<> continuation);
+ void worker(std::stop_token stop);
+
+ std::mutex mutex_;
+ std::condition_variable condition_;
+ std::priority_queue<std::shared_ptr<Entry>,
+ std::vector<std::shared_ptr<Entry>>,
+ EntryLater>
+ entries_;
+ std::jthread worker_;
+};
+
+} // namespace agent_smith::detail
diff --git a/src/runtime.cpp b/src/runtime.cpp
index beb1e45..3f2dd0a 100644
--- a/src/runtime.cpp
+++ b/src/runtime.cpp
@@ -1,22 +1,43 @@
#include "agent_smith/runtime.hpp"
+#include "detail/timer_service.hpp"
+
namespace agent_smith
{
-Runtime::Runtime(RuntimeConfig cfg) : cfg_(cfg)
+Runtime::Runtime(RuntimeConfig cfg)
+ : cfg_(cfg), http_(std::make_unique<mw::HTTPSessionAsync>()),
+ blocking_pool_(std::make_unique<detail::BlockingPool>(
+ cfg_.blocking_pool_size)),
+ timer_service_(std::make_unique<detail::TimerService>())
{
- auto& http = static_cast<mw::HTTPSessionAsyncInterface&>(http_);
+ auto& http = static_cast<mw::HTTPSessionAsyncInterface&>(*http_);
[[maybe_unused]] auto connection_timeout =
http.connectionTimeout(cfg_.connection_timeout);
[[maybe_unused]] auto transfer_timeout =
http.transferTimeout(cfg_.transfer_timeout);
}
-Runtime::~Runtime() = default;
+Runtime::~Runtime()
+{
+ http_.reset();
+}
mw::HTTPSessionAsync& Runtime::http()
{
- return http_;
+ return *http_;
+}
+
+mw::Task<mw::E<void>> Runtime::sleepFor(std::chrono::milliseconds duration)
+{
+ if(duration < std::chrono::milliseconds::zero())
+ {
+ co_return std::unexpected(
+ mw::runtimeError("sleep duration cannot be negative"));
+ }
+
+ co_await timer_service_->sleepFor(duration);
+ co_return mw::E<void>{};
}
} // namespace agent_smith
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index 4d24025..2645200 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -4,6 +4,7 @@ target_link_libraries(agent_smith_test_skeleton
agent_smith::agent_smith
GTest::gtest_main
)
+agent_smith_enable_test_sanitizer(agent_smith_test_skeleton)
include(GoogleTest)
gtest_discover_tests(agent_smith_test_skeleton)
diff --git a/tests/test_skeleton.cpp b/tests/test_skeleton.cpp
index 3302d55..13b5bf3 100644
--- a/tests/test_skeleton.cpp
+++ b/tests/test_skeleton.cpp
@@ -1,4 +1,19 @@
+#include <atomic>
+#include <chrono>
+#include <condition_variable>
+#include <coroutine>
+#include <exception>
+#include <mutex>
+#include <optional>
+#include <stdexcept>
+#include <thread>
+#include <utility>
+#include <vector>
+
#include <agent_smith/agent_smith.hpp>
+#include <agent_smith/detail/async_queue.hpp>
+#include <agent_smith/detail/gather_all.hpp>
+
#include <gtest/gtest.h>
namespace agent_smith
@@ -6,6 +21,140 @@ namespace agent_smith
namespace
{
+template <typename T> class SyncWaitTask
+{
+public:
+ struct promise_type
+ {
+ SyncWaitTask get_return_object()
+ {
+ return SyncWaitTask(
+ std::coroutine_handle<promise_type>::from_promise(*this));
+ }
+
+ std::suspend_always initial_suspend() noexcept
+ {
+ return {};
+ }
+
+ auto final_suspend() noexcept
+ {
+ struct FinalAwaiter
+ {
+ bool await_ready() noexcept
+ {
+ return false;
+ }
+
+ void await_suspend(
+ std::coroutine_handle<promise_type> handle) noexcept
+ {
+ auto& promise = handle.promise();
+ {
+ std::lock_guard lock(promise.mutex);
+ promise.done = true;
+ promise.condition.notify_one();
+ }
+ }
+
+ void await_resume() noexcept {}
+ };
+
+ return FinalAwaiter{};
+ }
+
+ template <typename U> void return_value(U&& value)
+ {
+ value_.emplace(std::forward<U>(value));
+ }
+
+ void unhandled_exception()
+ {
+ exception_ = std::current_exception();
+ }
+
+ std::mutex mutex;
+ std::condition_variable condition;
+ bool done = false;
+ std::optional<T> value_;
+ std::exception_ptr exception_;
+ };
+
+ explicit SyncWaitTask(std::coroutine_handle<promise_type> handle)
+ : handle_(handle)
+ {
+ }
+
+ SyncWaitTask(SyncWaitTask&& other) noexcept
+ : handle_(std::exchange(other.handle_, nullptr))
+ {
+ }
+
+ SyncWaitTask(const SyncWaitTask&) = delete;
+ SyncWaitTask& operator=(const SyncWaitTask&) = delete;
+
+ ~SyncWaitTask()
+ {
+ if(handle_ != nullptr)
+ {
+ handle_.destroy();
+ }
+ }
+
+ void start()
+ {
+ handle_.resume();
+ }
+
+ T get()
+ {
+ auto& promise = handle_.promise();
+ std::unique_lock lock(promise.mutex);
+ promise.condition.wait(lock, [&]() { return promise.done; });
+ lock.unlock();
+
+ if(promise.exception_ != nullptr)
+ {
+ std::rethrow_exception(promise.exception_);
+ }
+
+ return std::move(*promise.value_);
+ }
+
+private:
+ std::coroutine_handle<promise_type> handle_ = nullptr;
+};
+
+template <typename T> SyncWaitTask<T> makeWaiter(mw::Task<T> task)
+{
+ co_return co_await std::move(task);
+}
+
+template <typename T> T syncWait(mw::Task<T> task)
+{
+ auto waiter = makeWaiter(std::move(task));
+ waiter.start();
+ return waiter.get();
+}
+
+mw::Task<mw::E<int>> markAndSleep(Runtime& runtime, std::atomic<bool>& started,
+ int value, std::chrono::milliseconds duration)
+{
+ started.store(true);
+ auto slept = co_await runtime.sleepFor(duration);
+ if(!slept)
+ {
+ co_return std::unexpected(slept.error());
+ }
+
+ co_return value;
+}
+
+mw::Task<mw::E<int>> awaitStarted(detail::StartedTask<int> task)
+{
+ co_return co_await std::move(task);
+}
+
TEST(RuntimeTest, ConstructsSharedHttpSession)
{
Runtime runtime;
@@ -22,5 +171,124 @@ TEST(AgentTest, ConstructsEmptyAgent)
EXPECT_NE(&agent, nullptr);
}
+TEST(RuntimeTest, RunsPlainBlockingResult)
+{
+ Runtime runtime;
+
+ auto result = syncWait(runtime.runOnWorker([]() { return 42; }));
+
+ ASSERT_TRUE(result.has_value());
+ EXPECT_EQ(*result, 42);
+}
+
+TEST(RuntimeTest, FlattensExpectedBlockingResult)
+{
+ Runtime runtime;
+
+ auto result = syncWait(runtime.runOnWorker(
+ []() -> mw::E<int>
+ { return std::unexpected(mw::runtimeError("tool failed")); }));
+
+ ASSERT_FALSE(result.has_value());
+ EXPECT_EQ(result.error().msg(), "tool failed");
+}
+
+TEST(RuntimeTest, ConvertsBlockingExceptionsToErrors)
+{
+ Runtime runtime;
+
+ auto result = syncWait(runtime.runOnWorker(
+ []() -> int { throw std::runtime_error("unexpected failure"); }));
+
+ ASSERT_FALSE(result.has_value());
+ EXPECT_EQ(result.error().msg(), "unexpected failure");
+}
+
+TEST(RuntimeTest, SleepsWithoutBlockingTheCaller)
+{
+ Runtime runtime;
+ const auto start = std::chrono::steady_clock::now();
+
+ auto result = syncWait(runtime.sleepFor(std::chrono::milliseconds(25)));
+
+ ASSERT_TRUE(result.has_value());
+ EXPECT_GE(std::chrono::steady_clock::now() - start,
+ std::chrono::milliseconds(20));
+}
+
+TEST(StartedTaskTest, StartsImmediatelyAndCanBeAwaited)
+{
+ Runtime runtime;
+ std::atomic<bool> started = false;
+
+ auto task = detail::StartedTask<int>::start(
+ markAndSleep(runtime, started, 7, std::chrono::milliseconds(1)));
+
+ EXPECT_TRUE(started.load());
+ auto result = syncWait(awaitStarted(std::move(task)));
+ ASSERT_TRUE(result.has_value());
+ EXPECT_EQ(*result, 7);
+}
+
+TEST(AsyncQueueTest, DeliversItemsAcrossThreadsAndCloses)
+{
+ detail::AsyncQueue<int> queue;
+ std::thread producer(
+ [&queue]()
+ {
+ std::this_thread::sleep_for(std::chrono::milliseconds(5));
+ queue.push(9);
+ queue.close();
+ });
+
+ auto first = syncWait(queue.pop());
+ auto second = syncWait(queue.pop());
+ producer.join();
+
+ ASSERT_TRUE(first.has_value());
+ EXPECT_EQ(*first, 9);
+ EXPECT_FALSE(second.has_value());
+}
+
+TEST(AsyncQueueTest, DrainsBufferedItemsBeforeClosure)
+{
+ detail::AsyncQueue<int> queue;
+ queue.push(3);
+ queue.push(5);
+ queue.close();
+
+ auto first = syncWait(queue.pop());
+ auto second = syncWait(queue.pop());
+ auto closed = syncWait(queue.pop());
+
+ ASSERT_TRUE(first.has_value());
+ ASSERT_TRUE(second.has_value());
+ EXPECT_EQ(*first, 3);
+ EXPECT_EQ(*second, 5);
+ EXPECT_FALSE(closed.has_value());
+}
+
+TEST(GatherAllTest, StartsAllTasksAndPreservesInputOrder)
+{
+ Runtime runtime;
+ std::atomic<bool> first_started = false;
+ std::atomic<bool> second_started = false;
+ std::vector<mw::Task<mw::E<int>>> tasks;
+ tasks.push_back(
+ markAndSleep(runtime, first_started, 1, std::chrono::milliseconds(20)));
+ tasks.push_back(
+ markAndSleep(runtime, second_started, 2, std::chrono::milliseconds(1)));
+
+ auto results = syncWait(detail::gatherAll(std::move(tasks)));
+
+ ASSERT_EQ(results.size(), 2U);
+ ASSERT_TRUE(results[0].has_value());
+ ASSERT_TRUE(results[1].has_value());
+ EXPECT_TRUE(first_started.load());
+ EXPECT_TRUE(second_started.load());
+ EXPECT_EQ(*results[0], 1);
+ EXPECT_EQ(*results[1], 2);
+}
+
} // namespace
} // namespace agent_smith