fix(managed-agent): answer a full tool publication reservation with its own code (#14451)

A reservation refused at the tenant byte, Session byte or active-capture
limit answered the generic 400 invalid_request, so the Harness logged
only "The request is invalid." and could not tell a full tenant from a
malformed request. The store now names the limit in the error envelope
(`limit`) and in the message, and the status says whether the refusal
can clear while the Session lives:

- 409 managed_tool_publication_retention_exhausted when used bytes alone
  fill a byte limit, whichever limit failed first. Only the Session's
  retirement and the collection of its output free used bytes.
- 429 managed_tool_publication_capacity_exhausted for every other
  refusal: open captures give back what they hold beyond their used
  bytes when they finish, are fenced, expire or are referenced, and
  leave the active count.

Each refusal logs one WARN line with the limit, its class, the tenant
and Session scope keys, and the held, used, requested and allowed bytes.
The FG8d note and its design doc now say the Harness logged the old
refusal as "The request is invalid.".

Refs #14434

docs/design/2026-10-15-monitor-run-fault-gates-abd.md

@@ -219,7 +219,7 @@ call that launched the run returns, so under the default 60 s lease the
 capture holds its slot for up to a minute after the case moves on. On a
 fast host `drain-release` found all four slots held: its reservation was
 refused (the store's "Publication capacity exhausted", which the Harness
-logs as "The request is invalid."), and the run never started. Under a
+logged as "The request is invalid."), and the run never started. Under a
 10 s lease a capture stops counting within ten seconds of its last
 renewal. FG8d's one store arm, `dropNextStopRevision`, matches a single
 stop-revision commit, so the extra writer and activation renewals never

docs/design/2026-10-15-monitor-run-fault-gates-abd.zh-CN.md

@@ -75,7 +75,7 @@ driver 保留现有三个 arming 原语并新增：
 - **Broker control 通道观测。** control 调用连同 body 的 `operation.kind` 一起记录，让 FG8d 能把「routine 臂什么都没停」断言为维护 wire 上的**缺席**，而不是时序赛跑。
 - **`brokerRequest` helper**，用于 `:release`（routine 与 drain），用捕获的身份与 `protocolVersion: 1`；应答以 `{status, code}` 回给 case 做标志位断言。
 
-FG8d 的 Session 使用 10 s writer lease（`CANCEL_DRAIN_LEASE_MS`），与 FG8c 崩溃 Session 的 lease 等长（`CRASH_LEASE_MS`，已由 #14407 合入）。FG8d 的六个 case 共用一个 tenant，夹具只允许四个活跃 capture（`active-captures=4`）。一个 case 留下的 OPEN capture，其过期时间由最后一次 grant 续约设定（取 writer lease 与 activation 中先到期的那个），而发起该 run 的工具调用一返回，grant 续约就停止，所以默认 60 s lease 下，case 结束后这个 capture 还会占着名额最长约一分钟。快机器跑到 `drain-release` 时，四个名额全被占着：它的 reservation 被拒（store 报 "Publication capacity exhausted"，Harness 日志里是 "The request is invalid."），run 根本没有启动。10 s lease 下，capture 在最后一次续约后十秒以内就不再计入名额。FG8d 唯一的 store 臂 `dropNextStopRevision` 只匹配一条 stop revision 提交，多出来的 writer 与 activation 续约不会误触发它。
+FG8d 的 Session 使用 10 s writer lease（`CANCEL_DRAIN_LEASE_MS`），与 FG8c 崩溃 Session 的 lease 等长（`CRASH_LEASE_MS`，已由 #14407 合入）。FG8d 的六个 case 共用一个 tenant，夹具只允许四个活跃 capture（`active-captures=4`）。一个 case 留下的 OPEN capture，其过期时间由最后一次 grant 续约设定（取 writer lease 与 activation 中先到期的那个），而发起该 run 的工具调用一返回，grant 续约就停止，所以默认 60 s lease 下，case 结束后这个 capture 还会占着名额最长约一分钟。快机器跑到 `drain-release` 时，四个名额全被占着：它的 reservation 被拒（store 报 "Publication capacity exhausted"，Harness 日志里当时只记为 "The request is invalid."），run 根本没有启动。10 s lease 下，capture 在最后一次续约后十秒以内就不再计入名额。FG8d 唯一的 store 臂 `dropNextStopRevision` 只匹配一条 stop revision 提交，多出来的 writer 与 activation 续约不会误触发它。
 
 step/diagnose/FIFO 纪律不变；每个新等待都是状态而不是 sleep；producer 提供 `bg-started.txt`、`progress.txt` 与 gate FIFO；每 case 目录仍是冷加载证据基件（一个后来才启动的 case，其文件必须缺席，而不仅是读不出）。
 

