Cap A2A unary HTTP bodies against unbounded remotes

CHANGELOG.md

@@ -7,6 +7,64 @@ loosely while pre-1.0 (breaking changes can land on minor bumps).
 
 ## [Unreleased]
 
+### Fixed
+- **A2A unary HTTP body cap.** `rpc_call` / `http_get` now refuse
+  responses over 16 MiB (same ceiling as the inbound HTTP API and
+  `kSseMaxEventBytes`). A remote that omitted `Content-Length` could
+  previously grow the client buffer without bound; SSE was already
+  capped (#317).
+- **Provider pool cancel while waiting for a connection slot.** `complete()` /
+  `stream()` no longer hang on the per-provider cap (`kMaxConnsPerProvider`)
+  after Esc or `cancel()`. The wait predicate observes the request token and
+  the process-wide cancel bits; `cancel()` / `CancelToken::request_cancel()`
+  notify every pool CV. A cancelled waiter does not consume a slot.
+- **`/diff apply` directory and cross-device writes.** A directory at the
+  patch target is rejected instead of being treated as a missing file.
+  The rename fallback no longer uses `copy_file` (which followed a dest
+  symlink and reported success when only the temp file was removed).
+- **Sandbox `/read` TOCTOU test is deterministic.** The leaf-symlink-swap
+  case plants the swap via a post-resolve hook instead of a timed helper
+  thread (macos-arm64 CI was racing the 80 ms pause).
+- **Recovered `/write` bytes persist.** `send_streaming` and
+  `run_dispatch` now fold `recover_truncated_writes` into the
+  cumulative turn after the resume call, so HTTP/SSE conversation
+  saves and token/cost totals include the resumed file body instead
+  of dropping it when `resp.content` is overwritten with
+  `total_content`.
+- **Conversation list cursor.** `GET /v1/conversations` now orders by
+  `updated_at DESC, id DESC` and accepts `before_id` so pages that share
+  an epoch second no longer skip or duplicate rows. Timestamp-only
+  `before_updated_at` stays valid for old clients.
+- **Token cancel is not a client-wide kill-switch.** `complete()` /
+  `stream()` no longer set `hard_cancelled_` when only the thread's
+  `CancelToken` is set. Esc on one TUI pane or `/kill` of a `/loop`
+  was aborting sibling streams that share the same `ApiClient`
+  (#46 / #48). `cancel()` and kill-switch preflight stay sticky.
+- **A2A `tasks/cancel` survives `message/send`.** `update_a2a_task` no
+- **Reopened todos clear `completed_at`.** `update_todo` (HTTP PATCH and
+- **MCP string JSON-RPC ids.** `parse_response` now accepts a decimal-string
+- **Reconcile rollback no longer wipes the workspace on a failed restore.**
+- **LaTeX math recursion depth.** `latex_math_to_plain` now stops converting
+- **Secret key/token writes do not follow a planted dest symlink.**
+- **Remote `--connect` base URL query/userinfo.** `normalize_api_base_url`
+- **Intent reconcile Phase B (JIT ΔS waves).** `POST /v1/reconcile` `mode=ensure`
+- **A2A unary HTTP errors stay bounded.** `Client::rpc` no longer concatenates
+- **Remote `--connect` DELETE/PATCH body cap.** Conversation delete and
+- **Intent LLM prompt text cap.** `build_llm_user_prompt` now truncates
+- **Remote TUI: recoverable SSE `error` is not a failed turn.** `RemoteSseTurnConsumer::finish` copied accumulated `error` event text even when the terminal `done` event had `ok: true` (e.g. catalog skip of a stored agent whose JSON failed validation). `done` is authoritative: success clears the result error; failure still prefers `done.error` and falls back to prior `error` events when that field is empty.
+- **Unreadable TUI sessions are not empty.** `session_json_is_empty` no
+- **Advise-gate cancel is not a bad bearer.** `POST /v1/advise/gate`
+- **MCP registry writes do not follow a planted `.tmp` symlink.**
+- **`/schedule` calendar dates.** `on YYYY-MM-DD` now rejects impossible
+- **MCP registry `env` overrides parent keys.** Subprocess spawn skipped
+- **`/schedule` time math fail-closed.** `every hour` / `hourly` now use the
+- **JSON numbers require a complete fraction and exponent.** `json_parse`
+- **`/fetch` uses the same SSRF hostname preflight as `/browse`.**
+- **Loop `/kill` wakes the inter-iteration pause.** After each turn,
+- **Memory `tag=` LIKE wildcards are literals.** `/mem entries tag=` and
+- **Dispatch of stored agents past newest-200.** `GET /v1/agents/:id` already
+- **Session restore skips non-object message rows.** `decode_messages_json`
+- **Lesson search is a literal substring.** `search_lessons` (`GET /v1/lessons?q=`
 ## [0.13.9] — 2026-09-21
 
 ### Fixed

CMakeLists.txt

@@ -1597,6 +1597,24 @@ if(INDEX_BUILD_TESTS)
     )
     add_test(NAME unit_a2a COMMAND unit_a2a)
 
+    # Unary A2A HTTP body cap.  Separate binary so the in-process unit_a2a
+    # suite stays network-free; this one stands up a loopback server.
+    add_executable(unit_a2a_http
+        tests/test_a2a_http.cpp
+        src/a2a/http.cpp
+        src/a2a/sse_reader.cpp
+        src/ssrf_guard.cpp
+    )
+    target_include_directories(unit_a2a_http PRIVATE
+        ${CMAKE_SOURCE_DIR}/include
+        ${CMAKE_SOURCE_DIR}/third_party/doctest
+    )
+    target_link_libraries(unit_a2a_http PRIVATE
+        CURL::libcurl
+        Threads::Threads
+    )
+    add_test(NAME unit_a2a_http COMMAND unit_a2a_http)
+
     # Constitution / prompt composer.  Pins the capability-bundle resolution
     # and verifies that empty-capabilities still emits the legacy surface
     # (back-compat for the master orchestrator).

include/a2a/http.h

@@ -20,12 +20,19 @@
 #include "a2a/sse_reader.h"
 
 #include <atomic>
+#include <cstddef>
 #include <functional>
 #include <string>
 #include <vector>
 
 namespace arbiter::a2a {
 
+// Unary rpc_call / http_get body cap.  Matches kSseMaxEventBytes and the
+// inbound HTTP API body limit so a remote cannot OOM the process by
+// streaming without a Content-Length.  SSE already caps per-event and
+// diagnostic raw buffers; these calls previously appended unbounded.
+inline constexpr size_t kHttpMaxBodyBytes = 16 * 1024 * 1024;
+
 struct HttpHeader {
     std::string name;
     std::string value;

src/a2a/http.cpp

@@ -21,10 +21,40 @@ namespace {
 
 // libcurl write callback for buffering a complete response body.
 // CURL signals EOF by closing; our caller reads `body` after perform().
+// Returning 0 aborts with CURLE_WRITE_ERROR once the unary size cap
+// is hit so a remote without Content-Length cannot grow `body` without
+// bound (the SSE path already caps via SseReader + kRawCap).
+struct BodyBuf {
+    std::string* body     = nullptr;
+    bool         overflow = false;
+};
+
 size_t write_to_string(char* ptr, size_t size, size_t nmemb, void* userdata) {
-    auto* dst = static_cast<std::string*>(userdata);
-    dst->append(ptr, size * nmemb);
-    return size * nmemb;
+    auto* buf = static_cast<BodyBuf*>(userdata);
+    const size_t n = size * nmemb;
+    if (!buf || !buf->body || buf->overflow) return 0;
+    if (n > kHttpMaxBodyBytes || buf->body->size() > kHttpMaxBodyBytes - n) {
+        buf->overflow = true;
+        return 0;
+    }
+    buf->body->append(ptr, n);
+    return n;
+}
+
+void apply_unary_size_cap(CURL* curl) {
+    curl_easy_setopt(curl, CURLOPT_MAXFILESIZE_LARGE,
+                     static_cast<curl_off_t>(kHttpMaxBodyBytes));
+}
+
+void finish_unary(HttpResponse& out, CURLcode rc, const BodyBuf& buf) {
+    if (buf.overflow || rc == CURLE_FILESIZE_EXCEEDED) {
+        out.error = "response exceeded size limit";
+        out.body.clear();
+        return;
+    }
+    if (rc != CURLE_OK) {
+        out.error = curl_easy_strerror(rc);
+    }
 }
 
 // Build a curl_slist from our HttpHeader vector.  Caller frees with
@@ -132,18 +162,18 @@ HttpResponse rpc_call(const std::string& url,
     curl_easy_setopt(curl, CURLOPT_POST,          1L);
     curl_easy_setopt(curl, CURLOPT_POSTFIELDS,    body.c_str());
     curl_easy_setopt(curl, CURLOPT_POSTFIELDSIZE, static_cast<long>(body.size()));
+    BodyBuf buf{&out.body, false};
     curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, write_to_string);
-    curl_easy_setopt(curl, CURLOPT_WRITEDATA,     &out.body);
+    curl_easy_setopt(curl, CURLOPT_WRITEDATA,     &buf);
     apply_common_opts(curl, timeout_secs, ssrf_guard);
+    apply_unary_size_cap(curl);
 
     CURLcode rc = curl_easy_perform(curl);
     curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &out.status_code);
     curl_slist_free_all(headers);
     curl_easy_cleanup(curl);
 
-    if (rc != CURLE_OK) {
-        out.error = curl_easy_strerror(rc);
-    }
+    finish_unary(out, rc, buf);
     return out;
 }
 
@@ -165,21 +195,21 @@ HttpResponse http_get(const std::string& url,
                                                 /*accept_sse=*/false);
     headers = curl_slist_append(headers, "Accept: application/json");
 
+    BodyBuf buf{&out.body, false};
     curl_easy_setopt(curl, CURLOPT_URL,           url.c_str());
     curl_easy_setopt(curl, CURLOPT_HTTPHEADER,    headers);
     curl_easy_setopt(curl, CURLOPT_HTTPGET,       1L);
     curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, write_to_string);
-    curl_easy_setopt(curl, CURLOPT_WRITEDATA,     &out.body);
+    curl_easy_setopt(curl, CURLOPT_WRITEDATA,     &buf);
     apply_common_opts(curl, timeout_secs, ssrf_guard);
+    apply_unary_size_cap(curl);
 
     CURLcode rc = curl_easy_perform(curl);
     curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &out.status_code);
     curl_slist_free_all(headers);
     curl_easy_cleanup(curl);
 
-    if (rc != CURLE_OK) {
-        out.error = curl_easy_strerror(rc);
-    }
+    finish_unary(out, rc, buf);
     return out;
 }
 

tests/test_a2a_http.cpp

@@ -0,0 +1,146 @@
+// tests/test_a2a_http.cpp — Unary A2A HTTP body cap (#317's SSE caps
+// did not cover rpc_call / http_get).  A loopback server streams past
+// kHttpMaxBodyBytes; the client must abort and surface a size-limit
+// error instead of buffering the whole payload.
+#define DOCTEST_CONFIG_IMPLEMENT_WITH_MAIN
+#include "doctest.h"
+
+#include "a2a/http.h"
+
+#include <arpa/inet.h>
+#include <netinet/in.h>
+#include <sys/socket.h>
+#include <unistd.h>
+
+#include <atomic>
+#include <chrono>
+#include <cstring>
+#include <string>
+#include <thread>
+
+using namespace std::chrono_literals;
+using arbiter::a2a::http_get;
+using arbiter::a2a::rpc_call;
+using arbiter::a2a::kHttpMaxBodyBytes;
+
+namespace {
+
+struct LoopbackServer {
+    int         listen_fd = -1;
+    int         port      = 0;
+    std::thread acceptor;
+
+    // When true, send Content-Length larger than the cap and a tiny body
+    // so CURLOPT_MAXFILESIZE_LARGE can refuse before the write callback.
+    bool        advertise_oversize_length = false;
+    // When true, omit Content-Length and stream just past the cap so the
+    // write callback is the one that trips.
+    bool        stream_oversize_body      = false;
+
+    std::atomic<bool> client_gone{false};
+
+    void start() {
+        listen_fd = ::socket(AF_INET, SOCK_STREAM, 0);
+        REQUIRE(listen_fd >= 0);
+        int yes = 1;
+        ::setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes));
+
+        sockaddr_in addr{};
+        addr.sin_family      = AF_INET;
+        addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+        addr.sin_port        = 0;
+        REQUIRE(::bind(listen_fd, reinterpret_cast<sockaddr*>(&addr),
+                       sizeof(addr)) == 0);
+        REQUIRE(::listen(listen_fd, 1) == 0);
+
+        socklen_t len = sizeof(addr);
+        REQUIRE(::getsockname(listen_fd, reinterpret_cast<sockaddr*>(&addr),
+                              &len) == 0);
+        port = ntohs(addr.sin_port);
+
+        acceptor = std::thread([this] { handle_one(); });
+    }
+
+    void handle_one() {
+        int cs = ::accept(listen_fd, nullptr, nullptr);
+        if (cs < 0) return;
+        char scratch[4096];
+        ::recv(cs, scratch, sizeof(scratch), 0);
+
+        if (advertise_oversize_length) {
+            const std::string hdr =
+                "HTTP/1.1 200 OK\r\n"
+                "Content-Type: application/json\r\n"
+                "Content-Length: " +
+                std::to_string(kHttpMaxBodyBytes + 1) +
+                "\r\n\r\n{}";
+            ::send(cs, hdr.data(), hdr.size(), 0);
+        } else if (stream_oversize_body) {
+            const char hdr[] =
+                "HTTP/1.1 200 OK\r\n"
+                "Content-Type: application/json\r\n"
+                "\r\n";
+            ::send(cs, hdr, sizeof(hdr) - 1, 0);
+            std::string chunk(64 * 1024, 'x');
+            size_t sent = 0;
+            const size_t want = kHttpMaxBodyBytes + chunk.size();
+            while (sent < want) {
+                const ssize_t n = ::send(cs, chunk.data(), chunk.size(), 0);
+                if (n <= 0) {
+                    client_gone.store(true, std::memory_order_release);
+                    break;
+                }
+                sent += static_cast<size_t>(n);
+            }
+        } else {
+            const char reply[] =
+                "HTTP/1.1 200 OK\r\n"
+                "Content-Type: application/json\r\n"
+                "Content-Length: 2\r\n"
+                "\r\n"
+                "{}";
+            ::send(cs, reply, sizeof(reply) - 1, 0);
+        }
+        ::close(cs);
+    }
+
+    ~LoopbackServer() {
+        if (listen_fd >= 0) {
+            ::shutdown(listen_fd, SHUT_RDWR);
+            ::close(listen_fd);
+        }
+        if (acceptor.joinable()) acceptor.join();
+    }
+};
+
+} // namespace
+
+TEST_CASE("http_get accepts a small JSON body") {
+    LoopbackServer srv;
+    srv.start();
+    const std::string url = "http://127.0.0.1:" + std::to_string(srv.port) + "/";
+    auto r = http_get(url, {}, /*timeout=*/5, /*ssrf_guard=*/false);
+    CHECK(r.error.empty());
+    CHECK(r.status_code == 200);
+    CHECK(r.body == "{}");
+}
+
+TEST_CASE("http_get refuses Content-Length over the unary cap") {
+    LoopbackServer srv;
+    srv.advertise_oversize_length = true;
+    srv.start();
+    const std::string url = "http://127.0.0.1:" + std::to_string(srv.port) + "/";
+    auto r = http_get(url, {}, /*timeout=*/5, /*ssrf_guard=*/false);
+    CHECK(r.error.find("size limit") != std::string::npos);
+    CHECK(r.body.empty());
+}
+
+TEST_CASE("rpc_call refuses a streamed body past the unary cap") {
+    LoopbackServer srv;
+    srv.stream_oversize_body = true;
+    srv.start();
+    const std::string url = "http://127.0.0.1:" + std::to_string(srv.port) + "/";
+    auto r = rpc_call(url, {}, "{}", /*timeout=*/10, /*ssrf_guard=*/false);
+    CHECK(r.error.find("size limit") != std::string::npos);
+    CHECK(r.body.empty());
+}