Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,11 @@ loosely while pre-1.0 (breaking changes can land on minor bumps).
## [Unreleased]

### Fixed
- **SSE mailbox terminal after overflow.** `sse_mailbox_push(...,
force_deliver=true)` now evicts oldest events and delivers a terminal
envelope even when a prior push already set `overflowed`. Request-event
replay and A2A resubscribe close with the real `done` / task-status
frame instead of dropping it for `slow_consumer`.
- **Event ingest hydrates the routed constitution.** `POST /v1/events`
attaches the selected agent's `agent_def` on the synthetic orchestrate
body. File-backed matches and tenant agents off the newest-200 catalog
Expand Down
11 changes: 9 additions & 2 deletions include/sse_mailbox.h
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,9 @@ inline constexpr size_t kDefaultSseMailboxMaxDepth = 2048;
// Push one item into a per-connection mailbox. Returns false when the
// item was rejected because the client is too slow (overflowed is set).
// When `force_deliver` is true (terminal SSE envelopes), drop the oldest
// buffered items until there is room so the stream can close cleanly.
// buffered items until there is room so the stream can close cleanly —
// including after a prior overflow, which would otherwise reject the
// terminal frame and leave the subscriber with only `slow_consumer`.
template<typename T>
bool sse_mailbox_push(std::deque<T>& mailbox,
std::mutex& mu,
Expand All @@ -28,7 +30,8 @@ bool sse_mailbox_push(std::deque<T>& mailbox,
T item,
bool force_deliver = false) {
std::lock_guard<std::mutex> lk(mu);
if (overflowed.load(std::memory_order_relaxed)) return false;
if (overflowed.load(std::memory_order_relaxed) && !force_deliver)
return false;
const size_t cap = kDefaultSseMailboxMaxDepth;
if (mailbox.size() >= cap) {
if (force_deliver) {
Expand All @@ -42,6 +45,10 @@ bool sse_mailbox_push(std::deque<T>& mailbox,
}
}
mailbox.push_back(std::move(item));
// Terminal delivery wins over a prior overflow so the drain loop
// writes the real done/status envelope instead of aborting.
if (force_deliver)
overflowed.store(false, std::memory_order_relaxed);
cv.notify_one();
return true;
}
Expand Down
24 changes: 24 additions & 0 deletions tests/test_sse_mailbox.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -46,3 +46,27 @@ TEST_CASE("sse_mailbox_push force_deliver makes room for terminal item") {
CHECK(mailbox.back() == "terminal");
CHECK(mailbox.front() == "1");
}

TEST_CASE("sse_mailbox_push force_deliver still lands after overflow") {
std::mutex mu;
std::condition_variable cv;
std::deque<std::string> mailbox;
std::atomic<bool> overflowed{false};

for (size_t i = 0; i < kDefaultSseMailboxMaxDepth; ++i) {
REQUIRE(sse_mailbox_push(mailbox, mu, cv, overflowed,
std::to_string(i)));
}
CHECK_FALSE(sse_mailbox_push(mailbox, mu, cv, overflowed,
std::string("dropped")));
REQUIRE(overflowed.load());
CHECK(mailbox.size() == kDefaultSseMailboxMaxDepth);
CHECK(mailbox.back() != "dropped");

REQUIRE(sse_mailbox_push(mailbox, mu, cv, overflowed,
std::string("terminal"), /*force_deliver=*/true));
CHECK_FALSE(overflowed.load());
CHECK(mailbox.size() == kDefaultSseMailboxMaxDepth);
CHECK(mailbox.back() == "terminal");
CHECK(mailbox.front() == "1");
}
Loading