integration-tests/helpers/hosted-monitor-run-fault-driver.ts

@@ -1523,7 +1523,7 @@ async function runStoreFailureCase() {
 // holds its slot for up to a minute after its case moves on. On a fast
 // host drain-release found all four slots held and its reservation was
 // refused (the store's "Publication capacity exhausted", which the Harness
-// logs as "The request is invalid."), so the run never started. Under a
+// logged as "The request is invalid."), so the run never started. Under a
 // 10 s lease a capture stops counting within ten seconds of its last
 // renewal. FG8d's one store arm matches a single stop-revision commit, so
 // the extra writer and activation renewals a shorter lease brings never

packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/managedagent/store/ToolPublicationStore.java

@@ -17,6 +17,8 @@ import com.alibaba.qwen.code.runtimebroker.managedschema.ManagedSchemaTime;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.http.HttpStatus;
 import org.springframework.jdbc.core.ConnectionCallback;
 import org.springframework.jdbc.core.JdbcTemplate;
@@ -53,6 +55,7 @@ public final class ToolPublicationStore {
         }
     }
 
+    private static final Logger LOG = LoggerFactory.getLogger(ToolPublicationStore.class);
     private static final ObjectMapper JSON = new ObjectMapper();
     private final JdbcTemplate jdbc;
     private final TransactionTemplate transactions;
@@ -490,9 +493,36 @@ public final class ToolPublicationStore {
             Long sessionReserved = jdbc.queryForObject("SELECT COALESCE(SUM(capture_held_bytes + producer_held_bytes"
                     + " + admission_held_bytes), 0) FROM ma_output_capture"
                     + " WHERE scope_key = ? AND state <> 'NOT_STARTED'", Long.class, scope);
-            require(reserved <= capacity.tenantBytes() - allocation
-                    && sessionReserved <= capacity.sessionBytes() - allocation
-                    && count < capacity.activeCaptures(), "Publication capacity exhausted");
+            // A full tenant or Session is not a malformed request: it answers
+            // its own code and names the limit that refused. Captures hold
+            // more than they use until they finish, are fenced, expire or are
+            // referenced; only used bytes wait for their Session to retire
+            // and its output to be collected. A byte limit that used bytes
+            // alone fill cannot clear while the batch waits, so it is a 409
+            // whichever limit failed first; any other refusal is a 429.
+            boolean tenantFull = reserved > capacity.tenantBytes() - allocation;
+            boolean sessionFull = sessionReserved > capacity.sessionBytes() - allocation;
+            if (tenantFull || sessionFull) {
+                long tenantUsed = usedBytes("tenant_key", tenantKey);
+                long sessionUsed = usedBytes("scope_key", scope);
+                if (tenantUsed > capacity.tenantBytes() - allocation) {
+                    throw reservationRefused("tenant_bytes", true, tenantKey, scope, reserved, tenantUsed,
+                            allocation, capacity.tenantBytes());
+                }
+                if (sessionUsed > capacity.sessionBytes() - allocation) {
+                    throw reservationRefused("session_bytes", true, tenantKey, scope, sessionReserved,
+                            sessionUsed, allocation, capacity.sessionBytes());
+                }
+                throw tenantFull
+                        ? reservationRefused("tenant_bytes", false, tenantKey, scope, reserved, tenantUsed,
+                                allocation, capacity.tenantBytes())
+                        : reservationRefused("session_bytes", false, tenantKey, scope, sessionReserved,
+                                sessionUsed, allocation, capacity.sessionBytes());
+            }
+            if (count >= capacity.activeCaptures()) {
+                throw reservationRefused("active_captures", false, tenantKey, scope, count, 0, 1,
+                        capacity.activeCaptures());
+            }
             jdbc.update("INSERT INTO ma_output_capture (scope_key, tenant_key, tenant_id, workspace_id,"
                             + " session_id, publication_id, execution_key, capture_id, binding_json,"
                         + " binding_digest, token_hash, state, expires_at, capture_bytes,"
@@ -1111,6 +1141,26 @@ public final class ToolPublicationStore {
         return ToolPublicationContract.parse("grant", grant);
     }
 
+    // The Harness logs only the message, so it names the limit too.
+    private static ApiException reservationRefused(String limit, boolean retained, String tenantKey,
+            String scope, long held, long used, long requested, long allowed) {
+        LOG.warn("Tool publication reservation refused limit={} retained={} tenant={} scope={} held={}"
+                + " used={} requested={} allowed={}", limit, retained, tenantKey, scope, held, used,
+                requested, allowed);
+        String message = "Tool publication reservation refused at the " + limit + " limit";
+        return retained
+                ? new ApiException(HttpStatus.CONFLICT,
+                        "managed_tool_publication_retention_exhausted", message, Map.of("limit", limit))
+                : new ApiException(HttpStatus.TOO_MANY_REQUESTS,
+                        "managed_tool_publication_capacity_exhausted", message, Map.of("limit", limit));
+    }
+
+    private long usedBytes(String column, String key) {
+        return jdbc.queryForObject("SELECT COALESCE(SUM(capture_used_bytes + producer_used_bytes"
+                + " + admission_used_bytes), 0) FROM ma_output_capture WHERE " + column + " = ?"
+                + " AND state <> 'NOT_STARTED'", Long.class, key);
+    }
+
     private static long requireCurrentActivation(Long expiresAt, long nowEpoch, boolean deferLapse) {
         if (expiresAt == null || expiresAt <= nowEpoch) {
             if (deferLapse && expiresAt != null) {

packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/ToolPublicationStoreTest.java

@@ -1,5 +1,9 @@
 package com.alibaba.qwen.code.managedagent;
 
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
 import com.alibaba.qwen.code.managedagent.api.ApiException;
 import com.alibaba.qwen.code.managedagent.config.ManagedAgentProperties;
 import com.alibaba.qwen.code.managedagent.service.ManagedArtifactPolicy;
@@ -27,6 +31,7 @@ import com.alibaba.qwen.code.runtimebroker.managedschema.ManagedSchemaTime;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.assertj.core.api.ThrowableAssert.ThrowingCallable;
 import org.h2.jdbcx.JdbcDataSource;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -34,6 +39,8 @@ import org.junit.jupiter.api.io.TempDir;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.CsvSource;
 import org.junit.jupiter.params.provider.ValueSource;
+import org.slf4j.LoggerFactory;
+import org.springframework.http.HttpStatus;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.jdbc.datasource.DataSourceTransactionManager;
 import org.springframework.jdbc.datasource.DelegatingDataSource;
@@ -2389,11 +2396,219 @@ class ToolPublicationStoreTest {
         assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_output_capture", Integer.class)).isEqualTo(1);
         jdbc.update("DELETE FROM ma_output_capture");
         store = newStore(ALLOCATION - 1, 1);
-        assertThatThrownBy(this::reserve).hasMessageContaining("capacity exhausted");
+        assertRefused(this::reserve, true, "tenant_bytes");
         store = newStore(ALLOCATION, 1);
         assertThat(reserve().path("captureBytes").asLong()).isEqualTo(CAPTURE_BYTES);
     }
 
+    @Test
+    void refusesEachExhaustedLimitWithItsOwnCode() {
+        ObjectNode second = addSecondExecution();
+        store = new ToolPublicationStore(jdbc, manager, sessions, executions, bindings,
+                new ToolPublicationStore.Capacity(CAPTURE_BYTES * 2, 10 * ALLOCATION, ALLOCATION - 1, 10));
+        assertRefused(this::reserve, true, "tenant_bytes");
+        store = new ToolPublicationStore(jdbc, manager, sessions, executions, bindings,
+                new ToolPublicationStore.Capacity(CAPTURE_BYTES * 2, ALLOCATION - 1, 10 * ALLOCATION, 10));
+        assertRefused(this::reserve, true, "session_bytes");
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_output_capture", Integer.class)).isZero();
+        store = new ToolPublicationStore(jdbc, manager, sessions, executions, bindings,
+                new ToolPublicationStore.Capacity(CAPTURE_BYTES * 2, 10 * ALLOCATION, 10 * ALLOCATION, 1));
+        reserve();
+        ObjectNode secondRequest = request("reserve");
+        secondRequest.set("binding", second);
+        assertRefused(() -> store.apply(secondRequest, WRITER_TOKEN, PUBLICATION_TOKEN), false, "active_captures");
+        // A fenced capture stops counting, so the refused reservation now fits.
+        store.apply(request("fence"), WRITER_TOKEN, null);
+        assertThat(store.apply(secondRequest, WRITER_TOKEN, PUBLICATION_TOKEN).path("state").asText())
+                .isEqualTo("OPEN");
+    }
+
+    @Test
+    void namesTheByteLimitWhenTheCaptureSlotsAreFullToo() {
+        ObjectNode second = addSecondExecution();
+        store = new ToolPublicationStore(jdbc, manager, sessions, executions, bindings,
+                new ToolPublicationStore.Capacity(CAPTURE_BYTES * 2, ALLOCATION, 10 * ALLOCATION, 1));
+        reserve();
+        ObjectNode secondRequest = request("reserve");
+        secondRequest.set("binding", second);
+        // The Session's bytes and its one slot are both taken by an open
+        // capture: the byte limit is named, and it is retryable, because the
+        // capture gives both back when it ends.
+        assertRefused(() -> store.apply(secondRequest, WRITER_TOKEN, PUBLICATION_TOKEN), false, "session_bytes");
+        store.apply(request("fence"), WRITER_TOKEN, null);
+        assertThat(store.apply(secondRequest, WRITER_TOKEN, PUBLICATION_TOKEN).path("state").asText())
+                .isEqualTo("OPEN");
+    }
+
+    @Test
+    void refusesWithoutRetryOnlyWhenUsedBytesAloneFillTheLimit() {
+        ObjectNode second = addSecondExecution();
+        store = new ToolPublicationStore(jdbc, manager, sessions, executions, bindings,
+                new ToolPublicationStore.Capacity(CAPTURE_BYTES * 2, ALLOCATION, 10 * ALLOCATION, 10));
+        reserve();
+        ObjectNode secondRequest = request("reserve");
+        secondRequest.set("binding", second);
+        // One byte of the open capture is used: only that byte outlives it.
+        jdbc.update("UPDATE ma_output_capture SET capture_used_bytes = 1");
+        store = new ToolPublicationStore(jdbc, manager, sessions, executions, bindings,
+                new ToolPublicationStore.Capacity(CAPTURE_BYTES * 2, ALLOCATION + 1, 10 * ALLOCATION, 10));
+        assertRefused(() -> store.apply(secondRequest, WRITER_TOKEN, PUBLICATION_TOKEN), false, "session_bytes");
+        store = new ToolPublicationStore(jdbc, manager, sessions, executions, bindings,
+                new ToolPublicationStore.Capacity(CAPTURE_BYTES * 2, ALLOCATION, 10 * ALLOCATION, 10));
+        assertRefused(() -> store.apply(secondRequest, WRITER_TOKEN, PUBLICATION_TOKEN), true, "session_bytes");
+        // A finished capture keeps everything it used until its Session is
+        // collected: nothing about it frees on its own.
+        jdbc.update("UPDATE ma_output_capture SET state = 'FENCED', expires_at = NULL,"
+                + " capture_used_bytes = capture_held_bytes, producer_used_bytes = producer_held_bytes,"
+                + " admission_used_bytes = admission_held_bytes");
+        store = new ToolPublicationStore(jdbc, manager, sessions, executions, bindings,
+                new ToolPublicationStore.Capacity(CAPTURE_BYTES * 2, 2 * ALLOCATION - 1, 10 * ALLOCATION, 10));
+        assertRefused(() -> store.apply(secondRequest, WRITER_TOKEN, PUBLICATION_TOKEN), true, "session_bytes");
+    }
+
+    @Test
+    void anotherSessionsOpenCaptureDoesNotFreeThisSessionsRetainedBytes() {
+        ObjectNode second = addSecondExecution();
+        store = new ToolPublicationStore(jdbc, manager, sessions, executions, bindings,
+                new ToolPublicationStore.Capacity(CAPTURE_BYTES * 2, 10 * ALLOCATION, 10 * ALLOCATION, 10));
+        reserve();
+        ObjectNode secondRequest = request("reserve");
+        secondRequest.set("binding", second);
+        store.apply(secondRequest, WRITER_TOKEN, PUBLICATION_TOKEN);
+        // The first capture finished here with everything used; the second
+        // is still open but now belongs to another Session of the tenant.
+        jdbc.update("UPDATE ma_output_capture SET state = 'FENCED', expires_at = NULL,"
+                + " capture_used_bytes = capture_held_bytes, producer_used_bytes = producer_held_bytes,"
+                + " admission_used_bytes = admission_held_bytes WHERE publication_id = ?",
+                "pub-1".getBytes(StandardCharsets.UTF_8));
+        jdbc.update("UPDATE ma_output_capture SET scope_key = ? WHERE publication_id = ?", "f".repeat(64),
+                "pub-2".getBytes(StandardCharsets.UTF_8));
+        // The tenant fails first, and its open capture could make room
+        // there; the Session's used bytes still leave none.
+        store = new ToolPublicationStore(jdbc, manager, sessions, executions, bindings,
+                new ToolPublicationStore.Capacity(CAPTURE_BYTES * 2, 2 * ALLOCATION - 1, 3 * ALLOCATION - 1, 10));
+        assertRefused(() -> store.apply(secondRequest, WRITER_TOKEN, PUBLICATION_TOKEN), true, "session_bytes");
+    }
+
+    @Test
+    void anotherSessionsUsedBytesDoNotFillThisSessionsLimit() {
+        ObjectNode second = addSecondExecution();
+        store = new ToolPublicationStore(jdbc, manager, sessions, executions, bindings,
+                new ToolPublicationStore.Capacity(CAPTURE_BYTES * 2, 10 * ALLOCATION, 10 * ALLOCATION, 10));
+        ObjectNode firstRequest = request("reserve");
+        reserve();
+        ObjectNode secondRequest = request("reserve");
+        secondRequest.set("binding", second);
+        store.apply(secondRequest, WRITER_TOKEN, PUBLICATION_TOKEN);
+        // The first capture finished with everything used, in another Session
+        // of the tenant; this Session holds only the second, still open.
+        jdbc.update("UPDATE ma_output_capture SET state = 'FENCED', expires_at = NULL, scope_key = ?,"
+                + " capture_used_bytes = capture_held_bytes, producer_used_bytes = producer_held_bytes,"
+                + " admission_used_bytes = admission_held_bytes WHERE publication_id = ?", "f".repeat(64),
+                "pub-1".getBytes(StandardCharsets.UTF_8));
+        store = new ToolPublicationStore(jdbc, manager, sessions, executions, bindings,
+                new ToolPublicationStore.Capacity(CAPTURE_BYTES * 2, 2 * ALLOCATION - 1, 10 * ALLOCATION, 10));
+        assertRefused(() -> store.apply(firstRequest, WRITER_TOKEN, PUBLICATION_TOKEN), false, "session_bytes");
+    }
+
+    @Test
+    void anUnreferencedAdmissionDoesNotCountAsRetained() {
+        ObjectNode second = addSecondExecution();
+        store = new ToolPublicationStore(jdbc, manager, sessions, executions, bindings,
+                new ToolPublicationStore.Capacity(CAPTURE_BYTES * 2, 10 * ALLOCATION, 10 * ALLOCATION, 10));
+        reserve();
+        ObjectNode secondRequest = request("reserve");
+        secondRequest.set("binding", second);
+        // Finished and fenced with its output fully used, but not yet
+        // referenced: the receipt still gives back the admission it did not use.
+        jdbc.update("UPDATE ma_output_capture SET state = 'FENCED', expires_at = NULL, producer_phase = 'FINISHED',"
+                + " capture_used_bytes = capture_held_bytes, producer_used_bytes = producer_held_bytes,"
+                + " admission_used_bytes = 100");
+        store = new ToolPublicationStore(jdbc, manager, sessions, executions, bindings,
+                new ToolPublicationStore.Capacity(CAPTURE_BYTES * 2, 2 * ALLOCATION - 1, 10 * ALLOCATION, 10));
+        assertRefused(() -> store.apply(secondRequest, WRITER_TOKEN, PUBLICATION_TOKEN), false, "session_bytes");
+        jdbc.update("UPDATE ma_output_capture SET producer_phase = 'REFERENCED',"
+                + " admission_held_bytes = admission_used_bytes");
+        assertThat(store.apply(secondRequest, WRITER_TOKEN, PUBLICATION_TOKEN).path("state").asText())
+                .isEqualTo("OPEN");
+    }
+
+    @Test
+    void logsEachRefusalWithWhatItWeighed() {
+        Logger log = (Logger) LoggerFactory.getLogger(ToolPublicationStore.class);
+        ListAppender<ILoggingEvent> logged = new ListAppender<>();
+        logged.start();
+        log.addAppender(logged);
+        try {
+            ObjectNode second = addSecondExecution();
+            store = new ToolPublicationStore(jdbc, manager, sessions, executions, bindings,
+                    new ToolPublicationStore.Capacity(CAPTURE_BYTES * 2, 10 * ALLOCATION, 10 * ALLOCATION, 10));
+            reserve();
+            Map<String, Object> keys = jdbc.queryForMap("SELECT tenant_key, scope_key FROM ma_output_capture");
+            // Distinct numbers: the capture holds 10 bytes less than an
+            // allocation, has used 3, and now belongs to another Session.
+            jdbc.update("UPDATE ma_output_capture SET capture_held_bytes = capture_held_bytes - 10,"
+                    + " capture_used_bytes = 3, scope_key = ?", "f".repeat(64));
+            ObjectNode secondRequest = request("reserve");
+            secondRequest.set("binding", second);
+            store = new ToolPublicationStore(jdbc, manager, sessions, executions, bindings,
+                    new ToolPublicationStore.Capacity(CAPTURE_BYTES * 2, 10 * ALLOCATION, ALLOCATION + 5, 10));
+            assertRefused(() -> store.apply(secondRequest, WRITER_TOKEN, PUBLICATION_TOKEN), false, "tenant_bytes");
+            store = new ToolPublicationStore(jdbc, manager, sessions, executions, bindings,
+                    new ToolPublicationStore.Capacity(CAPTURE_BYTES * 2, 10 * ALLOCATION, ALLOCATION + 2, 10));
+            assertRefused(() -> store.apply(secondRequest, WRITER_TOKEN, PUBLICATION_TOKEN), true, "tenant_bytes");
+            store = new ToolPublicationStore(jdbc, manager, sessions, executions, bindings,
+                    new ToolPublicationStore.Capacity(CAPTURE_BYTES * 2, 10 * ALLOCATION, 10 * ALLOCATION, 1));
+            assertRefused(() -> store.apply(secondRequest, WRITER_TOKEN, PUBLICATION_TOKEN), false, "active_captures");
+            String identity = " tenant=" + keys.get("tenant_key") + " scope=" + keys.get("scope_key");
+            String held = " held=" + (ALLOCATION - 10) + " used=3 requested=" + ALLOCATION;
+            assertThat(logged.list).extracting(ILoggingEvent::getFormattedMessage).containsExactly(
+                    "Tool publication reservation refused limit=tenant_bytes retained=false" + identity + held
+                            + " allowed=" + (ALLOCATION + 5),
+                    "Tool publication reservation refused limit=tenant_bytes retained=true" + identity + held
+                            + " allowed=" + (ALLOCATION + 2),
+                    "Tool publication reservation refused limit=active_captures retained=false" + identity
+                            + " held=1 used=0 requested=1 allowed=1");
+            assertThat(logged.list).extracting(ILoggingEvent::getLevel).containsOnly(Level.WARN);
+        } finally {
+            log.detachAppender(logged);
+        }
+    }
+
+    @Test
+    void eachByteLimitChargesItsOwnSum() {
+        ObjectNode second = addSecondExecution();
+        store = new ToolPublicationStore(jdbc, manager, sessions, executions, bindings,
+                new ToolPublicationStore.Capacity(CAPTURE_BYTES * 2, 10 * ALLOCATION, 10 * ALLOCATION, 10));
+        reserve();
+        // Hand the first capture to another Session of the same tenant: the
+        // tenant still carries its bytes, this Session no longer does.
+        jdbc.update("UPDATE ma_output_capture SET scope_key = ?", "f".repeat(64));
+        ObjectNode secondRequest = request("reserve");
+        secondRequest.set("binding", second);
+        store = new ToolPublicationStore(jdbc, manager, sessions, executions, bindings,
+                new ToolPublicationStore.Capacity(CAPTURE_BYTES * 2, 10 * ALLOCATION, ALLOCATION, 10));
+        assertRefused(() -> store.apply(secondRequest, WRITER_TOKEN, PUBLICATION_TOKEN), false, "tenant_bytes");
+        store = new ToolPublicationStore(jdbc, manager, sessions, executions, bindings,
+                new ToolPublicationStore.Capacity(CAPTURE_BYTES * 2, ALLOCATION, 10 * ALLOCATION, 10));
+        assertThat(store.apply(secondRequest, WRITER_TOKEN, PUBLICATION_TOKEN).path("state").asText())
+                .isEqualTo("OPEN");
+    }
+
+    private static void assertRefused(ThrowingCallable reservation, boolean retained, String limit) {
+        assertThatThrownBy(reservation).isInstanceOfSatisfying(ApiException.class,
+                error -> assertRefusal(error, retained, limit));
+    }
+
+    // A refusal that used bytes alone cause is a 409; any other is a 429.
+    private static void assertRefusal(ApiException error, boolean retained, String limit) {
+        assertThat(error.getStatus()).isEqualTo(retained ? HttpStatus.CONFLICT : HttpStatus.TOO_MANY_REQUESTS);
+        assertThat(error.getCode()).isEqualTo(retained ? "managed_tool_publication_retention_exhausted"
+                : "managed_tool_publication_capacity_exhausted");
+        assertThat(error.getMessage()).isEqualTo("Tool publication reservation refused at the " + limit + " limit");
+        assertThat(error.getDetails()).isEqualTo(Map.of("limit", limit));
+    }
+
     @Test
     void refusesFirstReservationAfterDispatchHasBeenClaimed() {
         var execution = executions.claimDispatch("execution-1", "dispatcher", java.time.Duration.ofMinutes(1));
@@ -2468,7 +2683,8 @@ class ToolPublicationStoreTest {
             outcomes = List.of(one.get(), two.get());
         }
         assertThat(outcomes.stream().filter(JsonNode.class::isInstance).count()).isEqualTo(1);
-        assertThat(outcomes.stream().filter(IllegalArgumentException.class::isInstance).count()).isEqualTo(1);
+        assertThat(outcomes).filteredOn(ApiException.class::isInstance).singleElement()
+                .satisfies(error -> assertRefusal((ApiException) error, false, "tenant_bytes"));
         String winner = jdbc.queryForObject("SELECT publication_id FROM ma_output_capture", String.class);
         ObjectNode loser = "pub-1".equals(winner) ? secondRequest : firstRequest;
         ObjectNode fence = request("fence").put("publicationId", winner);
@@ -2487,7 +2703,7 @@ class ToolPublicationStoreTest {
         start.await();
         try {
             return store.apply(request, WRITER_TOKEN, PUBLICATION_TOKEN);
-        } catch (IllegalArgumentException error) {
+        } catch (ApiException error) {
             return error;
         }
     }