diff --git a/monarch-benchmark/workflowbench/tests/test_external_catalogue.py b/monarch-benchmark/workflowbench/tests/test_external_catalogue.py
index 30c4fffc..8e4b70a0 100644
--- a/monarch-benchmark/workflowbench/tests/test_external_catalogue.py
+++ b/monarch-benchmark/workflowbench/tests/test_external_catalogue.py
@@ -153,3 +153,51 @@ def test_mixed_scalar_union_remains_exact_inside_root_json_payload(source,tmp_pa
     assert payload["type"] == "object"
     assert json.loads(payload["constraints"]["helper_text"].split("Source JSON Schema: ",1)[1]) == schema
     assert payload["required"] is True
+
+
+BULK_DELETE = {"delete": {"operationId": "bulk_delete", "summary": "Delete several messages",
+    "requestBody": {"required": True, "content": {"application/json": {
+        "schema": {"type": "object", "properties": {"ids": {"type": "string"}}, "required": ["ids"]}}}},
+    "responses": {"200": {"content": {"application/json": {
+        "schema": {"type": "object", "properties": {"ok": {"type": "string"}}, "required": ["ok"]}}}}}}}
+
+
+def test_an_operation_monarch_cannot_represent_is_excluded_and_reported(source, tmp_path):
+    """One unrepresentable operation used to abort the whole product's catalogue.
+
+    `_action` raised, nothing caught it, `monarch_setup` turned it into
+    `Stop(2, "generate", ...)` and no seed file was written -- so a task set naming any
+    of the four bodyless-body DELETEs in the real EnterpriseOps-Gym ITSM document could
+    not set Monarch up on that product at all. Being unable to express one operation is
+    not a reason to teach Monarch nothing.
+    """
+    product, tasks, doc, closed = source
+    doc["paths"]["/messages/bulk"] = copy.deepcopy(BULK_DELETE)
+    result = catalog.generate(product, tasks, tmp_path / "seeds", PUBLIC)
+    out = tmp_path / "seeds"
+    assert result.operations_in_spec == result.files_written == 1, "the representable operation is still taught"
+    assert [(e["service"], e["method"], e["path"]) for e in result.excluded] == [("gmail", "delete", "/messages/bulk")]
+    assert "bodyless" in result.excluded[0]["reason"]
+
+    # The pack says what Monarch was not taught, and never advertises it either.
+    assert json.loads((out / "ok.txt").read_text())["excluded_operations"] == result.excluded
+    assert "/messages/bulk" not in (out / "source-contracts.yaml").read_text()
+
+
+def test_a_catalogue_with_nothing_excluded_is_byte_for_byte_what_it_was(source, tmp_path):
+    """The manifest gains the field only when there is something to report, so every
+    pack frozen before this change still regenerates identically -- `generate` refuses
+    to replace a frozen catalogue file, so a new key on every pack would break setup
+    for anyone holding one."""
+    product, tasks, doc, closed = source
+    catalog.generate(product, tasks, tmp_path / "seeds", PUBLIC)
+    manifest = json.loads((tmp_path / "seeds" / "ok.txt").read_text())
+    assert "excluded_operations" not in manifest
+
+
+def test_an_unrepresentable_operation_is_the_only_thing_excluded(source, tmp_path):
+    """Excluding on any ValueError would hide real structural faults in a document."""
+    product, tasks, doc, closed = source
+    doc["paths"]["/messages/{message_id}"]["patch"]["servers"] = [{"url": "https://elsewhere.test"}]
+    with pytest.raises(ValueError, match="server overrides"):
+        catalog.generate(product, tasks, tmp_path / "seeds", PUBLIC)
diff --git a/monarch-benchmark/workflowbench/tests/test_external_customer.py b/monarch-benchmark/workflowbench/tests/test_external_customer.py
index dd05b969..4a22b798 100644
--- a/monarch-benchmark/workflowbench/tests/test_external_customer.py
+++ b/monarch-benchmark/workflowbench/tests/test_external_customer.py
@@ -175,15 +175,20 @@ def test_attach_customer_failure_closes_budget_and_preserves_customer_billing(mo
     assert ledger.status().held_usd == ledger.reservations()[0].maximum_usd > 0
 
 
-@pytest.mark.parametrize("reason,kind", [
-    ("scope budget exhausted", "infra:budget"), ("run budget exhausted", "infra:budget"),
-    ("shared weekly budget exhausted", "infra:weekly_budget"),
-    ("recorded reservation overrun blocks further launches", "infra:weekly_budget")])
-def test_direct_arm_budget_refusal_is_infrastructure_with_partial_cost(monkeypatch, tmp_path, reason, kind):
+@pytest.mark.parametrize("reason,scope,kind", [
+    # Both budget refusals say `run budget exhausted`, because an attempt's own cap is
+    # a run reservation too. The scope separates them: the round's envelope stops the
+    # round, one attempt reaching its cap does not. See STOPS_THE_ROUND.
+    ("scope budget exhausted", "task/arm/t0#attempt-000", "infra:budget"),
+    ("run budget exhausted", "task/arm/t0#attempt-000", "infra:budget"),
+    ("run budget exhausted", "run-7#admission-000", "infra:run_budget"),
+    ("shared weekly budget exhausted", None, "infra:weekly_budget"),
+    ("recorded reservation overrun blocks further launches", None, "infra:weekly_budget")])
+def test_direct_arm_budget_refusal_is_infrastructure_with_partial_cost(monkeypatch, tmp_path, reason, scope, kind):
     config, ledger, calls = setup(monkeypatch, tmp_path, [], participant=False)
 
     def action(ep):
-        exc = BudgetExceeded(reason)
+        exc = BudgetExceeded(reason, scope_id=scope)
         exc.partial = ArmResult(cost_usd=0.2, tokens_output=4)
         raise exc
 
@@ -224,3 +229,83 @@ def action(ep):
     assert result.tokens_prompt == 1000 and result.flags == []
     assert calls[0][1] == [{"type": "function", **schema}]
     assert ledger.status().held_usd == 0
+
+
+def test_an_agent_timeout_keeps_its_verdict_when_the_customer_hit_the_same_deadline(monkeypatch, tmp_path):
+    """An expired episode deadline is a scored `timeout` verdict by deliberate rule
+    (`wb_arms/api_loop.py:670-673`), not an infrastructure retry.
+
+    The simulated customer is cut at that same deadline and records `infra:timeout`.
+    Letting the customer's sticky copy overwrite the arm's verdict dropped a real
+    agent timeout out of the pass denominator, so tau2 pass rates came out inflated
+    by exactly the attempts that ran out of time -- whenever the agent's last turn
+    happened to call the customer.
+    """
+    config, ledger, calls = setup(monkeypatch, tmp_path, [])
+    ep = Episode()
+
+    def action(ep):
+        assert "deadline" in fetch(ep)["error"]
+        return ArmResult(cost_usd=0.25, termination="timeout", final_text="ran out of time")
+
+    result = run(config, ledger, ep, action, deadline=time.monotonic() - 1)
+    assert result.termination == "timeout", "the customer overwrote the arm's scored verdict"
+    assert calls == [], "the deadline was already gone; no provider call should be made"
+    rows = [dict(task_id="task", arm="native", model="native", trial=0, passed=False,
+                 termination=result.termination, flags=result.flags, cost_usd=result.cost_usd)]
+    metrics = competitor_metrics(rows, 1)
+    assert metrics["strict_pass_denominator"] == 1 and metrics["infra"] == 0
+
+
+def test_a_customer_failure_that_is_not_the_deadline_still_wins(monkeypatch, tmp_path):
+    """The guard's own purpose, kept: a swallowed customer break must not be scored,
+    even when the arm goes on to finish and hand back a verdict."""
+    config, ledger, calls = setup(monkeypatch, tmp_path, [RuntimeError("provider disconnected")])
+    ep = Episode()
+
+    def action(ep):
+        assert "provider disconnected" in fetch(ep)["error"]
+        return ArmResult(cost_usd=0.25, termination="timeout", final_text="finished anyway")
+
+    with pytest.raises(InfraError) as caught:
+        run(config, ledger, ep, action)
+    assert caught.value.kind == "infra:customer"
+
+
+def test_only_the_round_wide_refusals_stop_the_round():
+    """The rule the kinds exist to carry, stated where it can be broken.
+
+    Stopping on `infra:budget` too would be a worse defect than the one the third kind
+    fixes: that kind is the correct, tested outcome of an attempt exhausting its own
+    scope cap (see the parametrized case above), so a round would die the first time
+    any single attempt reached its per-attempt cap.
+    """
+    from wb_orchestrator.orchestrator import STOPS_THE_ROUND
+    assert STOPS_THE_ROUND == {"infra:weekly_budget": "weekly_budget",
+                               "infra:run_budget": "run_budget"}
+    assert "infra:budget" not in STOPS_THE_ROUND
+
+
+def test_the_scope_tells_a_round_envelope_from_an_attempt_cap(tmp_path):
+    """The two refusals are word-for-word identical, so the scope is the discriminator.
+
+    An attempt's own cap is a run reservation too -- that is why matching on the
+    message classified every capped attempt as a round-wide stop. The round's envelope
+    is the one whose scope carries `ROUND_ENVELOPE_MARKER`.
+    """
+    from wb_orchestrator.budget import ROUND_ENVELOPE_MARKER, BudgetLedger
+    ledger = BudgetLedger(tmp_path / "ledger.sqlite", weekly_limit_usd="100")
+    refusals = {}
+    for scope in ("round-7" + ROUND_ENVELOPE_MARKER + "000", "task/arm/t0#attempt-000"):
+        ledger.reserve_run(scope, "1.00")
+        ledger.reserve(scope + "/first", "0.80", scope_id=scope, run_id=scope)
+        with pytest.raises(BudgetExceeded) as caught:
+            ledger.reserve(scope + "/second", "0.80", scope_id=scope, run_id=scope)
+        refusals[scope] = caught.value
+
+    messages = {str(exc) for exc in refusals.values()}
+    assert messages == {"run budget exhausted"}, "identical wording is the whole problem"
+    kinds = {scope: runtime._budget_failure(exc).kind for scope, exc in refusals.items()}
+    assert sorted(kinds.values()) == ["infra:budget", "infra:run_budget"]
+    assert kinds["round-7" + ROUND_ENVELOPE_MARKER + "000"] == "infra:run_budget"
+    assert ledger.status().available_usd > 0, "the week had room; only these scopes did not"
diff --git a/monarch-benchmark/workflowbench/tests/test_external_monarch_setup.py b/monarch-benchmark/workflowbench/tests/test_external_monarch_setup.py
index c9d9600f..26edca13 100644
--- a/monarch-benchmark/workflowbench/tests/test_external_monarch_setup.py
+++ b/monarch-benchmark/workflowbench/tests/test_external_monarch_setup.py
@@ -149,7 +149,7 @@ def test_source_setup_refuses_import_with_unverified_result(source_setup, monkey
 
 def test_external_kb_rejects_legacy_service_slug(source_setup):
     path = source_setup.product.with_name("enterprise-ops-gym.monarch-kb.yaml")
-    monarch_setup._write_kb(source_setup.product, "enterprise-ops-gym", URL, {"bench-gym-itsm-mcp": "old"})
+    monarch_setup._write_kb(source_setup.product, "enterprise-ops-gym", URL, {"bench-gym-itsm-mcp": "old"}, env={})
     with pytest.raises(config.ConfigError, match=SLUG):
         config.load_monarch_kb(path, config.load_product(source_setup.product))
 
diff --git a/monarch-benchmark/workflowbench/tests/test_external_pipeline.py b/monarch-benchmark/workflowbench/tests/test_external_pipeline.py
index b8a0f527..6cfead10 100644
--- a/monarch-benchmark/workflowbench/tests/test_external_pipeline.py
+++ b/monarch-benchmark/workflowbench/tests/test_external_pipeline.py
@@ -316,3 +316,37 @@ def test_budget_capability_status_uses_fresh_exact_harness_verification(tmp_path
     record['checked_at'] = (datetime.now(timezone.utc)-timedelta(hours=3)).isoformat()
     monarch_probe.probe_path(site).write_text(json.dumps(record))
     assert approvals.capabilities([harness], env)['monarch'] == approvals.MONARCH_REASON
+
+
+def test_a_failure_after_admission_releases_the_round_envelope(tmp_path, monkeypatch):
+    """The envelope was closed by `_execute`'s finally alone, so anything raising
+    between `_admit` and `_execute` left the round's whole liability held against the
+    week with nothing spent.
+
+    A duplicate `--run-id` is enough: `_admit` reserves the envelope, `create_run`
+    raises on the unique constraint, and `_execute` is never entered. Nothing can
+    release a run envelope afterwards -- `wb budget release` only looks in
+    `budget_reservations` -- and the scope never appears in `status.unknown_ids`, so
+    the operator cannot even name what is holding the money.
+    """
+    from types import SimpleNamespace
+    from wb_orchestrator.budget import ROUND_ENVELOPE_MARKER, BudgetLedger
+    pp, pl = product_plan(tmp_path, external_task())
+    rc = config.resolve(pp, pl, env={})
+    ledger = BudgetLedger(tmp_path / 'ledger.sqlite3', weekly_limit_usd='100')
+    store = Store(tmp_path / 'wb.sqlite3')
+    engine = orchestrator.Orchestrator.from_config(store, rc, tmp_path / 'out', ledger=ledger)
+    # A paid competitor, so admission opens an envelope at all, and a world that needs
+    # no container: this test is about the round's lifecycle, not about either of those.
+    paid = SimpleNamespace(name='paid', provider_key='mock', run=lambda ep, deadline=None: None)
+    monkeypatch.setattr(engine, '_arms', lambda: [paid])
+    monkeypatch.setattr(engine, 'world', SimpleNamespace(prerequisites=lambda: []))
+
+    store.create_run('run-x', engine._hash(), engine.suite, engine._config())
+    with pytest.raises(Exception):
+        engine.run('run-x')                       # UNIQUE constraint on runs.run_id
+
+    envelopes = [r for r in ledger.run_reservations() if ROUND_ENVELOPE_MARKER in r.scope_id]
+    assert envelopes, 'admission should have opened one, or this proves nothing'
+    assert all(r.closed_at is not None for r in envelopes), 'the round envelope leaked'
+    assert ledger.status().held_usd == 0
diff --git a/monarch-benchmark/workflowbench/tests/test_front_door_relay.py b/monarch-benchmark/workflowbench/tests/test_front_door_relay.py
index 10e2afa8..de5fb766 100644
--- a/monarch-benchmark/workflowbench/tests/test_front_door_relay.py
+++ b/monarch-benchmark/workflowbench/tests/test_front_door_relay.py
@@ -23,8 +23,9 @@
 from http.server import ThreadingHTTPServer
 
 import pytest
+import yaml
 
-from wb_orchestrator.monarch_setup import front_door_path
+from wb_orchestrator.monarch_setup import _write_kb, front_door_path
 from wb_studio.app import ROOT, Studio, front_door_target, handler
 from wb_world.episode import load_suite
 
@@ -142,3 +143,31 @@ def test_the_secret_segment_is_built_into_the_seed_url():
 
 def test_a_seed_url_without_a_secret_is_unchanged():
     assert front_door_path("https://studio.example.dev/front-door", {}) == "https://studio.example.dev/front-door"
+
+
+def test_the_tracked_knowledge_base_records_the_secret_by_name(tmp_path):
+    """`config/products/<product>.monarch-kb.yaml` is git-tracked and this repo is
+    public, so a secret written into it once is in history, not just in a file.
+    Only the config hash reads this field, and never as an address, so the name
+    carries everything the record needs."""
+    env = {"STUDIO_FRONT_DOOR_SECRET": SECRET}
+    live = front_door_path("https://studio.example.dev/front-door", env)
+    assert SECRET in live, "the address Monarch is handed must carry the secret"
+    path, _ = _write_kb(tmp_path / "tau2-retail.yaml", "tau2-retail", live,
+                        {"bench-gym-itsm-mcp": "sha"}, env=env)
+    text = path.read_text(encoding="utf-8")
+    assert SECRET not in text, "the secret reached a tracked file"
+    assert "${STUDIO_FRONT_DOOR_SECRET}" in text
+
+
+def test_rotating_the_secret_leaves_the_configuration_unchanged(tmp_path):
+    """Rotating per round is what the docstring tells an operator to do. Recording
+    the value made that read as a different configuration and moved the hash."""
+    written = []
+    for secret in ("round-47", "round-48"):
+        env = {"STUDIO_FRONT_DOOR_SECRET": secret}
+        live = front_door_path("https://studio.example.dev/front-door", env)
+        path, _ = _write_kb(tmp_path / "tau2-retail.yaml", "tau2-retail", live,
+                            {"bench-gym-itsm-mcp": "sha"}, env=env)
+        written.append(yaml.safe_load(path.read_text(encoding="utf-8"))["shim_public_url"])
+    assert written[0] == written[1]
diff --git a/monarch-benchmark/workflowbench/tests/test_worlds_enterprise_ops.py b/monarch-benchmark/workflowbench/tests/test_worlds_enterprise_ops.py
index 121eb787..9cbd997b 100644
--- a/monarch-benchmark/workflowbench/tests/test_worlds_enterprise_ops.py
+++ b/monarch-benchmark/workflowbench/tests/test_worlds_enterprise_ops.py
@@ -242,6 +242,60 @@ def test_search_does_not_surface_them_either():
     assert not any(h["path"].startswith("/api/") for h in hits)
 
 
+class _Recording:
+    """A server that records the target it was handed instead of sending it."""
+
+    def __init__(self):
+        self.target = None
+
+    def request(self, method, path, body=None, extra=None):
+        self.target = path
+        return b'{"ok": true}'
+
+
+def _with_location_route():
+    world = _Offline()
+    world._specs["gym-itsm-mcp"]["paths"]["/locations/name/{name}"] = {
+        "get": {"summary": "location by name", "operationId": "location_by_name"}}
+    world.servers["gym-itsm-mcp"] = _Recording()
+    return world, world.servers["gym-itsm-mcp"]
+
+
+def test_a_path_value_with_a_space_reaches_the_world_still_encoded():
+    """`/locations/name/TechCorp%20NYC%20Headquarters` is a route and a value the
+    shipped ITSM document and its seed data both contain.
+
+    The decoded path was sent as the request target. `urllib` refuses a space
+    outright -- `InvalidURL: URL can't contain control characters` -- and a server
+    that accepted it would read the target as `/locations/name/TechCorp`. Either way
+    a real EOG task fails, and the approval rule reads that as the competitor
+    failing rather than as our own bug.
+    """
+    world, server = _with_location_route()
+    out = world._call("GET", "/locations/name/TechCorp%20NYC%20Headquarters", None, None)
+    assert json.loads(out) == {"ok": True}, out
+    assert " " not in server.target, "a space in the request target is not sendable"
+    assert server.target == "/locations/name/TechCorp%20NYC%20Headquarters"
+
+
+def test_the_permission_check_reads_the_decoded_path():
+    """Decoding still has to happen, just not on the way to the wire: an encoded
+    separator must not slip past the published-surface check as one segment."""
+    world, server = _with_location_route()
+    out = json.loads(world._call("GET", "/locations/name/Tech%2FCorp", None, None))
+    assert out["error"]["code"] == 403
+    assert server.target is None, "a refused call must not reach the world"
+
+
+def test_an_encoded_administrative_path_is_refused_as_administrative():
+    """The admin guard read the raw path while everything after it read the decoded
+    one. Two forms of the same path in one function is how a guard gets walked past."""
+    out = json.loads(_Offline()._call("POST", "/api/%73ql-runner", None,
+                                      json.dumps({"query": "UPDATE incident SET state='closed'"})))
+    assert out["error"]["code"] == 403
+    assert "administrative operation" in out["error"]["message"]
+
+
 def test_a_call_to_no_known_server_is_refused_by_name():
     out = json.loads(_Offline(("gym-itsm-mcp", "gym-calendar"))._call("GET", "/nowhere/x", None, None))
     assert out["error"]["code"] == 404
diff --git a/monarch-benchmark/workflowbench/wb_orchestrator/budget.py b/monarch-benchmark/workflowbench/wb_orchestrator/budget.py
index 79169067..4f960a1d 100644
--- a/monarch-benchmark/workflowbench/wb_orchestrator/budget.py
+++ b/monarch-benchmark/workflowbench/wb_orchestrator/budget.py
@@ -63,8 +63,23 @@ def default_ledger_path(repo: Path | None = None) -> Path:
     return root / 'research' / 'budget.sqlite3'
 
 
+# A round's admission envelope is a run reservation whose scope carries this marker
+# (`<run_id>#admission-000`). An attempt's own cap is also a run reservation, so the
+# refusal message is identical for both and the scope is the only thing that tells
+# them apart -- which decides whether a refusal stops the round or just the attempt.
+ROUND_ENVELOPE_MARKER = "#admission-"
+
+
 class BudgetExceeded(RuntimeError):
-    """Admission denied; no new reservation was written."""
+    """Admission denied; no new reservation was written.
+
+    `scope_id` is the scope that ran out, when the refusal came from one. Callers
+    classify on it rather than on the message: see `external_runtime._budget_failure`.
+    """
+
+    def __init__(self, *args, scope_id: str | None = None):
+        super().__init__(*args)
+        self.scope_id = scope_id
 
 
 class ReservationConflict(ValueError):
@@ -446,7 +461,7 @@ def reserve_run(self, scope_id: str, maximum_usd: str | Decimal | int, *,
                 raise ReservationConflict('scope reservations already belong to another run')
             used = sum(r['maximum_microusd'] if r['actual_microusd'] is None else r['actual_microusd'] for r in children)
             if used > maximum:
-                raise BudgetExceeded('run budget exhausted')
+                raise BudgetExceeded('run budget exhausted', scope_id=scope_id)
             if any(datetime.fromisoformat(timestamp) < datetime.fromisoformat(r['created_at']) for r in children):
                 raise ValueError('run reservation cannot predate existing requests')
             status = self._status(connection, week)
@@ -529,7 +544,7 @@ def reserve(self, reservation_id: str, maximum_usd: str | Decimal | int, *,
                 if datetime.fromisoformat(timestamp) < datetime.fromisoformat(envelope['created_at']):
                     raise ValueError('request cannot be before run reservation')
                 if self._run_used(connection, envelope['scope_id']) + maximum > envelope['maximum_microusd']:
-                    raise BudgetExceeded('run budget exhausted')
+                    raise BudgetExceeded('run budget exhausted', scope_id=envelope['scope_id'])
             status = self._status(connection, week)
             if status.overrun_ids:
                 raise BudgetExceeded('recorded reservation overrun blocks further launches')
@@ -542,7 +557,7 @@ def reserve(self, reservation_id: str, maximum_usd: str | Decimal | int, *,
             costs = connection.execute('SELECT maximum_microusd, actual_microusd FROM budget_reservations WHERE scope_id=?', (scope_id,)).fetchall()
             used = sum(row['maximum_microusd'] if row['actual_microusd'] is None else row['actual_microusd'] for row in costs)
             if scope_limit is not None and used + maximum > scope_limit:
-                raise BudgetExceeded('scope budget exhausted')
+                raise BudgetExceeded('scope budget exhausted', scope_id=scope_id)
             result = Reservation(reservation_id, scope_id, maximum, None, week, timestamp, None, metadata_json)
             connection.execute('INSERT INTO budget_reservations VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)', tuple(result.__dict__.values()))
             if envelope is not None:
diff --git a/monarch-benchmark/workflowbench/wb_orchestrator/external_catalogue.py b/monarch-benchmark/workflowbench/wb_orchestrator/external_catalogue.py
index eacd182a..35db5f54 100644
--- a/monarch-benchmark/workflowbench/wb_orchestrator/external_catalogue.py
+++ b/monarch-benchmark/workflowbench/wb_orchestrator/external_catalogue.py
@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 import copy
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 import hashlib
 import json
 from pathlib import Path
@@ -19,6 +19,14 @@
 PARAM_TYPES = {"string", "integer", "number", "boolean", "array", "object"}
 
 
+class Unrepresentable(ValueError):
+    """One source operation Monarch's contract cannot express.
+
+    Its own type so `generate` can exclude it and report it without swallowing any
+    other ValueError -- a malformed document should still refuse the whole catalogue.
+    """
+
+
 @dataclass
 class CatalogueSummary:
     operations_in_spec: int
@@ -26,6 +34,10 @@ class CatalogueSummary:
     folders: list[str]
     service_slugs: dict[str, str]
     sha256: str
+    # Operations left out because Monarch's contract cannot express them, each as
+    # {service, method, path, reason}. Reported rather than fatal, and recorded in the
+    # pack: what Monarch was not taught belongs in the evidence beside what it was.
+    excluded: list[dict] = field(default_factory=list)
 
 
 def _slug(value):
@@ -184,7 +196,7 @@ def _action(product, service, server, path, method, op, public_url):
     if root_body:
         body_fields = {}
     if method in BODYLESS and (body_fields or root_body):
-        raise ValueError(f"{method.upper()} {path}: source body cannot be represented by Monarch's bodyless contract")
+        raise Unrepresentable(f"{method.upper()} {path}: source body cannot be represented by Monarch's bodyless contract")
     specs = list(op.get("parameters") or [])
     if any(spec.get("in") not in ("path", "query") for spec in specs):
         raise ValueError(f"{method.upper()} {path}: unsupported source parameter location")
@@ -277,7 +289,7 @@ def generate(product, task_dir, out_dir, public_url):
     missing = world_type.prerequisites()
     if missing:
         raise ValueError("; ".join(missing))
-    actions, seen_docs = {}, {}
+    actions, seen_docs, excluded = {}, {}, {}
     selected = set(product.services)
     found = set()
     for task in tasks:
@@ -301,8 +313,14 @@ def generate(product, task_dir, out_dir, public_url):
                         op["parameters"] = list(merged.values())
                         if op.get("servers"):
                             raise ValueError("operation-specific server overrides are not supported by this front door")
+                        try:
+                            action = _action(product,service,server,path,method,op,public_url)
+                        except Unrepresentable as exc:
+                            # Excluded, not fatal, and never published: an operation with
+                            # no action must not reach the document Monarch reads either.
+                            excluded[(service, method, path)] = str(exc)
+                            continue
                         public_doc["paths"].setdefault(path,{})[method] = op
-                        action = _action(product,service,server,path,method,op,public_url)
                         key = (service, method, path)
                         if key in actions and actions[key] != action:
                             raise ValueError(f"published operation changed between frozen tasks: {service} {method} {path}")
@@ -339,6 +357,8 @@ def generate(product, task_dir, out_dir, public_url):
         if relative in files:
             raise ValueError("source operations collided in a seed filename")
         files[relative] = _dump(action)
+    left_out = [{"service":service, "method":method, "path":path, "reason":reason}
+                for (service, method, path), reason in sorted(excluded.items())]
     digest = hashlib.sha256()
     for name, text in sorted(files.items()):
         digest.update(name.encode("utf-8")); digest.update(text.encode("utf-8"))
@@ -346,6 +366,7 @@ def generate(product, task_dir, out_dir, public_url):
     manifest = {"format":"workflowbench-external-catalogue@1", "product":product.name, "source":pin,
         "task_contracts":[t["contract_sha256"] for t in tasks], "front_door":public_url,
         "service_slugs":service_slugs,"products":len(service_slugs),"actions":len(actions),"sha256":sha,
+        **({"excluded_operations":left_out} if left_out else {}),
         "parameter_translation":"Published names, source required flags and declared defaults/examples only; all front-door bodies use JSON.",
         "response_translation":"Lowest declared numeric 2xx status; complete published response alternatives retained in source-contracts.yaml."}
     files["ok.txt"] = _dump(manifest)
@@ -362,4 +383,4 @@ def generate(product, task_dir, out_dir, public_url):
         target = out / name
         target.parent.mkdir(parents=True,exist_ok=True)
         if not target.exists(): target.write_bytes(text.encode("utf-8"))
-    return CatalogueSummary(len(actions),len(actions),sorted(service_slugs.values()),service_slugs,sha)
+    return CatalogueSummary(len(actions),len(actions),sorted(service_slugs.values()),service_slugs,sha,left_out)
diff --git a/monarch-benchmark/workflowbench/wb_orchestrator/external_runtime.py b/monarch-benchmark/workflowbench/wb_orchestrator/external_runtime.py
index 5e4099f4..726cfa63 100644
--- a/monarch-benchmark/workflowbench/wb_orchestrator/external_runtime.py
+++ b/monarch-benchmark/workflowbench/wb_orchestrator/external_runtime.py
@@ -12,7 +12,7 @@
 from wb_arms.api_loop import ArmResult, InfraError, _OpenAIResponsesAdapter
 from wb_arms import providers, reservations
 from wb_studio.runtime import Runtime
-from wb_orchestrator.budget import BudgetExceeded
+from wb_orchestrator.budget import ROUND_ENVELOPE_MARKER, BudgetExceeded
 
 
 class AttemptBudget:
@@ -78,9 +78,31 @@ def customer_input(messages):
 
 
 def _budget_failure(exc):
-    # The ledger distinguishes per-attempt/round admission from shared-week stops.
-    weekly = "weekly" in str(exc) or "overrun" in str(exc)
-    return InfraError("infra:weekly_budget" if weekly else "infra:budget", str(exc), retryable=False)
+    """Which refusal the ledger gave, as a termination the orchestrator can act on.
+
+    Three outcomes, not two, and the difference decides whether the round continues.
+    `infra:budget` is the attempt's own scope cap: that attempt is over, the round is
+    not. `infra:run_budget` (the round's admission envelope) and `infra:weekly_budget`
+    both mean nothing further can be paid for at all.
+
+    Collapsing the envelope case into `infra:budget` is what let a round run on to its
+    last attempt recording refusals, then finish and report as though it had measured
+    them. See `orchestrator.STOPS_THE_ROUND`.
+
+    The message cannot tell the first two apart: an attempt's cap is itself a run
+    reservation, so both say `run budget exhausted`. The scope that ran out is the
+    discriminator -- a round's envelope carries `ROUND_ENVELOPE_MARKER`, an attempt's
+    does not.
+    """
+    message = str(exc)
+    scope = getattr(exc, "scope_id", None) or ""
+    if "weekly" in message or "overrun" in message:
+        kind = "infra:weekly_budget"
+    elif ROUND_ENVELOPE_MARKER in scope:
+        kind = "infra:run_budget"
+    else:
+        kind = "infra:budget"
+    return InfraError(kind, message, retryable=False)
 
 
 class Customer:
@@ -215,8 +237,17 @@ def run_attempt(arm, ep, run_config, ledger, run_id, deadline):
         result = scoped_arm.run(ep, deadline=deadline)
         # Harness tool wrappers may have converted the callback's exception into
         # a normal tool response. That must never become a scored agent failure.
+        #
+        # One exception, and only one: the arm's own `timeout` verdict and a customer
+        # call cut at the same deadline are one event, not two. `api_loop.py:670-673`
+        # makes an expired episode deadline a scored verdict on purpose, and the
+        # customer's `infra:timeout` is that same clock. Raising the customer's copy
+        # moved a real agent timeout out of the pass denominator, inflating the rate
+        # by exactly the attempts that ran out of time.
         if customer is not None and customer.failure is not None:
-            raise customer.failure
+            if not (getattr(result, "termination", None) == "timeout"
+                    and getattr(customer.failure, "kind", None) == "infra:timeout"):
+                raise customer.failure
     except Exception as exc:
         failure = customer.failure if customer is not None and customer.failure is not None else exc
         if isinstance(failure, BudgetExceeded):
diff --git a/monarch-benchmark/workflowbench/wb_orchestrator/monarch_setup.py b/monarch-benchmark/workflowbench/wb_orchestrator/monarch_setup.py
index 9ce833fd..3d13a10c 100644
--- a/monarch-benchmark/workflowbench/wb_orchestrator/monarch_setup.py
+++ b/monarch-benchmark/workflowbench/wb_orchestrator/monarch_setup.py
@@ -67,6 +67,9 @@ def sub(m):
 
     return _VAR.sub(sub, value).rstrip("/")
 
+FRONT_DOOR_SECRET_ENV = "STUDIO_FRONT_DOOR_SECRET"
+
+
 def front_door_secret(env) -> str:
     """The segment the Studio's front door demands, or "" when it is not configured.
 
@@ -75,7 +78,22 @@ def front_door_secret(env) -> str:
     sends nothing it did not send before. Rotate it per round by changing the variable
     before `wb monarch setup` writes the seeds.
     """
-    return (env.get("STUDIO_FRONT_DOOR_SECRET") or "").strip().strip("/")
+    return (env.get(FRONT_DOOR_SECRET_ENV) or "").strip().strip("/")
+
+
+def recorded_front_door_url(url: str, env) -> str:
+    """The address as a tracked file may record it: the secret segment by name.
+
+    `config/products/<product>.monarch-kb.yaml` is committed and this repository is
+    public, so a secret written into it once is in history rather than in a file.
+    Nothing reads this field as an address -- `RunConfig.hash` is its only consumer --
+    so the name carries everything the record needs, and rotating the secret stops
+    moving the config hash as a consequence.
+    """
+    secret = front_door_secret(env)
+    if secret and url.endswith("/" + secret):
+        return url[: -len(secret)] + "${%s}" % FRONT_DOOR_SECRET_ENV
+    return url
 
 
 def front_door_path(base: str, env) -> str:
@@ -232,6 +250,11 @@ def say(mark: str, step: str, detail: str = "") -> None:
             say("ok", "generate", f"operations_in_spec={summary.operations_in_spec} "
                 f"files_written={summary.files_written} folders={len(summary.folders)} "
                 f"source_catalogue_sha256={summary.sha256}")
+            for item in summary.excluded:
+                # Named one per line rather than counted: this is an operation the
+                # competitor is not taught, so it belongs in the round's evidence.
+                say("warn", "generate", f"excluded {item['service']} {item['method'].upper()} "
+                                        f"{item['path']}: {item['reason']}")
         else:
             _generate(out, shim_public_url, stdout)
             taught = _enrich(out, product_path, knowledge, knowledge_map, stdout) if knowledge else None
@@ -281,7 +304,7 @@ def say(mark: str, step: str, detail: str = "") -> None:
         path, changed = _write_kb(Path(product_path), product.name, shim_public_url, kb,
                                   taught={"knowledge_source": taught.knowledge_source,
                                           "knowledge_sha256": taught.knowledge_sha256}
-                                  if taught else {})
+                                  if taught else {}, env=env)
         say("ok", "write", f"{path} ({'changed' if changed else 'unchanged'})")
         return 0
     except Stop as stop:
@@ -401,9 +424,16 @@ def _override_snippet(out: Path) -> str:
 
 
 def _write_kb(product_path: Path, name: str, shim_public_url: str,
-              kb: dict[str, str], taught: dict[str, str] | None = None) -> tuple[Path, bool]:
+              kb: dict[str, str], taught: dict[str, str] | None = None, *,
+              env: dict) -> tuple[Path, bool]:
     """The knowledge-base hash file; `taught` adds the knowledge catalog's name and sha256
-    when the seeds were the lab set, so the file says which knowledge the instance holds."""
+    when the seeds were the lab set, so the file says which knowledge the instance holds.
+
+    `env` is required rather than defaulted: this writes a tracked file in a public
+    repository, and the redaction below is the only thing standing between the front
+    door's secret and git history. A caller that forgot it should not silently leak.
+    """
+    shim_public_url = recorded_front_door_url(shim_public_url, env)
     path = product_path.with_name(f"{name}.monarch-kb.yaml")
     doc = {"product": name, "generated_at": "", "seeds_format": SEEDS_FORMAT,
            "shim_public_url": shim_public_url, "kb": dict(sorted(kb.items())), **(taught or {})}
diff --git a/monarch-benchmark/workflowbench/wb_orchestrator/orchestrator.py b/monarch-benchmark/workflowbench/wb_orchestrator/orchestrator.py
index ebbd721b..6ceb4aa0 100644
--- a/monarch-benchmark/workflowbench/wb_orchestrator/orchestrator.py
+++ b/monarch-benchmark/workflowbench/wb_orchestrator/orchestrator.py
@@ -34,11 +34,22 @@
 from wb_results.store import Store
 from wb_results import evidence
 from wb_orchestrator import config as config_mod
+from wb_orchestrator.budget import ROUND_ENVELOPE_MARKER
 from wb_orchestrator.config import ConfigError
 from wb_world.episode import (  # noqa: F401  (Episode, contract_hash, load_suite re-exported)
     LEGACY_SUITE, Episode, contract_hash, load_suite, suite_id)
 
 MAX_INFRA_RETRIES = 2
+# Terminations that mean nothing further in this round can be paid for, mapped to the
+# stop reason each records. A round that hits one has to stop and be resumable: running
+# on to the last attempt collecting refusals and then finishing leaves a round that
+# reads as complete but measured only part of its work.
+#
+# `infra:budget` -- an attempt exhausting its own scope cap -- is deliberately absent.
+# That is one attempt's outcome and the round carries on, which is why the three cases
+# need three kinds (`external_runtime._budget_failure`). Stopping on it would kill a
+# whole round the first time any single attempt reached its per-attempt cap.
+STOPS_THE_ROUND = {"infra:weekly_budget": "weekly_budget", "infra:run_budget": "run_budget"}
 # The label of every set that records no world. A round's real suite id comes
 # from its tasks (wb_world.episode.suite_id): the world's version is in it.
 SUITE = LEGACY_SUITE
@@ -303,10 +314,13 @@ def run(self, run_id: str | None = None) -> str:
         run_id = run_id or f"run-{datetime.now(timezone.utc):%Y%m%d-%H%M%S}"
         arms = self._arms()
         self._admit(run_id, arms, skip=set())   # a refused round leaves no run row
-        self.store.create_run(run_id, self._hash(), self.suite, self._config())
-        if self.run_config and self.run_config.config_source:
-            evidence.write_json(self._run_dir(run_id) / "config-source.json", self.run_config.config_source)
-        self._execute(run_id, skip=set(), arms=arms)
+        try:
+            self.store.create_run(run_id, self._hash(), self.suite, self._config())
+            if self.run_config and self.run_config.config_source:
+                evidence.write_json(self._run_dir(run_id) / "config-source.json", self.run_config.config_source)
+            self._execute(run_id, skip=set(), arms=arms)
+        finally:
+            self._close_envelope()
         return run_id
 
     def cancel(self) -> None:
@@ -344,8 +358,11 @@ def resume(self, run_id: str) -> str:
         skip = self.store.completed_identities(run_id)
         arms = self._arms()
         self._admit(run_id, arms, skip)     # only what is left to run is counted
-        self.store.set_stop_reason(run_id, None)  # the run is going again
-        self._execute(run_id, skip=skip, arms=arms)
+        try:
+            self.store.set_stop_reason(run_id, None)  # the run is going again
+            self._execute(run_id, skip=skip, arms=arms)
+        finally:
+            self._close_envelope()
         return run_id
 
     def _arms(self) -> list:
@@ -477,18 +494,31 @@ def _admit(self, run_id: str, arms: list, skip: set[tuple[str, str, int]]) -> No
                 "(Monday 00:00 America/Sao_Paulo); nothing was reserved")
 
         if external:
-            prior_envelopes = [r for r in self.ledger.run_reservations() if r.scope_id.startswith(run_id + "#admission-")]
-            self._budget_run_id = run_id + f"#admission-{len(prior_envelopes):03d}"
+            marker = run_id + ROUND_ENVELOPE_MARKER
+            prior_envelopes = [r for r in self.ledger.run_reservations() if r.scope_id.startswith(marker)]
+            self._budget_run_id = marker + f"{len(prior_envelopes):03d}"
             self.ledger.reserve_run(self._budget_run_id, liability, metadata={"product":self.run_config.product.name,
                 "configuration":self.run_config.hash, "includes_infrastructure_retries":MAX_INFRA_RETRIES,
                 "participants":[p.role + ":" + p.model for p in self.run_config.product.participants]})
 
+    def _close_envelope(self) -> None:
+        """Release the round's unallocated capacity. Idempotent; child holds survive.
+
+        Every path out of a round that opened an envelope has to reach this. It used to
+        hang off `_execute` alone, which left everything between `_admit` and `_execute`
+        -- `create_run`, the config-source evidence, `set_stop_reason` -- able to leak
+        the round's whole liability against the week with nothing spent. A duplicate
+        `--run-id` was enough to do it, and no command can release a run envelope
+        afterwards, so the capacity was gone until someone edited the ledger by hand.
+        """
+        if self.ledger and getattr(self, "_budget_run_id", None):
+            self.ledger.finish_run(self._budget_run_id)
+
     def _execute(self, run_id: str, skip: set[tuple[str, str, int]], arms: list | None = None) -> None:
         try:
             return self._execute_inner(run_id, skip, arms)
         finally:
-            if self.ledger and getattr(self, "_budget_run_id", None):
-                self.ledger.finish_run(self._budget_run_id)
+            self._close_envelope()
 
     def _execute_inner(self, run_id: str, skip: set[tuple[str, str, int]], arms: list | None = None) -> None:
         arms = arms if arms is not None else self._arms()
@@ -531,6 +561,14 @@ def _execute_inner(self, run_id: str, skip: set[tuple[str, str, int]], arms: lis
                 f"run {run_id} stopped: spend US$ {self._spent:.2f} exceeds ceiling "
                 f"US$ {self.run_config.plan.cost_ceiling_usd:.2f} after {self._recorded} attempts; "
                 f"raise cost_ceiling_usd in the plan and run: wb resume {run_id}")
+        if self._stop_reason == "run_budget":
+            self.store.set_stop_reason(run_id, "run_budget")
+            raise RunKilled(
+                f"run {run_id} stopped: this round's admission envelope is exhausted after "
+                f"{self._recorded} attempts (US$ {self._spent:.2f} settled; unsettled holds occupy "
+                f"their full maximum); the attempts it cut are recorded as infra:run_budget and run "
+                f"again on resume. Raise cost_ceiling_usd in the plan, or wait for the holds to "
+                f"settle, and run: wb resume {run_id}")
         if self._stop_reason == "weekly_budget":
             self.store.set_stop_reason(run_id, "weekly_budget")
             raise RunKilled(
@@ -877,10 +915,10 @@ def _run_episode(self, run_id: str, arm, task: dict, trial: int) -> bool:
                     and self._stop_reason is None):
                 self._stop_reason = "cost_ceiling"
                 self._abort.set()
-            if termination == "infra:weekly_budget" and self._stop_reason is None:
-                # The ledger refused a request: nothing else can be paid for this
-                # week. Stop scheduling; the cut attempts run again on resume.
-                self._stop_reason = "weekly_budget"
+            if termination in STOPS_THE_ROUND and self._stop_reason is None:
+                # The ledger refused a request and nothing else in this round can be
+                # paid for. Stop scheduling; the cut attempts run again on resume.
+                self._stop_reason = STOPS_THE_ROUND[termination]
                 self._abort.set()
         return self._earns_a_retry(row.passed, termination)
 
diff --git a/monarch-benchmark/workflowbench/wb_worlds/enterprise_ops/adapter.py b/monarch-benchmark/workflowbench/wb_worlds/enterprise_ops/adapter.py
index cc2277e5..af8f9f78 100644
--- a/monarch-benchmark/workflowbench/wb_worlds/enterprise_ops/adapter.py
+++ b/monarch-benchmark/workflowbench/wb_worlds/enterprise_ops/adapter.py
@@ -197,21 +197,28 @@ def _call(self, method: str, url: str, params: str | None, body: str | None) ->
         if name is None:
             return _error(404, f"no server of this product serves {url!r}; "
                                f"the services are {', '.join(sorted(self.servers))}")
+        # Decode once, up front. Every check below reads the decoded path and the
+        # target sent on the wire is re-encoded from it. Mixing the two forms is what
+        # broke: the admin guard read the raw path while the surface check read the
+        # decoded one, and the decoded one was then sent as the request target --
+        # `/locations/name/TechCorp%20NYC%20Headquarters` became a space in the
+        # request line, which urllib refuses outright.
+        path = urllib.parse.unquote(path)
         if any(path.startswith(a) for a in ADMIN_PATHS):
             # Refused here as well as at the front door: a world reached by any other
             # route is still the same world.
             return _error(403, f"{path} is an administrative operation of the product "
                                f"under test, not part of the task")
-        path = urllib.parse.unquote(path)
         if method.upper() == "GET" and path == "/openapi.json":
             return json.dumps(self._public_spec(name))
         public = self._public_spec(name)
-        route = path.split("?", 1)[0]
+        route, separator, query = path.partition("?")
         permitted = any(method.lower() in ops and
                         re.fullmatch(re.sub(r"\{[^}]+\}", "[^/]+", p), route)
                         for p, ops in public.get("paths", {}).items())
         if not permitted:
             return _error(403, f"{method.upper()} {route} is outside this task's published tool mode")
+        path = urllib.parse.quote(route, safe="/") + separator + query
         if params:
             path = f"{path}?{urllib.parse.urlencode(json.loads(params))}"
         try: