From c0538df78874d40949037adb80e5db0fda88d7f9 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Sun, 25 Oct 2026 00:45:04 -0700
Subject: [PATCH 01/12] fix(gateway): check permit allowlist before tool lookup
 so unnamed tools are permit denials

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 gateway/app/routers/mcp.py                | 129 ++++++++++
 gateway/app/routers/mcp_standard.py       |  12 +-
 gateway/tests/test_permit_unknown_tool.py | 278 ++++++++++++++++++++++
 3 files changed, 417 insertions(+), 2 deletions(-)
 create mode 100644 gateway/tests/test_permit_unknown_tool.py

diff --git a/gateway/app/routers/mcp.py b/gateway/app/routers/mcp.py
index a6750f46..6e03036a 100644
--- a/gateway/app/routers/mcp.py
+++ b/gateway/app/routers/mcp.py
@@ -1197,6 +1197,112 @@ async def _assert_action_current_access(
         raise ToolPermissionDenied(policy.reason or "policy_denied")
 
 
+async def _deny_unregistered_tool_outside_permit(
+    *,
+    idem: Any,
+    idem_begin: IdempotencyBegin | None,
+    idem_started: bool,
+    permit_id: str,
+    wallet_id: str | None,
+    auth: AuthContext,
+    tool_name: str,
+    arguments: dict[str, Any],
+    endpoint: str,
+    idempotency_endpoint: str,
+    idempotency_key: str | None,
+    transport: str,
+    request_id: str | None,
+    effective_request_payload: dict[str, Any],
+    effective_request_hash: str,
+) -> None:
+    """Refuse an unregistered tool the explicit permit does not name.
+
+    Returns without effect when the permit is unknown or names the tool, so
+    the caller falls through to its unchanged "Tool not found" handling.
+    Otherwise the permit's own validation decides the reason (normally
+    ``permit_tool_not_allowed``; an earlier failure such as a wallet or key
+    mismatch keeps its narrower code) and a signed denial receipt is issued
+    through the same path as every other governed permit denial. Nothing is
+    reserved, charged, or dispatched.
+    """
+    async with get_session_factory()() as session:
+        model = await session.get(PermitModel, permit_id)
+    if model is None:
+        return
+    try:
+        allowed_tools = json.loads(model.allowed_tools_json or "[]")
+    except (TypeError, ValueError):
+        allowed_tools = []
+    if isinstance(allowed_tools, list) and tool_name in allowed_tools:
+        return
+    validation = await get_permit_service().validate_for_action(
+        permit_id=permit_id,
+        wallet_id=wallet_id or "",
+        tool_name=tool_name,
+        estimated_credits=Decimal("0"),
+        key_id=auth.key_id,
+        arguments=arguments,
+    )
+    if validation.allowed:  # pragma: no cover - allowlist already excludes it
+        return
+    reason = validation.reason or "permit_tool_not_allowed"
+    decision = evaluate_tool_invocation(
+        auth=auth,
+        wallet_id=wallet_id,
+        tool_name=tool_name,
+        estimated_cost=None,
+        request_id=request_id,
+    )
+    audit_event = await _audit_mcp_invocation(
+        effects_committed=False,
+        decision=decision,
+        endpoint=endpoint,
+        transport=transport,
+        ok=False,
+        error=reason,
+        extra_metadata={
+            "permit_id": permit_id,
+            "idempotency_key": idempotency_key,
+            "request_hash": effective_request_hash,
+            "tool_registered": False,
+        },
+    )
+    receipt_payload = None
+    if validation.permit is not None and wallet_id:
+        receipt_payload = await _finalize_governed_denial(
+            unaccepted_action_owner_id=None,
+            idem=idem,
+            effects_committed=False,
+            approval_consumed=False,
+            permit_model=validation.permit,
+            wallet_id=wallet_id,
+            key_id=auth.key_id,
+            endpoint=idempotency_endpoint,
+            idempotency_key=idempotency_key,
+            tool_name=tool_name,
+            request_payload=effective_request_payload,
+            arguments=arguments,
+            registered_cost=Decimal("0"),
+            audit_event_id=audit_event.event_id,
+            reason=reason,
+            reason_code=reason,
+            outcome="denied",
+            status_code=403,
+        )
+    else:
+        await _complete_governed_denial_idempotency(
+            idem=idem,
+            idem_started=idem_started,
+            wallet_id=wallet_id,
+            endpoint=idempotency_endpoint,
+            idempotency_key=idempotency_key,
+            reason=reason,
+        )
+    raise ToolPermissionDenied(
+        reason, receipt=receipt_payload, details=validation.details
+    )
+
+
 async def _execute_registered_tool_inner(
     *,
     tool_name: str,
@@ -1367,6 +1473,29 @@ async def _execute_registered_tool_inner(
                     )
                 await _raise_replayed_error(replay)
                 return replay.response_json
+        if governed_call and permit_id:
+            # A permit is the caller's declared authority. Its allowlist is
+            # checked before the registry so a tool the permit never named is
+            # refused as out of scope (-32003 / 403 with a denial receipt)
+            # whether or not the gateway has it registered; "not found" is
+            # kept for tools the permit does name but the registry lacks.
+            await _deny_unregistered_tool_outside_permit(
+                idem=idem,
+                idem_begin=idem_begin,
+                idem_started=idem_started,
+                permit_id=permit_id,
+                wallet_id=wallet_id,
+                auth=auth,
+                tool_name=tool_name,
+                arguments=arguments,
+                endpoint=endpoint,
+                idempotency_endpoint=idempotency_endpoint,
+                idempotency_key=idempotency_key,
+                transport=transport,
+                request_id=request_id,
+                effective_request_payload=effective_request_payload,
+                effective_request_hash=effective_request_hash,
+            )
         reason = f"Tool not found: {tool_name}"
         await _complete_governed_denial_idempotency(
             idem=idem,
diff --git a/gateway/app/routers/mcp_standard.py b/gateway/app/routers/mcp_standard.py
index 31f8fe1d..7f5b8b40 100644
--- a/gateway/app/routers/mcp_standard.py
+++ b/gateway/app/routers/mcp_standard.py
@@ -465,10 +465,18 @@ async def _governed_tools_call(
     # require a prior tools/list before they resolve here.
     _ensure_local_mcp_tools_registered()
     record = get_service_registry().get_local(tool_name)
-    if record is None:
+    if record is None and not permit_id:
+        # No permit is minted for a tool that does not exist. With an explicit
+        # permit the governed pipeline decides instead: it checks the permit's
+        # allowlist first, so an unnamed tool is a -32003 permit denial with a
+        # receipt, and only a tool the permit names falls back to -32001.
         raise _mcp_error(-32001, f"Tool not found: {tool_name}")
 
-    if get_service_registry().get_action_binding(record) is not None and not permit_id:
+    if (
+        record is not None
+        and get_service_registry().get_action_binding(record) is not None
+        and not permit_id
+    ):
         raise _mcp_error(-32003, "action_permit_required")
 
     # Wallet policy shapes the permit this surface mints. A policy demanding a
diff --git a/gateway/tests/test_permit_unknown_tool.py b/gateway/tests/test_permit_unknown_tool.py
new file mode 100644
index 00000000..dcfc7df2
--- /dev/null
+++ b/gateway/tests/test_permit_unknown_tool.py
@@ -0,0 +1,278 @@
+"""A permit's allowlist is checked before the tool registry.
+
+When a governed call carries an explicit permit and names a tool the permit
+does not allow, the refusal is ``permit_tool_not_allowed`` (-32003 / HTTP 403)
+with a signed denial receipt, whether or not the tool is registered. Calls
+without a permit, and tools the permit does name, keep "Tool not found".
+"""
+
+from __future__ import annotations
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+
+from app.core.config import get_settings
+from app.main import app
+from app.routers import mcp as mcp_router
+from app.services.receipts import get_receipt_service
+from app.services.service_registry import get_service_registry
+from tests.test_mcp_trust_mode import (
+    _assert_denial_audited,
+    _assert_no_tool_debits,
+    _jsonrpc_body,
+    _register_echo_tool,
+    strict_trust_mode,  # noqa: F401 - fixture
+)
+from tests.test_trust_helpers import create_tool_permit, provision_agent_wallet
+
+MCP_HEADERS = {
+    "Accept": "application/json, text/event-stream",
+    "MCP-Protocol-Version": "2025-06-18",
+}
+
+ALLOWED = "permit-scope-allowed-tool"
+UNKNOWN = "permit-scope-unregistered-tool"
+
+
+@pytest.fixture
+async def client():
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as c:
+        yield c
+
+
+@pytest.fixture
+def standard_mcp_enabled(monkeypatch):
+    monkeypatch.setenv("ENABLE_STANDARD_MCP_ENDPOINT", "true")
+    get_settings.cache_clear()
+    yield
+    monkeypatch.setenv("ENABLE_STANDARD_MCP_ENDPOINT", "false")
+    get_settings.cache_clear()
+
+
+async def _permit_for_allowed_tool(client: AsyncClient) -> tuple[dict, dict]:
+    provisioned = await provision_agent_wallet(client)
+    permit = await create_tool_permit(
+        client,
+        wallet_id=provisioned["agent_wallet_id"],
+        key_id=provisioned["key_id"],
+        tool_name=ALLOWED,
+    )
+    return provisioned, permit
+
+
+async def _assert_denial_receipt(receipt: dict, *, permit_id: str, wallet_id: str):
+    assert receipt["outcome"] == "denied"
+    assert receipt["reason_code"] == "permit_tool_not_allowed"
+    assert receipt["permit_id"] == permit_id
+    assert receipt["wallet_id"] == wallet_id
+    assert receipt["tool"] == UNKNOWN
+    assert receipt["ledger_entry_id"] is None
+    assert str(receipt["credits_charged"]) in {"0", "0.0", "0.00"}
+    assert (await get_receipt_service().verify_receipt(receipt["receipt_id"]))[0]
+
+
+@pytest.mark.anyio
+@pytest.mark.usefixtures("strict_trust_mode")
+async def test_jsonrpc_unregistered_tool_outside_permit_is_permit_denial(
+    client, clean_database
+):
+    registry = get_service_registry()
+    _register_echo_tool(ALLOWED)
+    try:
+        provisioned, permit = await _permit_for_allowed_tool(client)
+        resp = await client.post(
+            "/mcp/messages",
+            json=_jsonrpc_body(
+                tool_name=UNKNOWN,
+                wallet_id=provisioned["agent_wallet_id"],
+                request_id="unknown-tool-permit-call",
+                permit_id=permit["permit_id"],
+                idempotency_key="unknown-tool-permit-idem",
+            ),
+            headers=provisioned["agent_headers"],
+        )
+    finally:
+        registry.unregister_local(ALLOWED)
+
+    error = resp.json()["error"]
+    assert error["code"] == -32003
+    assert error["message"] == "permit_tool_not_allowed"
+    assert error["data"]["details"]["requested_tool"] == UNKNOWN
+    assert error["data"]["details"]["allowed_tools"] == [ALLOWED]
+    await _assert_denial_receipt(
+        error["data"]["receipt"],
+        permit_id=permit["permit_id"],
+        wallet_id=provisioned["agent_wallet_id"],
+    )
+    await _assert_denial_audited(
+        wallet_id=provisioned["agent_wallet_id"],
+        tool_name=UNKNOWN,
+        reason="permit_tool_not_allowed",
+    )
+    await _assert_no_tool_debits(
+        client=client,
+        wallet_id=provisioned["agent_wallet_id"],
+        headers=provisioned["agent_headers"],
+        tool_name=UNKNOWN,
+    )
+
+
+@pytest.mark.anyio
+@pytest.mark.usefixtures("strict_trust_mode")
+async def test_jsonrpc_permit_denial_for_unregistered_tool_replays(
+    client, clean_database
+):
+    provisioned, permit = await _permit_for_allowed_tool(client)
+    body = _jsonrpc_body(
+        tool_name=UNKNOWN,
+        wallet_id=provisioned["agent_wallet_id"],
+        request_id="unknown-tool-permit-replay",
+        permit_id=permit["permit_id"],
+        idempotency_key="unknown-tool-permit-replay-idem",
+    )
+    first = await client.post(
+        "/mcp/messages", json=body, headers=provisioned["agent_headers"]
+    )
+    replay = await client.post(
+        "/mcp/messages", json=body, headers=provisioned["agent_headers"]
+    )
+    assert first.json()["error"]["message"] == "permit_tool_not_allowed"
+    assert replay.json()["error"]["code"] == -32003
+    assert replay.json()["error"]["message"] == "permit_tool_not_allowed"
+
+
+@pytest.mark.anyio
+@pytest.mark.usefixtures("strict_trust_mode")
+async def test_rest_unregistered_tool_outside_permit_is_403_with_receipt(
+    client, clean_database
+):
+    provisioned, permit = await _permit_for_allowed_tool(client)
+    resp = await client.post(
+        f"/mcp/tools/{UNKNOWN}/invoke",
+        json={
+            "name": UNKNOWN,
+            "arguments": {"message": "hello"},
+            "mcp_context": {
+                "wallet_id": provisioned["agent_wallet_id"],
+                "permit_id": permit["permit_id"],
+                "idempotency_key": "unknown-tool-permit-rest-idem",
+            },
+        },
+        headers=provisioned["agent_headers"],
+    )
+    assert resp.status_code == 403
+    detail = resp.json()["detail"]
+    assert detail["error"] == "permit_tool_not_allowed"
+    await _assert_denial_receipt(
+        detail["receipt"],
+        permit_id=permit["permit_id"],
+        wallet_id=provisioned["agent_wallet_id"],
+    )
+
+
+@pytest.mark.anyio
+@pytest.mark.usefixtures("strict_trust_mode")
+async def test_unregistered_tool_named_by_permit_is_still_not_found(
+    client, clean_database
+):
+    provisioned = await provision_agent_wallet(client)
+    permit = await create_tool_permit(
+        client,
+        wallet_id=provisioned["agent_wallet_id"],
+        key_id=provisioned["key_id"],
+        tool_name=UNKNOWN,
+    )
+    body = {
+        "name": UNKNOWN,
+        "arguments": {"message": "hello"},
+        "mcp_context": {
+            "wallet_id": provisioned["agent_wallet_id"],
+            "permit_id": permit["permit_id"],
+            "idempotency_key": "named-unknown-rest-idem",
+        },
+    }
+    rest = await client.post(
+        f"/mcp/tools/{UNKNOWN}/invoke", json=body, headers=provisioned["agent_headers"]
+    )
+    assert rest.status_code == 404
+    rpc = await client.post(
+        "/mcp/messages",
+        json=_jsonrpc_body(
+            tool_name=UNKNOWN,
+            wallet_id=provisioned["agent_wallet_id"],
+            request_id="named-unknown-rpc",
+            permit_id=permit["permit_id"],
+            idempotency_key="named-unknown-rpc-idem",
+        ),
+        headers=provisioned["agent_headers"],
+    )
+    assert rpc.json()["error"]["code"] == -32001
+
+
+@pytest.mark.anyio
+@pytest.mark.usefixtures("strict_trust_mode")
+async def test_unregistered_tool_without_permit_is_unchanged(client, clean_database):
+    provisioned = await provision_agent_wallet(client)
+    rpc = await client.post(
+        "/mcp/messages",
+        json=_jsonrpc_body(
+            tool_name=UNKNOWN,
+            wallet_id=provisioned["agent_wallet_id"],
+            request_id="no-permit-unknown",
+            idempotency_key="no-permit-unknown-idem",
+        ),
+        headers=provisioned["agent_headers"],
+    )
+    assert rpc.json()["error"]["code"] == -32001
+    assert rpc.json()["error"]["message"] == f"Tool not found: {UNKNOWN}"
+    assert mcp_router.settings.TRUST_MODE_ENABLED is True
+
+
+@pytest.mark.anyio
+@pytest.mark.usefixtures("strict_trust_mode")
+async def test_standard_mcp_explicit_permit_unregistered_tool_is_permit_denial(
+    client, clean_database, standard_mcp_enabled
+):
+    provisioned, permit = await _permit_for_allowed_tool(client)
+    resp = await client.post(
+        "/mcp",
+        json={
+            "jsonrpc": "2.0",
+            "id": 1,
+            "method": "tools/call",
+            "params": {
+                "name": UNKNOWN,
+                "arguments": {},
+                "_meta": {"io.agentmiddleware/permit_id": permit["permit_id"]},
+            },
+        },
+        headers={**provisioned["agent_headers"], **MCP_HEADERS},
+    )
+    error = resp.json()["error"]
+    assert error["code"] == -32003
+    assert error["message"] == "permit_tool_not_allowed"
+    await _assert_denial_receipt(
+        error["data"]["receipt"],
+        permit_id=permit["permit_id"],
+        wallet_id=provisioned["agent_wallet_id"],
+    )
+
+
+@pytest.mark.anyio
+async def test_standard_mcp_without_permit_unknown_tool_still_not_found(
+    client, clean_database, standard_mcp_enabled
+):
+    """No permit is auto-minted for a tool that does not exist."""
+    provisioned = await provision_agent_wallet(client)
+    resp = await client.post(
+        "/mcp",
+        json={
+            "jsonrpc": "2.0",
+            "id": 1,
+            "method": "tools/call",
+            "params": {"name": UNKNOWN, "arguments": {}},
+        },
+        headers={**provisioned["agent_headers"], **MCP_HEADERS},
+    )
+    assert resp.json()["error"]["code"] == -32001

From 776faaf16ad65482454e4c2a1568b721af883723 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Sun, 25 Oct 2026 19:50:25 -0700
Subject: [PATCH 02/12] fix(mcp): preserve unknown-tool denial replay details

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 gateway/app/routers/mcp.py                | 10 ++++--
 gateway/app/routers/mcp_standard.py       | 16 +++++----
 gateway/tests/test_permit_unknown_tool.py | 42 +++++++++++++++++++++--
 3 files changed, 56 insertions(+), 12 deletions(-)

diff --git a/gateway/app/routers/mcp.py b/gateway/app/routers/mcp.py
index 6e03036a..20ffe77b 100644
--- a/gateway/app/routers/mcp.py
+++ b/gateway/app/routers/mcp.py
@@ -1203,7 +1203,7 @@ async def _deny_unregistered_tool_outside_permit(
     idem_begin: IdempotencyBegin | None,
     idem_started: bool,
     permit_id: str,
-    wallet_id: str | None,
+    wallet_id: str,
     auth: AuthContext,
     tool_name: str,
     arguments: dict[str, Any],
@@ -1288,6 +1288,7 @@ async def _deny_unregistered_tool_outside_permit(
             reason_code=reason,
             outcome="denied",
             status_code=403,
+            response_extra_data={"details": validation.details},
         )
     else:
         await _complete_governed_denial_idempotency(
@@ -4353,6 +4354,7 @@ async def _raise_replayed_error(replay: Any) -> None:
         return
     reason = str(replay.response_json.get("error") or "governed_call_failed")
     receipt = replay.response_json.get("receipt")
+    details = replay.response_json.get("details")
     extra_data = {}
     reconciliation = replay.response_json.get("refund_reconciliation")
     if isinstance(reconciliation, dict):
@@ -4387,7 +4389,11 @@ async def _raise_replayed_error(replay: Any) -> None:
     if isinstance(dispatch, dict):
         extra_data["dispatch"] = dispatch
     if replay.status_code == 403:
-        raise ToolPermissionDenied(reason, receipt=receipt)
+        raise ToolPermissionDenied(
+            reason,
+            receipt=receipt,
+            details=details if isinstance(details, dict) else None,
+        )
     raise GovernedToolError(
         reason,
         receipt=receipt,
diff --git a/gateway/app/routers/mcp_standard.py b/gateway/app/routers/mcp_standard.py
index 7f5b8b40..4c302030 100644
--- a/gateway/app/routers/mcp_standard.py
+++ b/gateway/app/routers/mcp_standard.py
@@ -507,13 +507,15 @@ async def _governed_tools_call(
 
     auto_minted = permit_id is None
     idempotency_key = client_idempotency_key or f"mcp-auto-{uuid.uuid4().hex}"
-    permit_id = permit_id or await _mint_auto_permit(
-        auth=auth,
-        tool_name=tool_name,
-        record=record,
-        client_key=client_idempotency_key,
-        requires_human_approval=requires_human_approval,
-    )
+    if not permit_id:
+        assert record is not None  # Checked before any auto-mint.
+        permit_id = await _mint_auto_permit(
+            auth=auth,
+            tool_name=tool_name,
+            record=record,
+            client_key=client_idempotency_key,
+            requires_human_approval=requires_human_approval,
+        )
 
     call_params = {
         "name": tool_name,
diff --git a/gateway/tests/test_permit_unknown_tool.py b/gateway/tests/test_permit_unknown_tool.py
index dcfc7df2..3631e776 100644
--- a/gateway/tests/test_permit_unknown_tool.py
+++ b/gateway/tests/test_permit_unknown_tool.py
@@ -10,8 +10,11 @@
 
 import pytest
 from httpx import ASGITransport, AsyncClient
+from sqlalchemy import func, select
 
 from app.core.config import get_settings
+from app.db.database import get_session_factory
+from app.db.models import LedgerEntryModel, ReceiptModel
 from app.main import app
 from app.routers import mcp as mcp_router
 from app.services.receipts import get_receipt_service
@@ -137,9 +140,42 @@ async def test_jsonrpc_permit_denial_for_unregistered_tool_replays(
     replay = await client.post(
         "/mcp/messages", json=body, headers=provisioned["agent_headers"]
     )
-    assert first.json()["error"]["message"] == "permit_tool_not_allowed"
-    assert replay.json()["error"]["code"] == -32003
-    assert replay.json()["error"]["message"] == "permit_tool_not_allowed"
+    first_error = first.json()["error"]
+    replay_error = replay.json()["error"]
+    assert first_error["code"] == replay_error["code"] == -32003
+    assert first_error["message"] == "permit_tool_not_allowed"
+    assert replay_error["message"] == first_error["message"]
+    expected_details = {
+        "requested_tool": UNKNOWN,
+        "allowed_tools": [ALLOWED],
+    }
+    assert first_error["data"]["details"] == expected_details
+    assert replay_error["data"]["details"] == expected_details
+    assert first_error["data"]["receipt"] == replay_error["data"]["receipt"]
+    await _assert_denial_receipt(
+        replay_error["data"]["receipt"],
+        permit_id=permit["permit_id"],
+        wallet_id=provisioned["agent_wallet_id"],
+    )
+    async with get_session_factory()() as session:
+        receipts = await session.scalar(
+            select(func.count())
+            .select_from(ReceiptModel)
+            .where(
+                ReceiptModel.wallet_id == provisioned["agent_wallet_id"],
+                ReceiptModel.tool == UNKNOWN,
+            )
+        )
+        debits = await session.scalar(
+            select(func.count())
+            .select_from(LedgerEntryModel)
+            .where(
+                LedgerEntryModel.wallet_id == provisioned["agent_wallet_id"],
+                LedgerEntryModel.action == "debit",
+            )
+        )
+    assert receipts == 1
+    assert debits == 0
 
 
 @pytest.mark.anyio

From f2f51afabbf2fd309669c393c06c593ea872fcf2 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Mon, 26 Oct 2026 07:01:45 -0700
Subject: [PATCH 03/12] fix(mcp): validate permit before reading its allowlist
 for unregistered tools (#232)

- Run validate_for_action first; a foreign or invalid permit denies with
  its own reason and no allowlist details (no cross-wallet tool oracle).
- Action permits naming an unregistered tool get the same permit decision
  instead of action_tool_binding_required.
- Mint a denial receipt only when the call carries an idempotency key.
- Tests use real registered tool ids (partner.notes.write / .count).

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 gateway/app/routers/mcp.py                |  47 ++++--
 gateway/tests/test_permit_unknown_tool.py | 169 +++++++++++++++++++---
 2 files changed, 183 insertions(+), 33 deletions(-)

diff --git a/gateway/app/routers/mcp.py b/gateway/app/routers/mcp.py
index 2aeff045..f787cbd4 100644
--- a/gateway/app/routers/mcp.py
+++ b/gateway/app/routers/mcp.py
@@ -1229,12 +1229,11 @@ async def _deny_unregistered_tool_outside_permit(
         model = await session.get(PermitModel, permit_id)
     if model is None:
         return
-    try:
-        allowed_tools = json.loads(model.allowed_tools_json or "[]")
-    except (TypeError, ValueError):
-        allowed_tools = []
-    if isinstance(allowed_tools, list) and tool_name in allowed_tools:
-        return
+    # Validate the permit (wallet, key, signature, expiry) BEFORE its allowlist
+    # influences the outcome. Reading allowed_tools first let a caller probe a
+    # permit_id belonging to another wallet: "not found" vs. a denial revealed
+    # whether that permit names the tool. A permit that fails validation now
+    # denies with its own narrower reason and no allowlist details.
     validation = await get_permit_service().validate_for_action(
         permit_id=permit_id,
         wallet_id=wallet_id or "",
@@ -1243,9 +1242,11 @@ async def _deny_unregistered_tool_outside_permit(
         key_id=auth.key_id,
         arguments=arguments,
     )
-    if validation.allowed:  # pragma: no cover - allowlist already excludes it
+    if validation.allowed:
+        # The validated permit names this tool; keep "Tool not found".
         return
     reason = validation.reason or "permit_tool_not_allowed"
+    details = validation.details if reason == "permit_tool_not_allowed" else None
     decision = evaluate_tool_invocation(
         auth=auth,
         wallet_id=wallet_id,
@@ -1268,7 +1269,9 @@ async def _deny_unregistered_tool_outside_permit(
         },
     )
     receipt_payload = None
-    if validation.permit is not None and wallet_id:
+    # A signed receipt is minted only for an idempotent call: without a key
+    # there is no record to bind it to and retries would mint one per call.
+    if validation.permit is not None and wallet_id and idempotency_key:
         receipt_payload = await _finalize_governed_denial(
             unaccepted_action_owner_id=None,
             idem=idem,
@@ -1288,7 +1291,7 @@ async def _deny_unregistered_tool_outside_permit(
             reason_code=reason,
             outcome="denied",
             status_code=403,
-            response_extra_data={"details": validation.details},
+            response_extra_data={"details": details},
         )
     else:
         await _complete_governed_denial_idempotency(
@@ -1299,9 +1302,7 @@ async def _deny_unregistered_tool_outside_permit(
             idempotency_key=idempotency_key,
             reason=reason,
         )
-    raise ToolPermissionDenied(
-        reason, receipt=receipt_payload, details=validation.details
-    )
+    raise ToolPermissionDenied(reason, receipt=receipt_payload, details=details)
 
 
 async def _execute_registered_tool_inner(
@@ -1418,6 +1419,28 @@ async def _execute_registered_tool_inner(
     )
     if action_binding is not None or is_action:
         if action_binding is None:
+            if not service and is_action and permit_id:
+                # An action permit naming an unregistered tool gets the same
+                # validated permit-scope decision as a tool permit. No
+                # idempotency record exists yet on this path, so the denial
+                # carries no receipt (see _deny_unregistered_tool_outside_permit).
+                await _deny_unregistered_tool_outside_permit(
+                    idem=idem,
+                    idem_begin=None,
+                    idem_started=False,
+                    permit_id=permit_id,
+                    wallet_id=wallet_id,
+                    auth=auth,
+                    tool_name=tool_name,
+                    arguments=arguments,
+                    endpoint=endpoint,
+                    idempotency_endpoint=idempotency_endpoint,
+                    idempotency_key=None,
+                    transport=transport,
+                    request_id=request_id,
+                    effective_request_payload=effective_request_payload,
+                    effective_request_hash=effective_request_hash,
+                )
             raise ToolPermissionDenied("action_tool_binding_required")
         if not is_action or action_model is None:
             raise ToolPermissionDenied("action_permit_required")
diff --git a/gateway/tests/test_permit_unknown_tool.py b/gateway/tests/test_permit_unknown_tool.py
index 3631e776..1e51361c 100644
--- a/gateway/tests/test_permit_unknown_tool.py
+++ b/gateway/tests/test_permit_unknown_tool.py
@@ -14,16 +14,20 @@
 
 from app.core.config import get_settings
 from app.db.database import get_session_factory
-from app.db.models import LedgerEntryModel, ReceiptModel
+from app.db.models import LedgerEntryModel, PermitModel, ReceiptModel
 from app.main import app
 from app.routers import mcp as mcp_router
+from app.services.dogfood_tool import (
+    DOGFOOD_SECOND_TOOL_ID,
+    DOGFOOD_TOOL_ID,
+    sync_dogfood_tool_registration,
+)
 from app.services.receipts import get_receipt_service
 from app.services.service_registry import get_service_registry
 from tests.test_mcp_trust_mode import (
     _assert_denial_audited,
     _assert_no_tool_debits,
     _jsonrpc_body,
-    _register_echo_tool,
     strict_trust_mode,  # noqa: F401 - fixture
 )
 from tests.test_trust_helpers import create_tool_permit, provision_agent_wallet
@@ -33,8 +37,26 @@
     "MCP-Protocol-Version": "2025-06-18",
 }
 
-ALLOWED = "permit-scope-allowed-tool"
-UNKNOWN = "permit-scope-unregistered-tool"
+# Real product tool ids from app/services/dogfood_tool.py. The fixture below
+# registers partner.notes.write (ENABLE_DOGFOOD_TOOL) and keeps
+# partner.notes.count unregistered (ENABLE_DOGFOOD_SECOND_TOOL off).
+ALLOWED = DOGFOOD_TOOL_ID
+UNKNOWN = DOGFOOD_SECOND_TOOL_ID
+
+
+@pytest.fixture(autouse=True)
+def real_tools(monkeypatch):
+    monkeypatch.setenv("ENABLE_DOGFOOD_TOOL", "true")
+    monkeypatch.setenv("ENABLE_DOGFOOD_SECOND_TOOL", "false")
+    get_settings.cache_clear()
+    sync_dogfood_tool_registration()
+    registry = get_service_registry()
+    assert registry.get_local(ALLOWED) is not None
+    assert registry.get_local(UNKNOWN) is None
+    yield
+    monkeypatch.setenv("ENABLE_DOGFOOD_TOOL", "false")
+    get_settings.cache_clear()
+    sync_dogfood_tool_registration()
 
 
 @pytest.fixture
@@ -80,23 +102,18 @@ async def _assert_denial_receipt(receipt: dict, *, permit_id: str, wallet_id: st
 async def test_jsonrpc_unregistered_tool_outside_permit_is_permit_denial(
     client, clean_database
 ):
-    registry = get_service_registry()
-    _register_echo_tool(ALLOWED)
-    try:
-        provisioned, permit = await _permit_for_allowed_tool(client)
-        resp = await client.post(
-            "/mcp/messages",
-            json=_jsonrpc_body(
-                tool_name=UNKNOWN,
-                wallet_id=provisioned["agent_wallet_id"],
-                request_id="unknown-tool-permit-call",
-                permit_id=permit["permit_id"],
-                idempotency_key="unknown-tool-permit-idem",
-            ),
-            headers=provisioned["agent_headers"],
-        )
-    finally:
-        registry.unregister_local(ALLOWED)
+    provisioned, permit = await _permit_for_allowed_tool(client)
+    resp = await client.post(
+        "/mcp/messages",
+        json=_jsonrpc_body(
+            tool_name=UNKNOWN,
+            wallet_id=provisioned["agent_wallet_id"],
+            request_id="unknown-tool-permit-call",
+            permit_id=permit["permit_id"],
+            idempotency_key="unknown-tool-permit-idem",
+        ),
+        headers=provisioned["agent_headers"],
+    )
 
     error = resp.json()["error"]
     assert error["code"] == -32003
@@ -312,3 +329,113 @@ async def test_standard_mcp_without_permit_unknown_tool_still_not_found(
         headers={**provisioned["agent_headers"], **MCP_HEADERS},
     )
     assert resp.json()["error"]["code"] == -32001
+
+
+async def _receipt_count(wallet_id: str) -> int:
+    async with get_session_factory()() as session:
+        return await session.scalar(
+            select(func.count())
+            .select_from(ReceiptModel)
+            .where(ReceiptModel.wallet_id == wallet_id)
+        )
+
+
+@pytest.mark.anyio
+@pytest.mark.usefixtures("strict_trust_mode")
+async def test_foreign_wallet_permit_does_not_leak_allowlist(client, clean_database):
+    """Another wallet's permit is validated before its allowlist is read.
+
+    The caller gets the same wallet-mismatch denial whether or not the foreign
+    permit names the tool, and never sees that permit's allowed_tools.
+    """
+    victim = await provision_agent_wallet(client)
+    names_unknown = await create_tool_permit(
+        client,
+        wallet_id=victim["agent_wallet_id"],
+        key_id=victim["key_id"],
+        tool_name=UNKNOWN,
+        idem_key="victim-permit-names-unknown",
+    )
+    names_other = await create_tool_permit(
+        client,
+        wallet_id=victim["agent_wallet_id"],
+        key_id=victim["key_id"],
+        tool_name=ALLOWED,
+        idem_key="victim-permit-names-other",
+    )
+    attacker = await provision_agent_wallet(client)
+    errors = []
+    for i, permit in enumerate((names_unknown, names_other)):
+        resp = await client.post(
+            "/mcp/messages",
+            json=_jsonrpc_body(
+                tool_name=UNKNOWN,
+                wallet_id=attacker["agent_wallet_id"],
+                request_id=f"foreign-permit-{i}",
+                permit_id=permit["permit_id"],
+                idempotency_key=f"foreign-permit-idem-{i}",
+            ),
+            headers=attacker["agent_headers"],
+        )
+        errors.append(resp.json()["error"])
+    for error in errors:
+        assert error["code"] == -32003
+        assert error["message"] == "permit_wallet_mismatch"
+        assert "allowed_tools" not in str(error.get("data", {}).get("details"))
+        assert ALLOWED not in str(error)
+    assert errors[0]["message"] == errors[1]["message"]
+
+
+@pytest.mark.anyio
+@pytest.mark.usefixtures("strict_trust_mode")
+async def test_permit_denial_without_idempotency_key_mints_no_receipt(
+    client, clean_database
+):
+    provisioned, permit = await _permit_for_allowed_tool(client)
+    before = await _receipt_count(provisioned["agent_wallet_id"])
+    resp = await client.post(
+        "/mcp/messages",
+        json=_jsonrpc_body(
+            tool_name=UNKNOWN,
+            wallet_id=provisioned["agent_wallet_id"],
+            request_id="no-idem-unknown",
+            permit_id=permit["permit_id"],
+        ),
+        headers=provisioned["agent_headers"],
+    )
+    error = resp.json()["error"]
+    if error["code"] == -32003:
+        assert error["message"] == "permit_tool_not_allowed"
+        assert not error.get("data", {}).get("receipt")
+    assert await _receipt_count(provisioned["agent_wallet_id"]) == before
+
+
+@pytest.mark.anyio
+@pytest.mark.usefixtures("strict_trust_mode")
+async def test_action_permit_unregistered_tool_gets_permit_decision(
+    client, clean_database
+):
+    """Action permits are no longer excluded from the permit-scope check."""
+    provisioned, permit = await _permit_for_allowed_tool(client)
+    async with get_session_factory()() as session:
+        model = await session.get(PermitModel, permit["permit_id"])
+        model.action_contract_version = 1
+        await session.commit()
+    before = await _receipt_count(provisioned["agent_wallet_id"])
+    resp = await client.post(
+        "/mcp/messages",
+        json=_jsonrpc_body(
+            tool_name=UNKNOWN,
+            wallet_id=provisioned["agent_wallet_id"],
+            request_id="action-permit-unknown",
+            permit_id=permit["permit_id"],
+            idempotency_key="action-permit-unknown-idem",
+        ),
+        headers=provisioned["agent_headers"],
+    )
+    error = resp.json()["error"]
+    assert error["code"] == -32003
+    assert error["message"] != "action_tool_binding_required"
+    assert error["message"].startswith("permit_")
+    # No idempotency record exists on the action path, so no receipt.
+    assert await _receipt_count(provisioned["agent_wallet_id"]) == before

From c83b15ceebac7b73e2ec095b64543568f2760105 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Mon, 26 Oct 2026 07:22:19 -0700
Subject: [PATCH 04/12] fix(mcp): hide tool registration on permit denials

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 gateway/app/routers/mcp.py                | 333 ++++++++++------------
 gateway/tests/test_permit_unknown_tool.py | 146 +++++++++-
 2 files changed, 297 insertions(+), 182 deletions(-)

diff --git a/gateway/app/routers/mcp.py b/gateway/app/routers/mcp.py
index f787cbd4..7e6a223b 100644
--- a/gateway/app/routers/mcp.py
+++ b/gateway/app/routers/mcp.py
@@ -1197,10 +1197,9 @@ async def _assert_action_current_access(
         raise ToolPermissionDenied(policy.reason or "policy_denied")
 
 
-async def _deny_unregistered_tool_outside_permit(
+async def _deny_tool_outside_permit(
     *,
     idem: Any,
-    idem_begin: IdempotencyBegin | None,
     idem_started: bool,
     permit_id: str,
     wallet_id: str,
@@ -1215,25 +1214,16 @@ async def _deny_unregistered_tool_outside_permit(
     effective_request_payload: dict[str, Any],
     effective_request_hash: str,
 ) -> None:
-    """Refuse an unregistered tool the explicit permit does not name.
-
-    Returns without effect when the permit is unknown or names the tool, so
-    the caller falls through to its unchanged "Tool not found" handling.
-    Otherwise the permit's own validation decides the reason (normally
-    ``permit_tool_not_allowed``; an earlier failure such as a wallet or key
-    mismatch keeps its narrower code) and a signed denial receipt is issued
-    through the same path as every other governed permit denial. Nothing is
-    reserved, charged, or dispatched.
+    """Refuse a permit denial before registration can influence its evidence.
+
+    Returns without effect when the permit allows the tool. No credits are
+    authorized, reserved, charged, or dispatched for a refusal.
     """
-    async with get_session_factory()() as session:
-        model = await session.get(PermitModel, permit_id)
-    if model is None:
-        return
     # Validate the permit (wallet, key, signature, expiry) BEFORE its allowlist
     # influences the outcome. Reading allowed_tools first let a caller probe a
     # permit_id belonging to another wallet: "not found" vs. a denial revealed
-    # whether that permit names the tool. A permit that fails validation now
-    # denies with its own narrower reason and no allowlist details.
+    # whether that permit names the tool. A permit that fails validation keeps
+    # its narrower reason and details without exposing the allowlist.
     validation = await get_permit_service().validate_for_action(
         permit_id=permit_id,
         wallet_id=wallet_id or "",
@@ -1243,10 +1233,10 @@ async def _deny_unregistered_tool_outside_permit(
         arguments=arguments,
     )
     if validation.allowed:
-        # The validated permit names this tool; keep "Tool not found".
+        # Continue to registration and price checks.
         return
     reason = validation.reason or "permit_tool_not_allowed"
-    details = validation.details if reason == "permit_tool_not_allowed" else None
+    details = validation.details
     decision = evaluate_tool_invocation(
         auth=auth,
         wallet_id=wallet_id,
@@ -1265,7 +1255,6 @@ async def _deny_unregistered_tool_outside_permit(
             "permit_id": permit_id,
             "idempotency_key": idempotency_key,
             "request_hash": effective_request_hash,
-            "tool_registered": False,
         },
     )
     receipt_payload = None
@@ -1400,6 +1389,9 @@ async def _execute_registered_tool_inner(
     service = registry.get_local(tool_name)
     if not service:
         service = await registry.get_persistent(tool_name)
+    if service and service.get("require_permit"):
+        governed_call = True
+        idempotency_endpoint = GOVERNED_MCP_IDEMPOTENCY_ENDPOINT
     action_identity = None
     action_binding = registry.get_action_binding(service) if service else None
     action_model = None
@@ -1423,10 +1415,9 @@ async def _execute_registered_tool_inner(
                 # An action permit naming an unregistered tool gets the same
                 # validated permit-scope decision as a tool permit. No
                 # idempotency record exists yet on this path, so the denial
-                # carries no receipt (see _deny_unregistered_tool_outside_permit).
-                await _deny_unregistered_tool_outside_permit(
+                # carries no receipt (see _deny_tool_outside_permit).
+                await _deny_tool_outside_permit(
                     idem=idem,
-                    idem_begin=None,
                     idem_started=False,
                     permit_id=permit_id,
                     wallet_id=wallet_id,
@@ -1462,65 +1453,130 @@ async def _execute_registered_tool_inner(
         effective_request_payload = action_identity.request_payload
         effective_request_hash = sha256_hex(effective_request_payload)
 
-    if not service:
-        # Durable replay is authoritative even if an operator has since
-        # removed the executable registration. This is essential for terminal
-        # evidence recovery and avoids turning a completed invocation into a
-        # fresh "not found" response after restart or reconfiguration.
-        if governed_call and idempotency_key:
-            try:
-                idem_begin = await _begin_governed_mcp_idempotency(
-                    idem=idem,
-                    wallet_id=wallet_id,
-                    idempotency_key=idempotency_key,
-                    tool_name=tool_name,
-                    endpoint=endpoint,
-                    logical_request_payload=effective_request_payload,
-                    legacy_request_payload=request_payload,
-                    operation_kind="unresolved",
-                )
-                replay = idem_begin.replay
-                idem_started = True
-                if owned_record is not None:
-                    owned_record["record_id"] = idem_begin.record_id
-            except IdempotencyInProgressError:
-                raise
-            except IdempotencyConflictError as exc:
-                raise ValueError(str(exc)) from exc
-            if replay and replay.response_json:
-                if permit_id:
-                    await _assert_governed_replay_access(
-                        permit_id=permit_id,
-                        wallet_id=wallet_id,
-                        tool_name=tool_name,
-                        key_id=auth.key_id,
-                        replay=replay,
-                    )
-                await _raise_replayed_error(replay)
-                return replay.response_json
-        if governed_call and permit_id:
-            # A permit is the caller's declared authority. Its allowlist is
-            # checked before the registry so a tool the permit never named is
-            # refused as out of scope (-32003 / 403 with a denial receipt)
-            # whether or not the gateway has it registered; "not found" is
-            # kept for tools the permit does name but the registry lacks.
-            await _deny_unregistered_tool_outside_permit(
+    if action_identity is None:
+        if governed_call and permit_id and not idempotency_key:
+            await _audit_mcp_invocation(
+                effects_committed=False,
+                decision=tenant_decision,
+                endpoint=endpoint,
+                transport=transport,
+                ok=False,
+                error="idempotency_key_required",
+                extra_metadata={
+                    "permit_id": permit_id,
+                    "request_hash": effective_request_hash,
+                },
+            )
+            raise ValueError("idempotency_key_required")
+
+    # Quote validation and idempotency replay do not depend on registration.
+    # A refused request must cross the same gates for either registry state.
+    quoted = None
+    if quote_id and action_identity is None:
+        quoted = await get_quote_service().validate_for_action(
+            quote_id=quote_id,
+            wallet_id=wallet_id,
+            tool_name=tool_name,
+        )
+        if not quoted.allowed or quoted.quote is None:
+            reason = quoted.reason or "quote_invalid"
+            await _audit_mcp_invocation(
+                effects_committed=False,
+                decision=tenant_decision,
+                endpoint=endpoint,
+                transport=transport,
+                ok=False,
+                error=reason,
+                extra_metadata={
+                    "quote_id": quote_id,
+                    "permit_id": permit_id,
+                    "idempotency_key": idempotency_key,
+                    "request_hash": effective_request_hash,
+                },
+            )
+            raise PermissionError(reason)
+
+    if (
+        governed_call
+        and action_identity is None
+        and idempotency_key
+        and (permit_id or not service)
+    ):
+        execution_kind = (
+            str(service.get("execution_backend") or "metadata_only")
+            if service
+            else "unresolved"
+        )
+        try:
+            idem_begin = await _begin_governed_mcp_idempotency(
                 idem=idem,
-                idem_begin=idem_begin,
-                idem_started=idem_started,
-                permit_id=permit_id,
                 wallet_id=wallet_id,
-                auth=auth,
+                idempotency_key=idempotency_key,
                 tool_name=tool_name,
-                arguments=arguments,
                 endpoint=endpoint,
-                idempotency_endpoint=idempotency_endpoint,
-                idempotency_key=idempotency_key,
+                logical_request_payload=effective_request_payload,
+                legacy_request_payload=request_payload,
+                operation_kind=execution_kind,
+                wait_timeout_seconds=(
+                    idempotency_wait_seconds
+                    if execution_kind == "upstream_mcp"
+                    else 0.0
+                ),
+            )
+            replay = idem_begin.replay
+            idem_started = True
+            if owned_record is not None:
+                owned_record["record_id"] = idem_begin.record_id
+        except (IdempotencyConflictError, IdempotencyInProgressError) as exc:
+            await _audit_mcp_invocation(
+                effects_committed=False,
+                decision=tenant_decision,
+                endpoint=endpoint,
                 transport=transport,
-                request_id=request_id,
-                effective_request_payload=effective_request_payload,
-                effective_request_hash=effective_request_hash,
+                ok=False,
+                error=str(exc),
+                extra_metadata={
+                    "permit_id": permit_id,
+                    "idempotency_key": idempotency_key,
+                    "request_hash": effective_request_hash,
+                },
             )
+            if isinstance(exc, IdempotencyInProgressError):
+                raise
+            raise ValueError(str(exc)) from exc
+        if replay and replay.response_json:
+            if permit_id:
+                await _assert_governed_replay_access(
+                    permit_id=permit_id,
+                    wallet_id=wallet_id,
+                    tool_name=tool_name,
+                    key_id=auth.key_id,
+                    replay=replay,
+                )
+            await _raise_replayed_error(replay)
+            return replay.response_json
+
+    if governed_call and permit_id and action_identity is None:
+        # Decide permit refusals without a registered price. Denials authorize
+        # no credits and cannot reveal whether this tool is executable.
+        await _deny_tool_outside_permit(
+            idem=idem,
+            idem_started=idem_started,
+            permit_id=permit_id,
+            wallet_id=wallet_id,
+            auth=auth,
+            tool_name=tool_name,
+            arguments=arguments,
+            endpoint=endpoint,
+            idempotency_endpoint=idempotency_endpoint,
+            idempotency_key=idempotency_key,
+            transport=transport,
+            request_id=request_id,
+            effective_request_payload=effective_request_payload,
+            effective_request_hash=effective_request_hash,
+        )
+
+    if not service:
         reason = f"Tool not found: {tool_name}"
         await _complete_governed_denial_idempotency(
             idem=idem,
@@ -1576,13 +1632,6 @@ async def _execute_registered_tool_inner(
     category = ServiceCategory(
         service.get("category", ServiceCategory.PLATFORM_FEE.value)
     )
-    # High-value tools can force the permit path even when legacy
-    # unpermitted MCP is otherwise allowed.
-    if service.get("require_permit"):
-        governed_call = True
-    if governed_call and action_identity is None:
-        idempotency_endpoint = GOVERNED_MCP_IDEMPOTENCY_ENDPOINT
-
     registered_cost = _registered_tool_cost(service, category)
 
     # A quote is a signed commitment to a price. Resolve it before anything
@@ -1590,40 +1639,7 @@ async def _execute_registered_tool_inner(
     # check, and the charge all see the number the caller was promised — not
     # the live price it may have drifted from. Nothing is spent here; the
     # single-use consume happens immediately before the charge.
-    quoted = None
-    if quote_id:
-        quoted = await get_quote_service().validate_for_action(
-            quote_id=quote_id,
-            wallet_id=wallet_id,
-            tool_name=tool_name,
-        )
-        if not quoted.allowed or quoted.quote is None:
-            reason = quoted.reason or "quote_invalid"
-            await _audit_mcp_invocation(
-                effects_committed=False,
-                decision=tenant_decision,
-                endpoint=endpoint,
-                transport=transport,
-                ok=False,
-                error=reason,
-                extra_metadata={
-                    "quote_id": quote_id,
-                    "permit_id": permit_id,
-                    "idempotency_key": idempotency_key,
-                    "request_hash": effective_request_hash,
-                },
-            )
-            await _complete_governed_denial_idempotency(
-                idem=idem,
-                idem_started=idem_started,
-                wallet_id=wallet_id,
-                endpoint=idempotency_endpoint,
-                idempotency_key=idempotency_key,
-                reason=reason,
-            )
-            # Denying beats quietly charging a different number: a price lock
-            # that silently reprices is worse than no lock at all.
-            raise PermissionError(reason)
+    if quoted is not None and quoted.quote is not None:
         registered_cost = quoted.quote.quoted_credits
 
     charge_units = _charge_units_for_registered_cost(registered_cost, category)
@@ -1660,29 +1676,7 @@ async def _execute_registered_tool_inner(
                 "request_hash": effective_request_hash,
             },
         )
-        await _complete_governed_denial_idempotency(
-            idem=idem,
-            idem_started=idem_started,
-            wallet_id=wallet_id,
-            endpoint=idempotency_endpoint,
-            idempotency_key=idempotency_key,
-            reason="permit_required",
-        )
         raise PermissionError("permit_required")
-    if governed_call and not idempotency_key:
-        await _audit_mcp_invocation(
-            effects_committed=False,
-            decision=decision,
-            endpoint=endpoint,
-            transport=transport,
-            ok=False,
-            error="idempotency_key_required",
-            extra_metadata={
-                "permit_id": permit_id,
-                "request_hash": effective_request_hash,
-            },
-        )
-        raise ValueError("idempotency_key_required")
 
     if action_identity is not None:
         existing = await idem.get_action_record(
@@ -1722,37 +1716,19 @@ async def _execute_registered_tool_inner(
                         admission.reason or "action_permit_denied"
                     )
 
-    if governed_call and idempotency_key and not idem_started:
+    if governed_call and idempotency_key and action_identity is not None:
         try:
-            if action_identity is not None:
-                idem_begin = await idem.begin_action_with_record(
-                    wallet_id=wallet_id,
-                    identity=action_identity,
-                    wait_timeout_seconds=idempotency_wait_seconds,
-                )
-            else:
-                idem_begin = await _begin_governed_mcp_idempotency(
-                    idem=idem,
-                    wallet_id=wallet_id,
-                    idempotency_key=idempotency_key,
-                    tool_name=tool_name,
-                    endpoint=endpoint,
-                    logical_request_payload=effective_request_payload,
-                    legacy_request_payload=request_payload,
-                    operation_kind=execution_backend,
-                    wait_timeout_seconds=(
-                        idempotency_wait_seconds
-                        if execution_backend == "upstream_mcp"
-                        else 0.0
-                    ),
-                )
+            idem_begin = await idem.begin_action_with_record(
+                wallet_id=wallet_id,
+                identity=action_identity,
+                wait_timeout_seconds=idempotency_wait_seconds,
+            )
             replay = idem_begin.replay
             idem_started = True
             if owned_record is not None:
                 owned_record["record_id"] = idem_begin.record_id
-                if action_identity is not None:
-                    owned_record["endpoint"] = action_identity.endpoint
-                    owned_record["key"] = action_identity.idempotency_key
+                owned_record["endpoint"] = action_identity.endpoint
+                owned_record["key"] = action_identity.idempotency_key
         except (IdempotencyConflictError, IdempotencyInProgressError) as exc:
             await _audit_mcp_invocation(
                 effects_committed=False,
@@ -1771,19 +1747,18 @@ async def _execute_registered_tool_inner(
                 raise
             raise ValueError(str(exc))
         if replay and replay.response_json:
-            if action_identity is not None:
-                # A waiter may observe completion after authority was revoked.
-                # This denial must never finalize or overwrite the accepted owner.
-                await _assert_action_current_access(
-                    permit_id=permit_id or "",
-                    wallet_id=wallet_id,
-                    auth=auth,
-                    tool_name=tool_name,
-                    arguments=arguments,
-                    money=money,
-                    estimated_cost=Decimal("0"),
-                    phase="replay",
-                )
+            # A waiter may observe completion after authority was revoked.
+            # This denial must never finalize or overwrite the accepted owner.
+            await _assert_action_current_access(
+                permit_id=permit_id or "",
+                wallet_id=wallet_id,
+                auth=auth,
+                tool_name=tool_name,
+                arguments=arguments,
+                money=money,
+                estimated_cost=Decimal("0"),
+                phase="replay",
+            )
             await _assert_governed_replay_access(
                 permit_id=permit_id,
                 wallet_id=wallet_id,
@@ -4636,7 +4611,7 @@ async def _audit_mcp_invocation(
             error=error,
             metadata={
                 "transport": transport,
-                "estimated_cost": decision.estimated_cost,
+                **({"estimated_cost": decision.estimated_cost} if ok else {}),
                 "policy_reason": decision.reason,
                 **(extra_metadata or {}),
             },
diff --git a/gateway/tests/test_permit_unknown_tool.py b/gateway/tests/test_permit_unknown_tool.py
index 1e51361c..bd77ac34 100644
--- a/gateway/tests/test_permit_unknown_tool.py
+++ b/gateway/tests/test_permit_unknown_tool.py
@@ -8,6 +8,8 @@
 
 from __future__ import annotations
 
+from decimal import Decimal
+
 import pytest
 from httpx import ASGITransport, AsyncClient
 from sqlalchemy import func, select
@@ -22,6 +24,8 @@
     DOGFOOD_TOOL_ID,
     sync_dogfood_tool_registration,
 )
+from app.services.audit_log import list_audit_events
+from app.services.quotes import get_quote_service
 from app.services.receipts import get_receipt_service
 from app.services.service_registry import get_service_registry
 from tests.test_mcp_trust_mode import (
@@ -340,6 +344,142 @@ async def _receipt_count(wallet_id: str) -> int:
         )
 
 
+@pytest.mark.anyio
+@pytest.mark.usefixtures("strict_trust_mode")
+@pytest.mark.parametrize("transport", ["jsonrpc", "rest"])
+@pytest.mark.parametrize(
+    ("case", "expected_reason"),
+    [
+        ("missing_key", "idempotency_key_required"),
+        ("missing_quote", "permit_tool_not_allowed"),
+        ("invalid_quote", "quote_not_found"),
+        ("denied_permit", "permit_tool_not_allowed"),
+    ],
+)
+async def test_denials_do_not_reveal_registration(
+    client, clean_database, monkeypatch, transport, case, expected_reason
+):
+    provisioned, permit = await _permit_for_allowed_tool(client)
+    wallet_id = provisioned["agent_wallet_id"]
+    quote_id = None
+    if case == "denied_permit":
+        quote = await get_quote_service().create_quote(
+            wallet_id=wallet_id,
+            tool=UNKNOWN,
+            quoted_credits=Decimal("1"),
+            category="agent_comms",
+        )
+        quote_id = quote.quote_id
+    elif case == "invalid_quote":
+        quote_id = "quote-does-not-exist"
+
+    errors = []
+    audits = []
+    for registered in (True, False):
+        monkeypatch.setenv("ENABLE_DOGFOOD_SECOND_TOOL", str(registered).lower())
+        get_settings.cache_clear()
+        sync_dogfood_tool_registration()
+        assert (get_service_registry().get_local(UNKNOWN) is not None) == registered
+
+        request_id = f"registration-{case}-{registered}"
+        idempotency_key = None if case == "missing_key" else request_id
+        if transport == "jsonrpc":
+            body = _jsonrpc_body(
+                tool_name=UNKNOWN,
+                wallet_id=wallet_id,
+                request_id=request_id,
+                permit_id=permit["permit_id"],
+                idempotency_key=idempotency_key,
+            )
+            if quote_id is not None:
+                body["params"]["mcpContext"]["quote_id"] = quote_id
+            resp = await client.post(
+                "/mcp/messages", json=body, headers=provisioned["agent_headers"]
+            )
+            assert resp.status_code == 200
+            error = resp.json()["error"]
+            assert error["message"] == expected_reason
+            assert error["code"] == -32003
+        else:
+            context = {"wallet_id": wallet_id, "permit_id": permit["permit_id"]}
+            if idempotency_key is not None:
+                context["idempotency_key"] = idempotency_key
+            if quote_id is not None:
+                context["quote_id"] = quote_id
+            resp = await client.post(
+                f"/mcp/tools/{UNKNOWN}/invoke",
+                json={
+                    "name": UNKNOWN,
+                    "arguments": {"message": "hello"},
+                    "mcp_context": context,
+                },
+                headers=provisioned["agent_headers"],
+            )
+            assert resp.status_code == (400 if case == "missing_key" else 403)
+            error = resp.json()["detail"]
+            assert (error.get("error") if isinstance(error, dict) else error) == (
+                expected_reason
+            )
+        errors.append(error)
+        events = await list_audit_events(
+            event="mcp.invoke",
+            wallet_id=wallet_id,
+            tool=UNKNOWN,
+            request_id=request_id if transport == "jsonrpc" else None,
+            ok=False,
+            limit=1,
+        )
+        assert len(events) == 1
+        assert events[0].error == expected_reason
+        metadata = dict(events[0].metadata)
+        assert "tool_registered" not in metadata
+        assert "estimated_cost" not in metadata
+        metadata.pop("idempotency_key", None)
+        audits.append(metadata)
+
+    assert audits[0] == audits[1]
+    for error in errors:
+        if transport == "jsonrpc":
+            receipt = error.get("data", {}).get("receipt")
+        else:
+            receipt = error.get("receipt") if isinstance(error, dict) else None
+        if case in {"missing_key", "invalid_quote"}:
+            assert receipt is None
+        else:
+            assert receipt is not None
+            assert (await get_receipt_service().verify_receipt(receipt["receipt_id"]))[
+                0
+            ]
+
+    def comparable(error):
+        if not isinstance(error, dict):
+            return error
+        normalized = dict(error)
+        container = (
+            dict(error.get("data", {})) if transport == "jsonrpc" else normalized
+        )
+        receipt = container.get("receipt")
+        if receipt is not None:
+            container["receipt"] = {
+                key: value
+                for key, value in receipt.items()
+                if key
+                not in {
+                    "receipt_id",
+                    "audit_event_id",
+                    "idempotency_record_id",
+                    "signature",
+                    "created_at",
+                    "issued_at",
+                }
+            }
+        if transport == "jsonrpc":
+            normalized["data"] = container
+        return normalized
+
+    assert comparable(errors[0]) == comparable(errors[1])
+
+
 @pytest.mark.anyio
 @pytest.mark.usefixtures("strict_trust_mode")
 async def test_foreign_wallet_permit_does_not_leak_allowlist(client, clean_database):
@@ -404,9 +544,9 @@ async def test_permit_denial_without_idempotency_key_mints_no_receipt(
         headers=provisioned["agent_headers"],
     )
     error = resp.json()["error"]
-    if error["code"] == -32003:
-        assert error["message"] == "permit_tool_not_allowed"
-        assert not error.get("data", {}).get("receipt")
+    assert error["code"] == -32003
+    assert error["message"] == "idempotency_key_required"
+    assert not error.get("data", {}).get("receipt")
     assert await _receipt_count(provisioned["agent_wallet_id"]) == before
 
 

From 93e81a25325e0bdb78e020d26e570ba4e9d903bc Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Mon, 26 Oct 2026 07:29:39 -0700
Subject: [PATCH 05/12] fix(permits): verify signature before reading the
 allowlist (#232)

validate_for_action now checks the permit signature right after the wallet and key binding, before allowed_tools, scopes or budgets are read or echoed. That covers the tool path and the action path. verify_signature fails closed when the signed payload cannot be rebuilt. Tests: forged-signature cases (tool and action) fail on the old order, uuid4 request and idempotency ids, and an exact error code in the no-key test.

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 gateway/app/services/permits.py           | 26 ++++---
 gateway/tests/test_permit_unknown_tool.py | 82 ++++++++++++++++++-----
 2 files changed, 84 insertions(+), 24 deletions(-)

diff --git a/gateway/app/services/permits.py b/gateway/app/services/permits.py
index 19053f7a..167b04c3 100644
--- a/gateway/app/services/permits.py
+++ b/gateway/app/services/permits.py
@@ -988,6 +988,17 @@ async def _validate_model_for_action(
             return PermitValidation(
                 False, "permit_key_mismatch", model, {"bound_to": "subject_key_id"}
             )
+        # Verify the signature before reading any permit content. Every check
+        # below reads signed fields (allowed_tools, scopes, budgets) and some
+        # echo them back in denial details, so a forged or tampered row must
+        # be rejected before its allowlist is trusted or disclosed.
+        if not await self.verify_signature(model, session=session):
+            return PermitValidation(
+                False,
+                "permit_signature_invalid",
+                model,
+                {"key_id": model.key_id},
+            )
         # An empty allowlist names no tool, so it admits none. The previous
         # `if allowed_tools and ...` form skipped this check for empty lists
         # and let the scope check below decide, so a permit carrying explicit
@@ -1109,13 +1120,6 @@ async def _validate_model_for_action(
                     {"field": hit, "forbidden_fields": forbidden},
                 )
 
-        if not await self.verify_signature(model, session=session):
-            return PermitValidation(
-                False,
-                "permit_signature_invalid",
-                model,
-                {"key_id": model.key_id},
-            )
         return PermitValidation(True, None, model)
 
     @staticmethod
@@ -2248,7 +2252,13 @@ async def verify_signature(
         *,
         session: AsyncSession | None = None,
     ) -> bool:
-        payload = self._verification_payload(model)
+        # A row whose signed fields cannot even be rebuilt (for example a
+        # partial action contract) cannot carry a valid signature. Fail
+        # closed instead of raising, now that this runs before other checks.
+        try:
+            payload = self._verification_payload(model)
+        except ValueError:
+            return False
         return await get_signing_key_service().verify_payload(
             payload,
             signature=model.signature,
diff --git a/gateway/tests/test_permit_unknown_tool.py b/gateway/tests/test_permit_unknown_tool.py
index 1e51361c..ee0dc32e 100644
--- a/gateway/tests/test_permit_unknown_tool.py
+++ b/gateway/tests/test_permit_unknown_tool.py
@@ -8,6 +8,8 @@
 
 from __future__ import annotations
 
+from uuid import uuid4
+
 import pytest
 from httpx import ASGITransport, AsyncClient
 from sqlalchemy import func, select
@@ -108,9 +110,9 @@ async def test_jsonrpc_unregistered_tool_outside_permit_is_permit_denial(
         json=_jsonrpc_body(
             tool_name=UNKNOWN,
             wallet_id=provisioned["agent_wallet_id"],
-            request_id="unknown-tool-permit-call",
+            request_id=f"unknown-tool-permit-call-{uuid4()}",
             permit_id=permit["permit_id"],
-            idempotency_key="unknown-tool-permit-idem",
+            idempotency_key=f"unknown-tool-permit-idem-{uuid4()}",
         ),
         headers=provisioned["agent_headers"],
     )
@@ -147,9 +149,9 @@ async def test_jsonrpc_permit_denial_for_unregistered_tool_replays(
     body = _jsonrpc_body(
         tool_name=UNKNOWN,
         wallet_id=provisioned["agent_wallet_id"],
-        request_id="unknown-tool-permit-replay",
+        request_id=f"unknown-tool-permit-replay-{uuid4()}",
         permit_id=permit["permit_id"],
-        idempotency_key="unknown-tool-permit-replay-idem",
+        idempotency_key=f"unknown-tool-permit-replay-idem-{uuid4()}",
     )
     first = await client.post(
         "/mcp/messages", json=body, headers=provisioned["agent_headers"]
@@ -254,9 +256,9 @@ async def test_unregistered_tool_named_by_permit_is_still_not_found(
         json=_jsonrpc_body(
             tool_name=UNKNOWN,
             wallet_id=provisioned["agent_wallet_id"],
-            request_id="named-unknown-rpc",
+            request_id=f"named-unknown-rpc-{uuid4()}",
             permit_id=permit["permit_id"],
-            idempotency_key="named-unknown-rpc-idem",
+            idempotency_key=f"named-unknown-rpc-idem-{uuid4()}",
         ),
         headers=provisioned["agent_headers"],
     )
@@ -272,8 +274,8 @@ async def test_unregistered_tool_without_permit_is_unchanged(client, clean_datab
         json=_jsonrpc_body(
             tool_name=UNKNOWN,
             wallet_id=provisioned["agent_wallet_id"],
-            request_id="no-permit-unknown",
-            idempotency_key="no-permit-unknown-idem",
+            request_id=f"no-permit-unknown-{uuid4()}",
+            idempotency_key=f"no-permit-unknown-idem-{uuid4()}",
         ),
         headers=provisioned["agent_headers"],
     )
@@ -371,9 +373,9 @@ async def test_foreign_wallet_permit_does_not_leak_allowlist(client, clean_datab
             json=_jsonrpc_body(
                 tool_name=UNKNOWN,
                 wallet_id=attacker["agent_wallet_id"],
-                request_id=f"foreign-permit-{i}",
+                request_id=f"foreign-permit-{uuid4()}",
                 permit_id=permit["permit_id"],
-                idempotency_key=f"foreign-permit-idem-{i}",
+                idempotency_key=f"foreign-permit-idem-{uuid4()}",
             ),
             headers=attacker["agent_headers"],
         )
@@ -398,15 +400,15 @@ async def test_permit_denial_without_idempotency_key_mints_no_receipt(
         json=_jsonrpc_body(
             tool_name=UNKNOWN,
             wallet_id=provisioned["agent_wallet_id"],
-            request_id="no-idem-unknown",
+            request_id=f"no-idem-unknown-{uuid4()}",
             permit_id=permit["permit_id"],
         ),
         headers=provisioned["agent_headers"],
     )
     error = resp.json()["error"]
-    if error["code"] == -32003:
-        assert error["message"] == "permit_tool_not_allowed"
-        assert not error.get("data", {}).get("receipt")
+    assert error["code"] == -32003
+    assert error["message"] == "permit_tool_not_allowed"
+    assert not error.get("data", {}).get("receipt")
     assert await _receipt_count(provisioned["agent_wallet_id"]) == before
 
 
@@ -427,9 +429,9 @@ async def test_action_permit_unregistered_tool_gets_permit_decision(
         json=_jsonrpc_body(
             tool_name=UNKNOWN,
             wallet_id=provisioned["agent_wallet_id"],
-            request_id="action-permit-unknown",
+            request_id=f"action-permit-unknown-{uuid4()}",
             permit_id=permit["permit_id"],
-            idempotency_key="action-permit-unknown-idem",
+            idempotency_key=f"action-permit-unknown-idem-{uuid4()}",
         ),
         headers=provisioned["agent_headers"],
     )
@@ -439,3 +441,51 @@ async def test_action_permit_unregistered_tool_gets_permit_decision(
     assert error["message"].startswith("permit_")
     # No idempotency record exists on the action path, so no receipt.
     assert await _receipt_count(provisioned["agent_wallet_id"]) == before
+
+
+async def _forge_signature(permit_id: str) -> None:
+    async with get_session_factory()() as session:
+        model = await session.get(PermitModel, permit_id)
+        sig = model.signature
+        model.signature = ("A" if sig[0] != "A" else "B") + sig[1:]
+        await session.commit()
+
+
+@pytest.mark.anyio
+@pytest.mark.usefixtures("strict_trust_mode")
+@pytest.mark.parametrize("action", [False, True], ids=["tool", "action"])
+async def test_bad_signature_rejected_before_allowlist_is_read(
+    client, clean_database, action
+):
+    """A tampered permit is refused on its signature, never on its allowlist.
+
+    Before the fix the allowlist was read (and echoed) before the signature
+    was checked, so a forged row produced permit_tool_not_allowed with the
+    row's allowed_tools in the details.
+    """
+    provisioned, permit = await _permit_for_allowed_tool(client)
+    await _forge_signature(permit["permit_id"])
+    if action:
+        async with get_session_factory()() as session:
+            model = await session.get(PermitModel, permit["permit_id"])
+            model.action_contract_version = 1
+            await session.commit()
+    before = await _receipt_count(provisioned["agent_wallet_id"])
+    resp = await client.post(
+        "/mcp/messages",
+        json=_jsonrpc_body(
+            tool_name=UNKNOWN,
+            wallet_id=provisioned["agent_wallet_id"],
+            request_id=f"forged-permit-{uuid4()}",
+            permit_id=permit["permit_id"],
+            idempotency_key=f"forged-permit-idem-{uuid4()}",
+        ),
+        headers=provisioned["agent_headers"],
+    )
+    error = resp.json()["error"]
+    assert error["code"] == -32003
+    assert error["message"] == "permit_signature_invalid"
+    assert "allowed_tools" not in str(error)
+    assert ALLOWED not in str(error)
+    if action:
+        assert await _receipt_count(provisioned["agent_wallet_id"]) == before

From fc994a36dc3124b945b3c7b69057420b72a88d45 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Mon, 26 Oct 2026 20:49:34 -0700
Subject: [PATCH 06/12] fix(permits): verify the signature before any permit
 field (#232)

Move the signature check in validate_for_action ahead of the status,
expiry and wallet/key binding checks, not just ahead of allowed_tools, so
a forged row is answered only with permit_signature_invalid and never
reaches the denial-receipt paths. New test: a forged row also marked
revoked still gets permit_signature_invalid (fails on the previous order).
Tests that edit signed fields in place (backdated expiry, planted limits,
emptied allowlist) now re-sign the row through a shared helper, and the
REST test uses a uuid4 idempotency key.

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 gateway/app/services/permits.py               | 23 ++++++------
 gateway/scripts/red_team_trust_plane.py       |  9 +++++
 gateway/tests/test_mcp_upstream_governed.py   |  2 ++
 gateway/tests/test_permit_break_it.py         |  3 +-
 gateway/tests/test_permit_unknown_tool.py     | 35 +++++++++++++++++--
 gateway/tests/test_permit_v2_constraints.py   |  2 ++
 gateway/tests/test_permits.py                 |  4 +++
 gateway/tests/test_trust_helpers.py           | 17 +++++++++
 gateway/tests/test_trust_negative_security.py |  8 ++++-
 9 files changed, 88 insertions(+), 15 deletions(-)

diff --git a/gateway/app/services/permits.py b/gateway/app/services/permits.py
index 167b04c3..934e39fd 100644
--- a/gateway/app/services/permits.py
+++ b/gateway/app/services/permits.py
@@ -952,6 +952,18 @@ async def _validate_model_for_action(
         # reported as allowed and later subtracted from spent_credits.
         if not _storable_reservation_amount(estimated_credits):
             return PermitValidation(False, "permit_credits_invalid", model)
+        # Verify the signature before reading any permit content. Every check
+        # below reads permit fields (status, binding, allowed_tools, scopes,
+        # budgets) and some echo them back in denial details, so a forged or
+        # tampered row must be rejected before any of it is trusted or
+        # disclosed.
+        if not await self.verify_signature(model, session=session):
+            return PermitValidation(
+                False,
+                "permit_signature_invalid",
+                model,
+                {"key_id": model.key_id},
+            )
         now = utc_now()
         if model.status != "active":
             return PermitValidation(
@@ -988,17 +1000,6 @@ async def _validate_model_for_action(
             return PermitValidation(
                 False, "permit_key_mismatch", model, {"bound_to": "subject_key_id"}
             )
-        # Verify the signature before reading any permit content. Every check
-        # below reads signed fields (allowed_tools, scopes, budgets) and some
-        # echo them back in denial details, so a forged or tampered row must
-        # be rejected before its allowlist is trusted or disclosed.
-        if not await self.verify_signature(model, session=session):
-            return PermitValidation(
-                False,
-                "permit_signature_invalid",
-                model,
-                {"key_id": model.key_id},
-            )
         # An empty allowlist names no tool, so it admits none. The previous
         # `if allowed_tools and ...` form skipped this check for empty lists
         # and let the scope check below decide, so a permit carrying explicit
diff --git a/gateway/scripts/red_team_trust_plane.py b/gateway/scripts/red_team_trust_plane.py
index 130fe419..2ebc2962 100644
--- a/gateway/scripts/red_team_trust_plane.py
+++ b/gateway/scripts/red_team_trust_plane.py
@@ -71,7 +71,9 @@ def configure_environment() -> None:
 from app.db.models import PermitModel  # noqa: E402
 from app.main import app  # noqa: E402
 from app.schemas.billing import ServiceCategory  # noqa: E402
+from app.services.permits import get_permit_service  # noqa: E402
 from app.services.service_registry import get_service_registry  # noqa: E402
+from app.services.signing_keys import get_signing_key_service  # noqa: E402
 
 
 def remove_demo_db() -> None:
@@ -223,6 +225,13 @@ async def _force_permit_expired(permit_id: str) -> None:
         )
         model = result.scalar_one()
         model.expires_at = datetime.now(timezone.utc) - timedelta(minutes=5)
+        # The signature is verified before expiry is read, so re-sign the
+        # row to model a genuinely expired permit rather than a forged one.
+        signature, kid, _ = await get_signing_key_service().sign_payload(
+            get_permit_service()._unsigned_payload(model)
+        )
+        model.signature = signature
+        model.key_id = kid
         session.add(model)
         await session.commit()
 
diff --git a/gateway/tests/test_mcp_upstream_governed.py b/gateway/tests/test_mcp_upstream_governed.py
index 7aa8a573..fd0e23ea 100644
--- a/gateway/tests/test_mcp_upstream_governed.py
+++ b/gateway/tests/test_mcp_upstream_governed.py
@@ -62,6 +62,7 @@
     register_configured_upstream_mcp,
 )
 from tests.test_trust_helpers import (
+    resign_permit_model,
     BOOTSTRAP_HEADERS,
     create_tool_permit,
     provision_agent_wallet,
@@ -2777,6 +2778,7 @@ async def test_upstream_permit_denials_never_charge_or_dispatch(
                 model = await session.get(PermitModel, permit["permit_id"])
                 assert model is not None
                 model.expires_at = utc_now() - timedelta(seconds=1)
+                await resign_permit_model(model)
                 session.add(model)
                 await session.commit()
 
diff --git a/gateway/tests/test_permit_break_it.py b/gateway/tests/test_permit_break_it.py
index 7068a90d..24933558 100644
--- a/gateway/tests/test_permit_break_it.py
+++ b/gateway/tests/test_permit_break_it.py
@@ -25,7 +25,7 @@
 from app.services.ownership import CallerIdentity
 from app.services.permits import PermitError, get_permit_service
 from app.services.receipts import get_receipt_service
-from tests.test_trust_helpers import provision_agent_wallet
+from tests.test_trust_helpers import provision_agent_wallet, resign_permit_model
 
 
 @pytest.fixture
@@ -84,6 +84,7 @@ async def _backdate_expiry(permit_id: str) -> None:
             model = await session.get(PermitModel, permit_id)
             assert model is not None
             model.expires_at = utc_now() - timedelta(minutes=5)
+            await resign_permit_model(model)
             session.add(model)
 
 
diff --git a/gateway/tests/test_permit_unknown_tool.py b/gateway/tests/test_permit_unknown_tool.py
index ee0dc32e..c7f9bc4e 100644
--- a/gateway/tests/test_permit_unknown_tool.py
+++ b/gateway/tests/test_permit_unknown_tool.py
@@ -211,7 +211,7 @@ async def test_rest_unregistered_tool_outside_permit_is_403_with_receipt(
             "mcp_context": {
                 "wallet_id": provisioned["agent_wallet_id"],
                 "permit_id": permit["permit_id"],
-                "idempotency_key": "unknown-tool-permit-rest-idem",
+                "idempotency_key": f"unknown-tool-permit-rest-idem-{uuid4()}",
             },
         },
         headers=provisioned["agent_headers"],
@@ -367,7 +367,7 @@ async def test_foreign_wallet_permit_does_not_leak_allowlist(client, clean_datab
     )
     attacker = await provision_agent_wallet(client)
     errors = []
-    for i, permit in enumerate((names_unknown, names_other)):
+    for permit in (names_unknown, names_other):
         resp = await client.post(
             "/mcp/messages",
             json=_jsonrpc_body(
@@ -489,3 +489,34 @@ async def test_bad_signature_rejected_before_allowlist_is_read(
     assert ALLOWED not in str(error)
     if action:
         assert await _receipt_count(provisioned["agent_wallet_id"]) == before
+
+
+@pytest.mark.anyio
+@pytest.mark.usefixtures("strict_trust_mode")
+async def test_bad_signature_rejected_before_any_permit_field(client, clean_database):
+    """The signature is checked before status, expiry or binding are read.
+
+    A forged row that is also marked revoked must still be refused as
+    permit_signature_invalid, not answered from its unsigned status.
+    """
+    provisioned, permit = await _permit_for_allowed_tool(client)
+    await _forge_signature(permit["permit_id"])
+    async with get_session_factory()() as session:
+        model = await session.get(PermitModel, permit["permit_id"])
+        model.status = "revoked"
+        await session.commit()
+    resp = await client.post(
+        "/mcp/messages",
+        json=_jsonrpc_body(
+            tool_name=UNKNOWN,
+            wallet_id=provisioned["agent_wallet_id"],
+            request_id=f"forged-revoked-{uuid4()}",
+            permit_id=permit["permit_id"],
+            idempotency_key=f"forged-revoked-idem-{uuid4()}",
+        ),
+        headers=provisioned["agent_headers"],
+    )
+    error = resp.json()["error"]
+    assert error["code"] == -32003
+    assert error["message"] == "permit_signature_invalid"
+    assert "allowed_tools" not in str(error)
diff --git a/gateway/tests/test_permit_v2_constraints.py b/gateway/tests/test_permit_v2_constraints.py
index 3ff0b647..5cc5b785 100644
--- a/gateway/tests/test_permit_v2_constraints.py
+++ b/gateway/tests/test_permit_v2_constraints.py
@@ -22,6 +22,7 @@
 from app.services.permits import get_permit_service
 from app.services.service_registry import get_service_registry
 from tests.test_trust_helpers import (
+    resign_permit_model,
     BOOTSTRAP_HEADERS,
     provision_agent_wallet,
 )
@@ -1057,6 +1058,7 @@ async def test_max_calls_fractional_limit_fails_closed(client, clean_database):
             model = await session.get(PermitModel, permit_id)
             assert model is not None
             model.max_calls_per_tool_json = json.dumps({tool_name: 2.5})
+            await resign_permit_model(model)
             session.add(model)
             await session.commit()
 
diff --git a/gateway/tests/test_permits.py b/gateway/tests/test_permits.py
index 80e87e7d..70032b6d 100644
--- a/gateway/tests/test_permits.py
+++ b/gateway/tests/test_permits.py
@@ -21,6 +21,7 @@
 from app.services.ownership import CallerIdentity
 from app.services.permits import PermitError, get_permit_service
 from tests.test_trust_helpers import (
+    resign_permit_model,
     BOOTSTRAP_HEADERS,
     create_tool_permit,
     provision_agent_wallet,
@@ -640,6 +641,7 @@ async def test_authorize_and_reserve_rechecks_terminal_state_after_stale_prechec
             model.revoked_at = utc_now()
         else:
             model.expires_at = utc_now() - timedelta(seconds=1)
+            await resign_permit_model(model)
         session.add(model)
         await session.commit()
 
@@ -776,6 +778,7 @@ async def test_permit_service_validation_denial_reasons(client, clean_database):
         assert model is not None
         model.allowed_tools_json = json.dumps([])
         model.scopes_json = json.dumps(["billing:charge"])
+        await resign_permit_model(model)
         session.add(model)
         await session.commit()
 
@@ -853,6 +856,7 @@ async def test_reserve_budget_refuses_an_expired_permit(
             model = await session.get(PermitModel, permit.permit_id)
             assert model is not None
             model.expires_at = utc_now() - timedelta(seconds=1)
+            await resign_permit_model(model)
             session.add(model)
 
     with pytest.raises(PermitError) as excinfo:
diff --git a/gateway/tests/test_trust_helpers.py b/gateway/tests/test_trust_helpers.py
index 3b84fb37..a9b740d8 100644
--- a/gateway/tests/test_trust_helpers.py
+++ b/gateway/tests/test_trust_helpers.py
@@ -83,3 +83,20 @@ async def create_tool_permit(
     )
     assert permit_resp.status_code == 201
     return permit_resp.json()
+
+
+async def resign_permit_model(model: Any) -> None:
+    """Re-sign a permit row after a test edits its signed fields in place.
+
+    validate_for_action verifies the signature before reading any permit
+    field, so a test that backdates expires_at or plants a malformed limit
+    must re-sign the row to reach the check it is exercising.
+    """
+    from app.services.permits import get_permit_service
+    from app.services.signing_keys import get_signing_key_service
+
+    signature, kid, _ = await get_signing_key_service().sign_payload(
+        get_permit_service()._unsigned_payload(model)
+    )
+    model.signature = signature
+    model.key_id = kid
diff --git a/gateway/tests/test_trust_negative_security.py b/gateway/tests/test_trust_negative_security.py
index a0550f4f..565e3aa2 100644
--- a/gateway/tests/test_trust_negative_security.py
+++ b/gateway/tests/test_trust_negative_security.py
@@ -27,7 +27,11 @@
 from app.services.policies import evaluate_wallet_policy as real_evaluate_wallet_policy
 from app.services.receipts import get_receipt_service
 from app.services.service_registry import get_service_registry
-from tests.test_trust_helpers import create_tool_permit, provision_agent_wallet
+from tests.test_trust_helpers import (
+    create_tool_permit,
+    provision_agent_wallet,
+    resign_permit_model,
+)
 
 
 @pytest.fixture
@@ -101,6 +105,7 @@ async def test_expired_permit_is_denied_without_charge(client, clean_database):
             )
             model = result.scalar_one()
             model.expires_at = datetime.now(timezone.utc) - timedelta(minutes=5)
+            await resign_permit_model(model)
             session.add(model)
             await session.commit()
 
@@ -197,6 +202,7 @@ async def transition_permit_during_policy(**kwargs):
                     model.revoked_at = utc_now()
                 else:
                     model.expires_at = utc_now() - timedelta(seconds=1)
+                await resign_permit_model(model)
                 session.add(model)
                 await session.commit()
             return await real_evaluate_wallet_policy(**kwargs)

From 55d976cd69f0744a5be36db509a42fece40301e3 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Mon, 26 Oct 2026 20:52:21 -0700
Subject: [PATCH 07/12] test: re-sign the backdated permit in the expiry-denial
 test (#232)

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 gateway/tests/test_denial_details_and_self_service.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/gateway/tests/test_denial_details_and_self_service.py b/gateway/tests/test_denial_details_and_self_service.py
index df0040ca..bbb661bd 100644
--- a/gateway/tests/test_denial_details_and_self_service.py
+++ b/gateway/tests/test_denial_details_and_self_service.py
@@ -34,6 +34,7 @@
     BOOTSTRAP_HEADERS,
     create_tool_permit,
     provision_agent_wallet,
+    resign_permit_model,
 )
 
 TOOL = "denial-echo"
@@ -163,6 +164,7 @@ async def test_expired_denial_says_when_it_expired(
         model.expires_at = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(
             minutes=5
         )
+        await resign_permit_model(model)
         session.add(model)
         await session.commit()
 

From 0938d76593dfcbe92f2706e0031d1e6be3dfb169 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Mon, 26 Oct 2026 21:43:37 -0700
Subject: [PATCH 08/12] test(permits): per-item registration-parity tests for
 DeepSeek review

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 gateway/tests/test_permit_unknown_tool.py | 116 ++++++++++++++++++++++
 1 file changed, 116 insertions(+)

diff --git a/gateway/tests/test_permit_unknown_tool.py b/gateway/tests/test_permit_unknown_tool.py
index e5fa91d3..6f584bd4 100644
--- a/gateway/tests/test_permit_unknown_tool.py
+++ b/gateway/tests/test_permit_unknown_tool.py
@@ -659,3 +659,119 @@ async def test_bad_signature_rejected_before_any_permit_field(client, clean_data
     assert error["code"] == -32003
     assert error["message"] == "permit_signature_invalid"
     assert "allowed_tools" not in str(error)
+
+
+# DeepSeek review 5653058263: one test per leak path. Each sends the same
+# permitted-but-denied call twice, once with UNKNOWN registered (registered
+# but not allowed by the permit) and once unregistered, and compares a single
+# caller-visible surface so each item fails on its own on the old code.
+_VOLATILE_RECEIPT_FIELDS = {
+    "receipt_id",
+    "audit_event_id",
+    "idempotency_record_id",
+    "signature",
+    "created_at",
+    "issued_at",
+}
+
+
+async def _denial_for_each_registry_state(
+    client, monkeypatch, *, idempotency=True, quote_id=None
+):
+    provisioned, permit = await _permit_for_allowed_tool(client)
+    wallet_id = provisioned["agent_wallet_id"]
+    results = {}
+    for registered in (True, False):
+        monkeypatch.setenv("ENABLE_DOGFOOD_SECOND_TOOL", str(registered).lower())
+        get_settings.cache_clear()
+        sync_dogfood_tool_registration()
+        assert (get_service_registry().get_local(UNKNOWN) is not None) == registered
+        request_id = f"parity-{uuid4()}"
+        body = _jsonrpc_body(
+            tool_name=UNKNOWN,
+            wallet_id=wallet_id,
+            request_id=request_id,
+            permit_id=permit["permit_id"],
+            idempotency_key=f"parity-idem-{uuid4()}" if idempotency else None,
+        )
+        if quote_id is not None:
+            body["params"]["mcpContext"]["quote_id"] = quote_id
+        resp = await client.post(
+            "/mcp/messages", json=body, headers=provisioned["agent_headers"]
+        )
+        error = resp.json()["error"]
+        events = await list_audit_events(
+            event="mcp.invoke",
+            wallet_id=wallet_id,
+            tool=UNKNOWN,
+            request_id=request_id,
+            ok=False,
+            limit=1,
+        )
+        assert len(events) == 1
+        metadata = dict(events[0].metadata)
+        metadata.pop("idempotency_key", None)
+        metadata.pop("request_hash", None)
+        metadata.pop("request_id", None)
+        results[registered] = (error, events[0].error, metadata)
+    return results
+
+
+@pytest.mark.anyio
+@pytest.mark.usefixtures("strict_trust_mode")
+async def test_item1_audit_metadata_identical_for_registry_state(
+    client, clean_database, monkeypatch
+):
+    results = await _denial_for_each_registry_state(client, monkeypatch)
+    (_, reg_err, reg_meta), (_, unreg_err, unreg_meta) = results[True], results[False]
+    assert reg_err == unreg_err == "permit_tool_not_allowed"
+    assert "tool_registered" not in reg_meta and "tool_registered" not in unreg_meta
+    assert reg_meta == unreg_meta
+
+
+@pytest.mark.anyio
+@pytest.mark.usefixtures("strict_trust_mode")
+async def test_item2_denial_receipt_identical_for_registry_state(
+    client, clean_database, monkeypatch
+):
+    results = await _denial_for_each_registry_state(client, monkeypatch)
+    receipts = []
+    for registered in (True, False):
+        error = results[registered][0]
+        assert error["code"] == -32003
+        assert error["message"] == "permit_tool_not_allowed"
+        receipt = error["data"]["receipt"]
+        assert Decimal(str(receipt["credits_authorized"])) == 0
+        receipts.append(
+            {
+                k: v
+                for k, v in receipt.items()
+                if k not in _VOLATILE_RECEIPT_FIELDS | {"request_hash"}
+            }
+        )
+    assert receipts[0] == receipts[1]
+
+
+@pytest.mark.anyio
+@pytest.mark.usefixtures("strict_trust_mode")
+@pytest.mark.parametrize(
+    ("idempotency", "quote_id", "first_gate"),
+    [
+        (False, None, "idempotency_key_required"),
+        (True, "quote-does-not-exist", "quote_not_found"),
+    ],
+    ids=["no-idempotency-key", "bogus-quote"],
+)
+async def test_item3_same_gate_fires_first_for_registry_state(
+    client, clean_database, monkeypatch, idempotency, quote_id, first_gate
+):
+    results = await _denial_for_each_registry_state(
+        client, monkeypatch, idempotency=idempotency, quote_id=quote_id
+    )
+    for registered in (True, False):
+        error, audit_error, _ = results[registered]
+        assert error["code"] == -32003
+        assert error["message"] == first_gate, registered
+        assert audit_error == first_gate
+        assert not error.get("data", {}).get("receipt")
+    assert results[True][2] == results[False][2]

From 916ce1cd71f622d53ebc694dec3ce19a242cbd60 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Mon, 26 Oct 2026 21:51:36 -0700
Subject: [PATCH 09/12] test(permits): generate made-up quote and request ids
 at runtime

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 gateway/tests/test_permit_unknown_tool.py | 17 ++++++++++-------
 1 file changed, 10 insertions(+), 7 deletions(-)

diff --git a/gateway/tests/test_permit_unknown_tool.py b/gateway/tests/test_permit_unknown_tool.py
index 6f584bd4..81ee7083 100644
--- a/gateway/tests/test_permit_unknown_tool.py
+++ b/gateway/tests/test_permit_unknown_tool.py
@@ -372,7 +372,7 @@ async def test_denials_do_not_reveal_registration(
         )
         quote_id = quote.quote_id
     elif case == "invalid_quote":
-        quote_id = "quote-does-not-exist"
+        quote_id = f"quote-{uuid4()}"
 
     errors = []
     audits = []
@@ -382,7 +382,7 @@ async def test_denials_do_not_reveal_registration(
         sync_dogfood_tool_registration()
         assert (get_service_registry().get_local(UNKNOWN) is not None) == registered
 
-        request_id = f"registration-{case}-{registered}"
+        request_id = f"registration-{case}-{registered}-{uuid4()}"
         idempotency_key = None if case == "missing_key" else request_id
         if transport == "jsonrpc":
             body = _jsonrpc_body(
@@ -755,18 +755,21 @@ async def test_item2_denial_receipt_identical_for_registry_state(
 @pytest.mark.anyio
 @pytest.mark.usefixtures("strict_trust_mode")
 @pytest.mark.parametrize(
-    ("idempotency", "quote_id", "first_gate"),
+    ("idempotency", "bogus_quote", "first_gate"),
     [
-        (False, None, "idempotency_key_required"),
-        (True, "quote-does-not-exist", "quote_not_found"),
+        (False, False, "idempotency_key_required"),
+        (True, True, "quote_not_found"),
     ],
     ids=["no-idempotency-key", "bogus-quote"],
 )
 async def test_item3_same_gate_fires_first_for_registry_state(
-    client, clean_database, monkeypatch, idempotency, quote_id, first_gate
+    client, clean_database, monkeypatch, idempotency, bogus_quote, first_gate
 ):
     results = await _denial_for_each_registry_state(
-        client, monkeypatch, idempotency=idempotency, quote_id=quote_id
+        client,
+        monkeypatch,
+        idempotency=idempotency,
+        quote_id=f"quote-{uuid4()}" if bogus_quote else None,
     )
     for registered in (True, False):
         error, audit_error, _ = results[registered]

From f0916d6b8f9c8b9eef2da49158b2278b1c70af7f Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Mon, 26 Oct 2026 22:01:22 -0700
Subject: [PATCH 10/12] fix(mcp): release idempotency claim on Tool not
 found/not executable

A registry-state failure is no longer stored as the key's final answer, so a
same-key retry executes once the tool works. Fix the action-permit test to use
a validly signed action binding, update failure-semantics and
POLICY_ENFORCEMENT ordering, and replace a production assert.

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 gateway/app/routers/mcp.py                | 50 ++++++++-----
 gateway/app/routers/mcp_standard.py       |  3 +-
 gateway/docs/POLICY_ENFORCEMENT.md        | 13 ++--
 gateway/docs/failure-semantics.md         | 31 +++++---
 gateway/tests/test_permit_unknown_tool.py | 87 +++++++++++++++++++++--
 5 files changed, 145 insertions(+), 39 deletions(-)

diff --git a/gateway/app/routers/mcp.py b/gateway/app/routers/mcp.py
index 7e6a223b..a6cf1742 100644
--- a/gateway/app/routers/mcp.py
+++ b/gateway/app/routers/mcp.py
@@ -1578,14 +1578,12 @@ async def _execute_registered_tool_inner(
 
     if not service:
         reason = f"Tool not found: {tool_name}"
-        await _complete_governed_denial_idempotency(
+        await _release_unresolved_tool_idempotency(
             idem=idem,
-            idem_started=idem_started,
+            idem_begin=idem_begin if idem_started else None,
             wallet_id=wallet_id,
             endpoint=idempotency_endpoint,
             idempotency_key=idempotency_key,
-            reason=reason,
-            status_code=400,
         )
         raise ValueError(reason)
 
@@ -1594,38 +1592,32 @@ async def _execute_registered_tool_inner(
     upstream_executor = registry.get_executor(tool_name)
     if execution_backend == "local" and func is None:
         reason = f"Tool not executable: {tool_name}"
-        await _complete_governed_denial_idempotency(
+        await _release_unresolved_tool_idempotency(
             idem=idem,
-            idem_started=idem_started,
+            idem_begin=idem_begin if idem_started else None,
             wallet_id=wallet_id,
             endpoint=idempotency_endpoint,
             idempotency_key=idempotency_key,
-            reason=reason,
-            status_code=400,
         )
         raise ValueError(reason)
     if execution_backend == "upstream_mcp" and upstream_executor is None:
         reason = f"Tool not executable: {tool_name}"
-        await _complete_governed_denial_idempotency(
+        await _release_unresolved_tool_idempotency(
             idem=idem,
-            idem_started=idem_started,
+            idem_begin=idem_begin if idem_started else None,
             wallet_id=wallet_id,
             endpoint=idempotency_endpoint,
             idempotency_key=idempotency_key,
-            reason=reason,
-            status_code=400,
         )
         raise ValueError(reason)
     if execution_backend not in {"local", "upstream_mcp"}:
         reason = f"Tool not executable: {tool_name}"
-        await _complete_governed_denial_idempotency(
+        await _release_unresolved_tool_idempotency(
             idem=idem,
-            idem_started=idem_started,
+            idem_begin=idem_begin if idem_started else None,
             wallet_id=wallet_id,
             endpoint=idempotency_endpoint,
             idempotency_key=idempotency_key,
-            reason=reason,
-            status_code=400,
         )
         raise ValueError(reason)
 
@@ -3842,6 +3834,32 @@ def _governed_error_payload(
     return payload
 
 
+async def _release_unresolved_tool_idempotency(
+    *,
+    idem: Any,
+    idem_begin: IdempotencyBegin | None,
+    wallet_id: str,
+    endpoint: str,
+    idempotency_key: str | None,
+) -> None:
+    """Release a governed claim when the tool cannot run yet.
+
+    "Tool not found" / "Tool not executable" depend on the current registry,
+    not on the request, so they must not be stored as the key's final answer:
+    a same-key retry after the tool is registered or made executable has to
+    execute instead of replaying the old failure. Nothing has run and nothing
+    was charged, so the pinned record is simply deleted.
+    """
+    if idem_begin is None or not idempotency_key:
+        return
+    await idem.abandon(
+        wallet_id=wallet_id,
+        endpoint=endpoint,
+        idempotency_key=idempotency_key,
+        expected_record_id=idem_begin.record_id,
+    )
+
+
 async def _complete_governed_denial_idempotency(
     *,
     idem: Any,
diff --git a/gateway/app/routers/mcp_standard.py b/gateway/app/routers/mcp_standard.py
index 4c302030..d97939a9 100644
--- a/gateway/app/routers/mcp_standard.py
+++ b/gateway/app/routers/mcp_standard.py
@@ -508,7 +508,8 @@ async def _governed_tools_call(
     auto_minted = permit_id is None
     idempotency_key = client_idempotency_key or f"mcp-auto-{uuid.uuid4().hex}"
     if not permit_id:
-        assert record is not None  # Checked before any auto-mint.
+        if record is None:  # Checked before any auto-mint.
+            raise RuntimeError("auto-permit mint requires a resolved tool record")
         permit_id = await _mint_auto_permit(
             auth=auth,
             tool_name=tool_name,
diff --git a/gateway/docs/POLICY_ENFORCEMENT.md b/gateway/docs/POLICY_ENFORCEMENT.md
index 3d668961..05428303 100644
--- a/gateway/docs/POLICY_ENFORCEMENT.md
+++ b/gateway/docs/POLICY_ENFORCEMENT.md
@@ -45,11 +45,12 @@ only reachable in local/dev/test, and it logs a loud startup warning.
 | # | Gate | Source | On failure |
 |---|------|--------|------------|
 | 1 | **Wallet ownership — unpriced** (before the idempotency store is touched) | `evaluate_tool_invocation` | `wallet_access_denied` |
-| 2 | **Tool resolution + executability** | service registry | `Tool not found` / `Tool not executable` |
-| 3 | **Wallet ownership — priced** (re-run with real cost, for the audit record) | `evaluate_tool_invocation` | `wallet_access_denied` |
-| 4 | **Governed preconditions** | router | `permit_required` / `idempotency_key_required` |
-| 5 | **Idempotency begin** (completed prior call replays its stored signed receipt) | `get_idempotency_service()` | replay short-circuit |
-| 6 | **Permit validation** (Layer B) | `PermitService.validate_for_action` | `permit_*` |
+| 2 | **Governed preconditions** (independent of tool registration) | router | `idempotency_key_required` |
+| 3 | **Quote validation**, when a `quote_id` is supplied | quote service | `quote_*` |
+| 4 | **Idempotency begin** (completed prior call replays its stored signed receipt) | `get_idempotency_service()` | replay short-circuit |
+| 5 | **Permit validation** (Layer B), including `allowed_tools`, decided with no registered price. A denial is identical for registered and unregistered tools and, with an idempotency key, mints a signed denial receipt with `credits_authorized = 0` | `PermitService.validate_for_action` | `permit_*` |
+| 6 | **Tool resolution + executability**. The idempotency claim from step 4 is released, so a same-key retry runs normally once the tool is registered or executable | service registry | `Tool not found` / `Tool not executable` |
+| 6a | **Wallet ownership — priced** (re-run with real cost, for the audit record); calls without a permit get `permit_required` here on a resolved tool | `evaluate_tool_invocation` / router | `wallet_access_denied` / `permit_required` |
 | 7 | **Wallet policy bundles** (Layer C) | `evaluate_wallet_policy` | policy reason |
 | 8 | **Human-approval gate** (Layer D) | `_require_human_approval` | blocks / `human_approval_unavailable` |
 | 9 | **Recipient-domain binding** (Layer E) | router | `permit_recipient_domain_mismatch` |
@@ -82,6 +83,7 @@ capability bound to a wallet (and optionally an API key).
 
 | Check | Reason code |
 |-------|-------------|
+| Ed25519 signature over the permit verifies — checked **first**, right after the credits-amount check and before any other permit field is read | `permit_signature_invalid` |
 | Status must be `active` | `permit_{status}` (e.g. `permit_revoked`) |
 | Not past `expires_at` | `permit_expired` |
 | Permit's wallet == caller wallet | `permit_wallet_mismatch` |
@@ -92,7 +94,6 @@ capability bound to a wallet (and optionally an API key).
 | Per-tool call cap `max_calls_per_tool` (v2) — atomically reserved on local governed tools and remote (`upstream_mcp`) tools via a durable `call_slot_reserved` marker linked to the prepared dispatch attempt; cross-key retries are denied `permit_max_calls_exceeded` once the cap is reached; same-key retries replay the stored receipt | `permit_max_calls_exceeded` |
 | Cumulative `aggregate_value_cap` (v2) — enforced on the local MCP and AWI HTTP paths as a predicate of the atomic reservation against `spent_credits` floored to receipt history, so in-flight reservations and already-receipted charges both count; configured upstream calls fail closed | `permit_aggregate_value_cap_exceeded` locally; `permit_constraint_unsupported_for_upstream` remotely |
 | `forbidden_fields` (v2): deep scan of tool arguments for banned keys | `permit_forbidden_field:{field}` |
-| Ed25519 signature over the permit verifies — checked **last** | `permit_signature_invalid` |
 
 The local per-tool call limit is enforced with a persisted reservation counter,
 including an optimistic compare-and-swap when the database does not honor the
diff --git a/gateway/docs/failure-semantics.md b/gateway/docs/failure-semantics.md
index 5a366e39..913fddda 100644
--- a/gateway/docs/failure-semantics.md
+++ b/gateway/docs/failure-semantics.md
@@ -55,17 +55,26 @@ classified as already sent; they are never claimable again.
 | `failed_unrefunded` | A refund was owed and the refund **itself** failed | Charged; a durable operator work item is created atomically with the receipt | Depends on the underlying failure | 500 / `-32603` | `test_governed_refund_failure_keeps_permit_budget_reserved`, `test_refund_reconciliation_retries_exactly_once_and_preserves_agent_replay` |
 
 Requests rejected before a valid permit and an executable tool are established
-produce no receipt — there is no authority to bill against. The README states
-this limit. What they leave behind differs by reason:
-
-- An unknown tool (`Tool not found`) or `permit_not_found` on a governed call
-  that carries an idempotency key completes a terminal *idempotency record*;
-  a same-key replay returns the original error envelope.
-- `permit_required` on a resolved tool, and a resolved tool that is not
-  executable, record nothing: the governed idempotency record is only begun
-  after the permit check, so the completion call in `app/routers/mcp.py` is a
-  no-op. A same-key retry returns the same error while nothing has changed,
-  and after minting a permit it executes normally
+produce no billing receipt — there is no authority to bill against. The README
+states this limit. What they leave behind differs by reason:
+
+- A governed call with a valid permit that names a tool outside the permit's
+  `allowed_tools` gets 403 / `-32003` `permit_tool_not_allowed`, whether or not
+  the tool is registered. With an idempotency key it mints a signed denial
+  receipt (`credits_authorized = 0`, `credits_charged = 0`) and completes a
+  terminal idempotency record, so a same-key replay returns the same envelope
+  (`test_denials_do_not_reveal_registration`).
+- `permit_not_found` on a governed call that carries an idempotency key
+  completes a terminal *idempotency record*; a same-key replay returns the
+  original error envelope.
+- `Tool not found` and `Tool not executable` record nothing: they depend on the
+  current registry, not on the request, so the governed idempotency claim is
+  released (`_release_unresolved_tool_idempotency` in `app/routers/mcp.py`).
+  A same-key retry executes normally once the tool is registered or executable
+  (`test_unresolved_tool_failure_is_not_replayed_after_fix`).
+- `permit_required` on a resolved tool records nothing. A same-key retry returns
+  the same error while nothing has changed, and after minting a permit it
+  executes normally
   (`test_strict_mode_missing_permit_denial_leaves_no_idempotency_record`).
   Whether that stays the contract is the open D2 decision in
   [authority-required-flow.md](authority-required-flow.md).
diff --git a/gateway/tests/test_permit_unknown_tool.py b/gateway/tests/test_permit_unknown_tool.py
index 81ee7083..b6ca52a2 100644
--- a/gateway/tests/test_permit_unknown_tool.py
+++ b/gateway/tests/test_permit_unknown_tool.py
@@ -35,7 +35,11 @@
     _jsonrpc_body,
     strict_trust_mode,  # noqa: F401 - fixture
 )
-from tests.test_trust_helpers import create_tool_permit, provision_agent_wallet
+from tests.test_trust_helpers import (
+    create_tool_permit,
+    provision_agent_wallet,
+    resign_permit_model,
+)
 
 MCP_HEADERS = {
     "Accept": "application/json, text/event-stream",
@@ -560,7 +564,15 @@ async def test_action_permit_unregistered_tool_gets_permit_decision(
     provisioned, permit = await _permit_for_allowed_tool(client)
     async with get_session_factory()() as session:
         model = await session.get(PermitModel, permit["permit_id"])
+        # A complete, validly signed action binding for ALLOWED, so the call
+        # reaches the permit-scope check instead of a signature/binding error.
         model.action_contract_version = 1
+        model.action_payload_hash = "a" * 64
+        model.action_schema_id = f"{ALLOWED}.args"
+        model.action_schema_version = "1"
+        model.action_public_tool_id = ALLOWED
+        model.action_upstream_binding_hash = "b" * 64
+        await resign_permit_model(model)
         await session.commit()
     before = await _receipt_count(provisioned["agent_wallet_id"])
     resp = await client.post(
@@ -576,10 +588,16 @@ async def test_action_permit_unregistered_tool_gets_permit_decision(
     )
     error = resp.json()["error"]
     assert error["code"] == -32003
-    assert error["message"] != "action_tool_binding_required"
-    assert error["message"].startswith("permit_")
-    # No idempotency record exists on the action path, so no receipt.
-    assert await _receipt_count(provisioned["agent_wallet_id"]) == before
+    # The re-signed action permit reaches the permit-scope check.
+    assert error["message"] == "permit_tool_not_allowed", error
+    # A valid action permit's scope denial mints one signed denial receipt.
+    assert await _receipt_count(provisioned["agent_wallet_id"]) == before + 1
+    receipt = error["data"]["receipt"]
+    await _assert_denial_receipt(
+        receipt,
+        permit_id=permit["permit_id"],
+        wallet_id=provisioned["agent_wallet_id"],
+    )
 
 
 async def _forge_signature(permit_id: str) -> None:
@@ -778,3 +796,62 @@ async def test_item3_same_gate_fires_first_for_registry_state(
         assert audit_error == first_gate
         assert not error.get("data", {}).get("receipt")
     assert results[True][2] == results[False][2]
+
+
+async def _call_with_key(client, provisioned, permit_id, tool, key):
+    body = _jsonrpc_body(
+        tool_name=tool,
+        wallet_id=provisioned["agent_wallet_id"],
+        request_id=f"retry-{uuid4()}",
+        permit_id=permit_id,
+        idempotency_key=key,
+    )
+    # Real dogfood signatures: write takes ``text``, count takes nothing.
+    body["params"]["arguments"] = {"text": "hello"} if tool == ALLOWED else {}
+    return await client.post(
+        "/mcp/messages", json=body, headers=provisioned["agent_headers"]
+    )
+
+
+@pytest.mark.anyio
+@pytest.mark.usefixtures("strict_trust_mode")
+@pytest.mark.parametrize("state", ["not_executable", "not_registered"])
+async def test_unresolved_tool_failure_is_not_replayed_after_fix(
+    client, clean_database, monkeypatch, state
+):
+    """DeepSeek 5662392959 #1: a registry-state failure is not the key's
+    final answer; a same-key retry runs once the tool works."""
+    provisioned = await provision_agent_wallet(client)
+    tool = ALLOWED if state == "not_executable" else UNKNOWN
+    permit = await create_tool_permit(
+        client,
+        wallet_id=provisioned["agent_wallet_id"],
+        key_id=provisioned["key_id"],
+        tool_name=tool,
+    )
+    registry = get_service_registry()
+    if state == "not_executable":
+        real = registry.get_local_func
+        monkeypatch.setattr(
+            registry,
+            "get_local_func",
+            lambda name: None if name == ALLOWED else real(name),
+        )
+    key = f"retry-idem-{uuid4()}"
+    first = (
+        await _call_with_key(client, provisioned, permit["permit_id"], tool, key)
+    ).json()
+    expected = "not executable" if state == "not_executable" else "not found"
+    assert expected in first["error"]["message"], first
+
+    if state == "not_executable":
+        monkeypatch.setattr(registry, "get_local_func", real)
+    else:
+        monkeypatch.setenv("ENABLE_DOGFOOD_SECOND_TOOL", "true")
+        get_settings.cache_clear()
+        sync_dogfood_tool_registration()
+    retry = (
+        await _call_with_key(client, provisioned, permit["permit_id"], tool, key)
+    ).json()
+    assert "error" not in retry, retry
+    assert "result" in retry

From 468a36fc4c2ab80c9aca58af639c5979f06861e1 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Mon, 26 Oct 2026 22:05:59 -0700
Subject: [PATCH 11/12] fix(mcp): keep terminal Tool-not-found record for
 permitless calls

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 gateway/app/routers/mcp.py         | 27 ++++++++++++++++++++-------
 gateway/docs/POLICY_ENFORCEMENT.md |  2 +-
 gateway/docs/failure-semantics.md  | 13 +++++++------
 3 files changed, 28 insertions(+), 14 deletions(-)

diff --git a/gateway/app/routers/mcp.py b/gateway/app/routers/mcp.py
index a6cf1742..ef8f197e 100644
--- a/gateway/app/routers/mcp.py
+++ b/gateway/app/routers/mcp.py
@@ -1578,13 +1578,26 @@ async def _execute_registered_tool_inner(
 
     if not service:
         reason = f"Tool not found: {tool_name}"
-        await _release_unresolved_tool_idempotency(
-            idem=idem,
-            idem_begin=idem_begin if idem_started else None,
-            wallet_id=wallet_id,
-            endpoint=idempotency_endpoint,
-            idempotency_key=idempotency_key,
-        )
+        if permit_id:
+            # A permitted call: registry state is not the key's final answer.
+            await _release_unresolved_tool_idempotency(
+                idem=idem,
+                idem_begin=idem_begin if idem_started else None,
+                wallet_id=wallet_id,
+                endpoint=idempotency_endpoint,
+                idempotency_key=idempotency_key,
+            )
+        else:
+            # No permit: unchanged contract, a terminal replayable record.
+            await _complete_governed_denial_idempotency(
+                idem=idem,
+                idem_started=idem_started,
+                wallet_id=wallet_id,
+                endpoint=idempotency_endpoint,
+                idempotency_key=idempotency_key,
+                reason=reason,
+                status_code=400,
+            )
         raise ValueError(reason)
 
     execution_backend = str(service.get("execution_backend") or "metadata_only")
diff --git a/gateway/docs/POLICY_ENFORCEMENT.md b/gateway/docs/POLICY_ENFORCEMENT.md
index 05428303..41e774ff 100644
--- a/gateway/docs/POLICY_ENFORCEMENT.md
+++ b/gateway/docs/POLICY_ENFORCEMENT.md
@@ -49,7 +49,7 @@ only reachable in local/dev/test, and it logs a loud startup warning.
 | 3 | **Quote validation**, when a `quote_id` is supplied | quote service | `quote_*` |
 | 4 | **Idempotency begin** (completed prior call replays its stored signed receipt) | `get_idempotency_service()` | replay short-circuit |
 | 5 | **Permit validation** (Layer B), including `allowed_tools`, decided with no registered price. A denial is identical for registered and unregistered tools and, with an idempotency key, mints a signed denial receipt with `credits_authorized = 0` | `PermitService.validate_for_action` | `permit_*` |
-| 6 | **Tool resolution + executability**. The idempotency claim from step 4 is released, so a same-key retry runs normally once the tool is registered or executable | service registry | `Tool not found` / `Tool not executable` |
+| 6 | **Tool resolution + executability**. On a permit-bearing call the idempotency claim from step 4 is released, so a same-key retry runs normally once the tool is registered or executable | service registry | `Tool not found` / `Tool not executable` |
 | 6a | **Wallet ownership — priced** (re-run with real cost, for the audit record); calls without a permit get `permit_required` here on a resolved tool | `evaluate_tool_invocation` / router | `wallet_access_denied` / `permit_required` |
 | 7 | **Wallet policy bundles** (Layer C) | `evaluate_wallet_policy` | policy reason |
 | 8 | **Human-approval gate** (Layer D) | `_require_human_approval` | blocks / `human_approval_unavailable` |
diff --git a/gateway/docs/failure-semantics.md b/gateway/docs/failure-semantics.md
index 913fddda..64c760ec 100644
--- a/gateway/docs/failure-semantics.md
+++ b/gateway/docs/failure-semantics.md
@@ -64,12 +64,13 @@ states this limit. What they leave behind differs by reason:
   receipt (`credits_authorized = 0`, `credits_charged = 0`) and completes a
   terminal idempotency record, so a same-key replay returns the same envelope
   (`test_denials_do_not_reveal_registration`).
-- `permit_not_found` on a governed call that carries an idempotency key
-  completes a terminal *idempotency record*; a same-key replay returns the
-  original error envelope.
-- `Tool not found` and `Tool not executable` record nothing: they depend on the
-  current registry, not on the request, so the governed idempotency claim is
-  released (`_release_unresolved_tool_idempotency` in `app/routers/mcp.py`).
+- An unknown tool (`Tool not found`) on a governed call *without* a permit, or
+  `permit_not_found`, on a call that carries an idempotency key completes a
+  terminal *idempotency record*; a same-key replay returns the original error
+  envelope.
+- On a call that carries a valid permit naming the tool, `Tool not found` and
+  `Tool not executable` record nothing: they depend on the current registry,
+  not on the request, so the governed idempotency claim is released (`_release_unresolved_tool_idempotency` in `app/routers/mcp.py`).
   A same-key retry executes normally once the tool is registered or executable
   (`test_unresolved_tool_failure_is_not_replayed_after_fix`).
 - `permit_required` on a resolved tool records nothing. A same-key retry returns

From fe7671c56854631d160efcfeca06ee4494a297d4 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Mon, 26 Oct 2026 22:57:49 -0700
Subject: [PATCH 12/12] fix(mcp): action-permit scope denial is
 registry-independent and replayable

Out-of-scope action permits now take the governed tool-permit path before
any registry-dependent action branch: same code, audit and signed receipt
for registered and unregistered tools, and the real idempotency key so a
same-key retry replays the first receipt. Docs updated (DeepSeek
5662554086, QA 5662827098).

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 gateway/app/routers/mcp.py                | 109 ++++++++++++-----
 gateway/docs/POLICY_ENFORCEMENT.md        |   5 +-
 gateway/docs/failure-semantics.md         |  42 +++++--
 gateway/tests/test_permit_unknown_tool.py | 141 ++++++++++++++++++++++
 4 files changed, 253 insertions(+), 44 deletions(-)

diff --git a/gateway/app/routers/mcp.py b/gateway/app/routers/mcp.py
index ef8f197e..6e4985a3 100644
--- a/gateway/app/routers/mcp.py
+++ b/gateway/app/routers/mcp.py
@@ -1398,40 +1398,66 @@ async def _execute_registered_tool_inner(
     if permit_id:
         async with get_session_factory()() as session:
             action_model = await session.get(PermitModel, permit_id)
-    is_action = action_model is not None and any(
-        getattr(action_model, field) is not None
-        for field in (
-            "action_contract_version",
-            "action_payload_hash",
-            "action_schema_id",
-            "action_schema_version",
-            "action_public_tool_id",
-            "action_upstream_binding_hash",
-        )
+    action_fields = (
+        [getattr(action_model, field) for field in _ACTION_PERMIT_FIELDS]
+        if action_model is not None
+        else []
+    )
+    is_action = any(value is not None for value in action_fields)
+    # An action permit that does not name this tool is a permit-scope denial,
+    # decided before any registry-dependent action branch (binding lookup,
+    # validate_action_request). It takes the same governed path as a tool
+    # permit below: idempotency-key and quote gates, idempotency begin, then
+    # _deny_tool_outside_permit, which validates the permit (signature first)
+    # and, with an idempotency key, mints one signed denial receipt that a
+    # same-key retry replays. Registered and unregistered tools are
+    # indistinguishable.
+    action_out_of_scope = (
+        is_action
+        # A partially populated action row (some action_* fields set, others
+        # NULL) is malformed storage and stays on the action branch below.
+        and all(value is not None for value in action_fields)
+        and permit_id is not None
+        and tool_name not in _permit_allowed_tools(action_model)
     )
-    if action_binding is not None or is_action:
+    if (action_binding is not None or is_action) and not action_out_of_scope:
+        if (
+            is_action
+            and permit_id
+            and not all(value is not None for value in action_fields)
+        ):
+            # A partial action row is validated (signature first) before any
+            # binding decision, whatever the registry holds; a complete
+            # in-scope one is validated by validate_action_request below. No
+            # idempotency record has been begun here and none is passed, so
+            # _deny_tool_outside_permit itself mints nothing and completes
+            # nothing. The ToolPermissionDenied it raises reaches the caller
+            # through _execute_registered_tool, which attaches an unbound
+            # receipt (idempotency_record_id = None, one per call, never
+            # replayed) via _action_request_denial_receipt when the row still
+            # verifies; a row that fails signature verification (e.g.
+            # permit_signature_invalid) gets no receipt at all.
+            await _deny_tool_outside_permit(
+                idem=idem,
+                idem_started=False,
+                permit_id=permit_id,
+                wallet_id=wallet_id,
+                auth=auth,
+                tool_name=tool_name,
+                arguments=arguments,
+                endpoint=endpoint,
+                idempotency_endpoint=idempotency_endpoint,
+                idempotency_key=None,
+                transport=transport,
+                request_id=request_id,
+                effective_request_payload=effective_request_payload,
+                effective_request_hash=effective_request_hash,
+            )
         if action_binding is None:
-            if not service and is_action and permit_id:
-                # An action permit naming an unregistered tool gets the same
-                # validated permit-scope decision as a tool permit. No
-                # idempotency record exists yet on this path, so the denial
-                # carries no receipt (see _deny_tool_outside_permit).
-                await _deny_tool_outside_permit(
-                    idem=idem,
-                    idem_started=False,
-                    permit_id=permit_id,
-                    wallet_id=wallet_id,
-                    auth=auth,
-                    tool_name=tool_name,
-                    arguments=arguments,
-                    endpoint=endpoint,
-                    idempotency_endpoint=idempotency_endpoint,
-                    idempotency_key=None,
-                    transport=transport,
-                    request_id=request_id,
-                    effective_request_payload=effective_request_payload,
-                    effective_request_hash=effective_request_hash,
-                )
+            # An in-scope action permit for a tool with no action binding
+            # (registered without one, or unregistered). Nothing has been
+            # begun or recorded; _execute_registered_tool attaches an unbound
+            # denial receipt when the permit row verifies.
             raise ToolPermissionDenied("action_tool_binding_required")
         if not is_action or action_model is None:
             raise ToolPermissionDenied("action_permit_required")
@@ -3847,6 +3873,25 @@ def _governed_error_payload(
     return payload
 
 
+_ACTION_PERMIT_FIELDS = (
+    "action_contract_version",
+    "action_payload_hash",
+    "action_schema_id",
+    "action_schema_version",
+    "action_public_tool_id",
+    "action_upstream_binding_hash",
+)
+
+
+def _permit_allowed_tools(model: Any) -> list[str]:
+    """The permit row's allowed_tools; malformed storage reads as empty."""
+    try:
+        value = json.loads(model.allowed_tools_json or "[]")
+    except (TypeError, ValueError):
+        return []
+    return [str(item) for item in value] if isinstance(value, list) else []
+
+
 async def _release_unresolved_tool_idempotency(
     *,
     idem: Any,
diff --git a/gateway/docs/POLICY_ENFORCEMENT.md b/gateway/docs/POLICY_ENFORCEMENT.md
index 41e774ff..145fbde7 100644
--- a/gateway/docs/POLICY_ENFORCEMENT.md
+++ b/gateway/docs/POLICY_ENFORCEMENT.md
@@ -48,8 +48,9 @@ only reachable in local/dev/test, and it logs a loud startup warning.
 | 2 | **Governed preconditions** (independent of tool registration) | router | `idempotency_key_required` |
 | 3 | **Quote validation**, when a `quote_id` is supplied | quote service | `quote_*` |
 | 4 | **Idempotency begin** (completed prior call replays its stored signed receipt) | `get_idempotency_service()` | replay short-circuit |
-| 5 | **Permit validation** (Layer B), including `allowed_tools`, decided with no registered price. A denial is identical for registered and unregistered tools and, with an idempotency key, mints a signed denial receipt with `credits_authorized = 0` | `PermitService.validate_for_action` | `permit_*` |
-| 6 | **Tool resolution + executability**. On a permit-bearing call the idempotency claim from step 4 is released, so a same-key retry runs normally once the tool is registered or executable | service registry | `Tool not found` / `Tool not executable` |
+| 5 | **Permit validation** (Layer B), including `allowed_tools`, decided with no registered price. A tool outside the permit's allowlist is denied `permit_tool_not_allowed` for tool permits and action permits alike, identically for registered and unregistered tools; with an idempotency key the denial mints a signed receipt with `credits_authorized = 0` bound to the record from step 4, and a same-key retry replays that receipt | `PermitService.validate_for_action` | `permit_*` |
+| 5a | **Action binding** for an in-scope *action* permit (decided ahead of step 4, so no idempotency record is begun). A tool with no action binding (registered without one, or unregistered) is denied `action_tool_binding_required`, never `Tool not found`, and the unbound denial receipt is minted per call. A bound tool is then checked by `validate_action_request` | service registry + `validate_action_request` | `action_*` |
+| 6 | **Tool resolution + executability**. On a tool-permit call the idempotency claim from step 4 is released, so a same-key retry runs normally once the tool is registered or executable | service registry | `Tool not found` / `Tool not executable` |
 | 6a | **Wallet ownership — priced** (re-run with real cost, for the audit record); calls without a permit get `permit_required` here on a resolved tool | `evaluate_tool_invocation` / router | `wallet_access_denied` / `permit_required` |
 | 7 | **Wallet policy bundles** (Layer C) | `evaluate_wallet_policy` | policy reason |
 | 8 | **Human-approval gate** (Layer D) | `_require_human_approval` | blocks / `human_approval_unavailable` |
diff --git a/gateway/docs/failure-semantics.md b/gateway/docs/failure-semantics.md
index 64c760ec..18e74e01 100644
--- a/gateway/docs/failure-semantics.md
+++ b/gateway/docs/failure-semantics.md
@@ -60,19 +60,41 @@ states this limit. What they leave behind differs by reason:
 
 - A governed call with a valid permit that names a tool outside the permit's
   `allowed_tools` gets 403 / `-32003` `permit_tool_not_allowed`, whether or not
-  the tool is registered. With an idempotency key it mints a signed denial
-  receipt (`credits_authorized = 0`, `credits_charged = 0`) and completes a
-  terminal idempotency record, so a same-key replay returns the same envelope
-  (`test_denials_do_not_reveal_registration`).
+  the tool is registered and whether the permit is a tool permit or an action
+  permit. With an idempotency key it mints a signed denial receipt
+  (`credits_authorized = 0`, `credits_charged = 0`) bound to the idempotency
+  record (`idempotency_record_id` set) and completes that record as terminal,
+  so a same-key replay returns the same envelope and the same receipt
+  (`test_denials_do_not_reveal_registration`,
+  `test_action_permit_denial_identical_for_registry_state`,
+  `test_action_permit_denial_replays_same_receipt_for_same_key`).
+- An action permit that *does* name the tool, used against a tool that has no
+  action binding (registered without one, or not registered), gets 403 /
+  `-32003` `action_tool_binding_required`, not `Tool not found`. No idempotency
+  record is begun. The response carries an unbound denial receipt
+  (`idempotency_record_id = null`) minted per call while the permit row still
+  verifies, so a same-key retry returns a fresh receipt rather than a replay.
 - An unknown tool (`Tool not found`) on a governed call *without* a permit, or
   `permit_not_found`, on a call that carries an idempotency key completes a
   terminal *idempotency record*; a same-key replay returns the original error
-  envelope.
-- On a call that carries a valid permit naming the tool, `Tool not found` and
-  `Tool not executable` record nothing: they depend on the current registry,
-  not on the request, so the governed idempotency claim is released (`_release_unresolved_tool_idempotency` in `app/routers/mcp.py`).
-  A same-key retry executes normally once the tool is registered or executable
-  (`test_unresolved_tool_failure_is_not_replayed_after_fix`).
+  envelope
+  (`test_strict_mode_replays_missing_tool_lookup_error_with_idempotency_key`).
+- A call *without* a permit to a registered but non-executable tool (`Tool not
+  executable`) records nothing: no governed record is begun for a resolved tool
+  without a permit, so the key is untouched
+  (`test_strict_mode_replays_non_executable_tool_error_with_idempotency_key`).
+- On a call that carries a valid *tool* permit naming the tool, `Tool not found`
+  and `Tool not executable` record nothing: they depend on the current
+  registry, not on the request, so the governed idempotency claim is released
+  (`_release_unresolved_tool_idempotency` in `app/routers/mcp.py`). A same-key
+  retry executes normally once the tool is registered or executable
+  (`test_unresolved_tool_failure_is_not_replayed_after_fix`). This holds when
+  the release runs. If the process dies between saving the claim and releasing
+  it, the row stays behind with `operation_kind` `unresolved` or `local`;
+  `reconcile_stuck_records` sweeps only `upstream_mcp` rows, and no TTL
+  applies to idempotency records, so a same-key retry answers
+  `idempotency_in_progress` until an operator clears the row. Retrying under
+  a new key is unaffected.
 - `permit_required` on a resolved tool records nothing. A same-key retry returns
   the same error while nothing has changed, and after minting a permit it
   executes normally
diff --git a/gateway/tests/test_permit_unknown_tool.py b/gateway/tests/test_permit_unknown_tool.py
index b6ca52a2..87215985 100644
--- a/gateway/tests/test_permit_unknown_tool.py
+++ b/gateway/tests/test_permit_unknown_tool.py
@@ -32,6 +32,7 @@
 from tests.test_mcp_trust_mode import (
     _assert_denial_audited,
     _assert_no_tool_debits,
+    _governed_idempotency_record_count,
     _jsonrpc_body,
     strict_trust_mode,  # noqa: F401 - fixture
 )
@@ -855,3 +856,143 @@ async def test_unresolved_tool_failure_is_not_replayed_after_fix(
     ).json()
     assert "error" not in retry, retry
     assert "result" in retry
+
+
+async def _action_permit_for_allowed_tool(client, tool_name: str = ALLOWED):
+    """A validly signed action permit whose allowlist names only tool_name."""
+    provisioned = await provision_agent_wallet(client)
+    permit = await create_tool_permit(
+        client,
+        wallet_id=provisioned["agent_wallet_id"],
+        key_id=provisioned["key_id"],
+        tool_name=tool_name,
+    )
+    async with get_session_factory()() as session:
+        model = await session.get(PermitModel, permit["permit_id"])
+        model.action_contract_version = 1
+        model.action_payload_hash = "a" * 64
+        model.action_schema_id = f"{tool_name}.args"
+        model.action_schema_version = "1"
+        model.action_public_tool_id = tool_name
+        model.action_upstream_binding_hash = "b" * 64
+        await resign_permit_model(model)
+        await session.commit()
+    return provisioned, permit
+
+
+async def _action_call(client, provisioned, permit_id, key, request_id=None):
+    resp = await client.post(
+        "/mcp/messages",
+        json=_jsonrpc_body(
+            tool_name=UNKNOWN,
+            wallet_id=provisioned["agent_wallet_id"],
+            request_id=request_id or f"action-{uuid4()}",
+            permit_id=permit_id,
+            idempotency_key=key,
+        ),
+        headers=provisioned["agent_headers"],
+    )
+    return resp.json()["error"]
+
+
+@pytest.mark.anyio
+@pytest.mark.usefixtures("strict_trust_mode")
+async def test_action_permit_denial_identical_for_registry_state(
+    client, clean_database, monkeypatch
+):
+    """DeepSeek 5662554086 #1: an action permit must not reveal registration."""
+    provisioned, permit = await _action_permit_for_allowed_tool(client)
+    wallet_id = provisioned["agent_wallet_id"]
+    seen = {}
+    for registered in (True, False):
+        monkeypatch.setenv("ENABLE_DOGFOOD_SECOND_TOOL", str(registered).lower())
+        get_settings.cache_clear()
+        sync_dogfood_tool_registration()
+        assert (get_service_registry().get_local(UNKNOWN) is not None) == registered
+        request_id = f"action-parity-{uuid4()}"
+        error = await _action_call(
+            client,
+            provisioned,
+            permit["permit_id"],
+            f"action-parity-idem-{uuid4()}",
+            request_id,
+        )
+        assert error["code"] == -32003, (registered, error)
+        assert error["message"] == "permit_tool_not_allowed", (registered, error)
+        receipt = error["data"]["receipt"]
+        await _assert_denial_receipt(
+            receipt, permit_id=permit["permit_id"], wallet_id=wallet_id
+        )
+        events = await list_audit_events(
+            event="mcp.invoke",
+            wallet_id=wallet_id,
+            tool=UNKNOWN,
+            request_id=request_id,
+            ok=False,
+            limit=1,
+        )
+        assert len(events) == 1
+        metadata = dict(events[0].metadata)
+        for volatile in ("idempotency_key", "request_hash", "request_id"):
+            metadata.pop(volatile, None)
+        seen[registered] = (
+            events[0].error,
+            metadata,
+            {
+                k: v
+                for k, v in receipt.items()
+                if k not in _VOLATILE_RECEIPT_FIELDS | {"request_hash"}
+            },
+        )
+    assert seen[True] == seen[False]
+
+
+@pytest.mark.anyio
+@pytest.mark.usefixtures("strict_trust_mode")
+async def test_action_permit_denial_replays_same_receipt_for_same_key(
+    client, clean_database
+):
+    """DeepSeek 5662554086 #2: same key, same signed denial receipt."""
+    provisioned, permit = await _action_permit_for_allowed_tool(client)
+    key = f"action-replay-idem-{uuid4()}"
+    first = await _action_call(client, provisioned, permit["permit_id"], key)
+    before = await _receipt_count(provisioned["agent_wallet_id"])
+    second = await _action_call(client, provisioned, permit["permit_id"], key)
+    assert first["message"] == second["message"] == "permit_tool_not_allowed"
+    assert first["data"]["receipt"]["idempotency_record_id"] is not None
+    assert second["data"]["receipt"] == first["data"]["receipt"]
+    assert await _receipt_count(provisioned["agent_wallet_id"]) == before
+
+
+@pytest.mark.anyio
+@pytest.mark.usefixtures("strict_trust_mode")
+async def test_in_scope_action_permit_without_binding_is_binding_denial(
+    client, clean_database, monkeypatch
+):
+    """docs/failure-semantics.md: an in-scope action permit for a tool with
+    no action binding is action_tool_binding_required, not Tool not found,
+    for either registry state, with no idempotency record begun."""
+    provisioned, permit = await _action_permit_for_allowed_tool(client, UNKNOWN)
+    wallet_id = provisioned["agent_wallet_id"]
+    for registered in (True, False):
+        monkeypatch.setenv("ENABLE_DOGFOOD_SECOND_TOOL", str(registered).lower())
+        get_settings.cache_clear()
+        sync_dogfood_tool_registration()
+        service = get_service_registry().get_local(UNKNOWN)
+        assert (service is not None) == registered
+        if registered:
+            assert get_service_registry().get_action_binding(service) is None
+        key = f"action-binding-idem-{uuid4()}"
+        error = await _action_call(client, provisioned, permit["permit_id"], key)
+        assert error["code"] == -32003, (registered, error)
+        assert error["message"] == "action_tool_binding_required", (registered, error)
+        receipt = error["data"]["receipt"]
+        assert receipt["outcome"] == "denied"
+        assert receipt["permit_id"] == permit["permit_id"]
+        assert receipt["idempotency_record_id"] is None
+        assert (
+            await _governed_idempotency_record_count(
+                wallet_id=wallet_id, idempotency_key=key
+            )
+            == 0
+        )