From 63005dd12bc50e927ef46e000ef0fdbccb714843 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Sat, 24 Oct 2026 20:59:54 -0700
Subject: [PATCH 1/5] fix(gateway): address approval gate follow-ups from #216
 review

- CLI refuses to send the admin key unless the URL is https or http loopback
- single clock source; expiry is now >= deadline, matching the SQL guard
- reuse live approval per idempotency key; consume refuses if a sibling with the same key was already consumed
- document why gate receipts (incl. consumed) share one default sink

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 gateway/app/services/approval_gate.py   | 306 ++++++++++++++++++++----
 gateway/scripts/amw_approvals.py        |  53 ++++
 gateway/tests/test_amw_approvals_cli.py | 179 ++++++++++++++
 gateway/tests/test_approval_gate.py     | 262 +++++++++++++++++++-
 4 files changed, 758 insertions(+), 42 deletions(-)

diff --git a/gateway/app/services/approval_gate.py b/gateway/app/services/approval_gate.py
index 9a122e00..6daa1061 100644
--- a/gateway/app/services/approval_gate.py
+++ b/gateway/app/services/approval_gate.py
@@ -41,11 +41,11 @@
 from datetime import datetime, timezone
 from typing import Any, Callable, Literal, TypeVar, cast
 
-from sqlalchemy import select, update
+from sqlalchemy import and_, or_, select, update
 from sqlalchemy.sql.elements import ColumnElement
 from sqlmodel import col
 
-from app.core.time import to_naive_utc, utc_now
+from app.core.time import to_naive_utc
 from app.db.approval_gate_models import ApprovalGateModel
 from app.db.database import get_session_factory
 
@@ -162,12 +162,22 @@ class ApprovalRecord:
 class ApprovalStore:
     """Interface for pending approvals so a DB-backed store can replace memory."""
 
-    def save(self, record: ApprovalRecord) -> None:
+    def save(self, record: ApprovalRecord, *, now: float | None = None) -> None:
         raise NotImplementedError
 
     def get(self, nonce: str) -> ApprovalRecord | None:
         raise NotImplementedError
 
+    def find_active_by_key(
+        self, idempotency_key: str, *, now: float
+    ) -> ApprovalRecord | None:
+        """Live record for a key: pending, or approved and unconsumed.
+
+        Only unexpired records (``deadline`` after ``now``) count; None when
+        no such record exists.
+        """
+        raise NotImplementedError
+
     def list_pending(self, *, now: float | None = None) -> list[ApprovalRecord]:
         """Pending records, excluding ones past deadline when now is given."""
         raise NotImplementedError
@@ -184,7 +194,9 @@ def decide(
     def consume(self, nonce: str, *, now: float) -> ApprovalRecord | None:
         """Single-use: first call on approved marks consumed; later calls None.
 
-        Expired approvals (``deadline`` past ``now``) never consume.
+        Expired approvals (``now`` at or past ``deadline``) never consume,
+        and neither does a nonce whose idempotency key already spent on
+        another live record.
         """
         raise NotImplementedError
 
@@ -192,7 +204,7 @@ def mark_expired(self, nonce: str, *, now: float) -> ApprovalRecord | None:
         """Flip pending to expired; None when missing."""
         raise NotImplementedError
 
-    async def asave(self, record: ApprovalRecord) -> None:
+    async def asave(self, record: ApprovalRecord, *, now: float | None = None) -> None:
         """Async ``save`` for handlers already on the app event loop."""
         raise NotImplementedError
 
@@ -200,6 +212,12 @@ async def aget(self, nonce: str) -> ApprovalRecord | None:
         """Async ``get`` for handlers already on the app event loop."""
         raise NotImplementedError
 
+    async def afind_active_by_key(
+        self, idempotency_key: str, *, now: float
+    ) -> ApprovalRecord | None:
+        """Async ``find_active_by_key`` for handlers already on the app loop."""
+        raise NotImplementedError
+
     async def alist_pending(self, *, now: float | None = None) -> list[ApprovalRecord]:
         """Async ``list_pending`` for handlers already on the app event loop."""
         raise NotImplementedError
@@ -226,7 +244,8 @@ def __init__(self) -> None:
         self._lock = threading.Lock()
         self._records: dict[str, ApprovalRecord] = {}
 
-    def save(self, record: ApprovalRecord) -> None:
+    def save(self, record: ApprovalRecord, *, now: float | None = None) -> None:
+        del now  # in-memory records carry their own timestamps; nothing to stamp.
         with self._lock:
             self._records[record.nonce] = replace(record)
 
@@ -235,6 +254,24 @@ def get(self, nonce: str) -> ApprovalRecord | None:
             record = self._records.get(nonce)
             return replace(record) if record else None
 
+    def find_active_by_key(
+        self, idempotency_key: str, *, now: float
+    ) -> ApprovalRecord | None:
+        with self._lock:
+            candidates = [
+                record
+                for record in self._records.values()
+                if record.idempotency_key == idempotency_key
+                and record.deadline > now
+                and (
+                    record.status == "pending"
+                    or (record.status == "approved" and not record.consumed)
+                )
+            ]
+            if not candidates:
+                return None
+            return replace(min(candidates, key=lambda r: (r.created_at, r.nonce)))
+
     def list_pending(self, *, now: float | None = None) -> list[ApprovalRecord]:
         with self._lock:
             return [
@@ -267,6 +304,14 @@ def consume(self, nonce: str, *, now: float) -> ApprovalRecord | None:
                 return None
             if record.deadline <= now:
                 return None
+            for other in self._records.values():
+                if (
+                    other.nonce != nonce
+                    and other.idempotency_key == record.idempotency_key
+                    and other.consumed
+                    and other.deadline > now
+                ):
+                    return None
             updated = replace(record, consumed=True)
             self._records[nonce] = updated
             return replace(updated)
@@ -281,12 +326,17 @@ def mark_expired(self, nonce: str, *, now: float) -> ApprovalRecord | None:
                 self._records[nonce] = record
             return replace(record)
 
-    async def asave(self, record: ApprovalRecord) -> None:
-        self.save(record)
+    async def asave(self, record: ApprovalRecord, *, now: float | None = None) -> None:
+        self.save(record, now=now)
 
     async def aget(self, nonce: str) -> ApprovalRecord | None:
         return self.get(nonce)
 
+    async def afind_active_by_key(
+        self, idempotency_key: str, *, now: float
+    ) -> ApprovalRecord | None:
+        return self.find_active_by_key(idempotency_key, now=now)
+
     async def alist_pending(self, *, now: float | None = None) -> list[ApprovalRecord]:
         return self.list_pending(now=now)
 
@@ -336,7 +386,23 @@ def _from_db_time(value: datetime) -> float:
     return value.replace(tzinfo=timezone.utc).timestamp()
 
 
-def _model_from_record(record: ApprovalRecord) -> ApprovalGateModel:
+def _model_from_record(
+    record: ApprovalRecord, *, now: float | None = None
+) -> ApprovalGateModel:
+    if record.consumed:
+        # The consume UPDATE stamps the caller's now; a consumed record
+        # reaching save() (merge) falls back to the closest record timestamp
+        # so no wall-clock read sneaks in beside the gate clock.
+        stamp = now
+        if stamp is None:
+            stamp = (
+                record.decided_at
+                if record.decided_at is not None
+                else record.created_at
+            )
+        consumed_at = _to_db_time(stamp)
+    else:
+        consumed_at = None
     return ApprovalGateModel(
         nonce=record.nonce,
         idempotency_key=record.idempotency_key,
@@ -350,7 +416,7 @@ def _model_from_record(record: ApprovalRecord) -> ApprovalGateModel:
         decided_at=(
             _to_db_time(record.decided_at) if record.decided_at is not None else None
         ),
-        consumed_at=utc_now() if record.consumed else None,
+        consumed_at=consumed_at,
     )
 
 
@@ -384,13 +450,13 @@ class DbApprovalStore(ApprovalStore):
     retries can replay it, and an expired approval can never spend.
     """
 
-    def save(self, record: ApprovalRecord) -> None:
-        _run_sync(self.asave(record))
+    def save(self, record: ApprovalRecord, *, now: float | None = None) -> None:
+        _run_sync(self.asave(record, now=now))
 
-    async def asave(self, record: ApprovalRecord) -> None:
+    async def asave(self, record: ApprovalRecord, *, now: float | None = None) -> None:
         factory = get_session_factory()
         async with factory() as session:
-            await session.merge(_model_from_record(record))
+            await session.merge(_model_from_record(record, now=now))
             await session.commit()
 
     def get(self, nonce: str) -> ApprovalRecord | None:
@@ -407,6 +473,56 @@ async def aget(self, nonce: str) -> ApprovalRecord | None:
             model = result.scalar_one_or_none()
             return _record_from_model(model) if model is not None else None
 
+    def find_active_by_key(
+        self, idempotency_key: str, *, now: float
+    ) -> ApprovalRecord | None:
+        return _run_sync(self.afind_active_by_key(idempotency_key, now=now))
+
+    async def afind_active_by_key(
+        self, idempotency_key: str, *, now: float
+    ) -> ApprovalRecord | None:
+        factory = get_session_factory()
+        async with factory() as session:
+            result = await session.execute(
+                select(ApprovalGateModel)
+                .where(
+                    cast(
+                        ColumnElement[bool],
+                        ApprovalGateModel.idempotency_key == idempotency_key,
+                    ),
+                    cast(
+                        ColumnElement[bool],
+                        ApprovalGateModel.deadline > _to_db_time(now),
+                    ),
+                    cast(
+                        ColumnElement[bool],
+                        or_(
+                            cast(
+                                ColumnElement[bool],
+                                ApprovalGateModel.status == "pending",
+                            ),
+                            and_(
+                                cast(
+                                    ColumnElement[bool],
+                                    ApprovalGateModel.status == "approved",
+                                ),
+                                cast(
+                                    ColumnElement[bool],
+                                    cast(Any, ApprovalGateModel.consumed_at).is_(None),
+                                ),
+                            ),
+                        ),
+                    ),
+                )
+                .order_by(
+                    col(ApprovalGateModel.created_at).asc(),
+                    col(ApprovalGateModel.nonce).asc(),
+                )
+                .limit(1)
+            )
+            model = result.scalar_one_or_none()
+            return _record_from_model(model) if model is not None else None
+
     def list_pending(self, *, now: float | None = None) -> list[ApprovalRecord]:
         return _run_sync(self.alist_pending(now=now))
 
@@ -493,9 +609,42 @@ async def aconsume(self, nonce: str, *, now: float) -> ApprovalRecord | None:
         one caller (one retry, one transport) wins, a restart cannot
         resurrect a spent approval, and an approval whose deadline passed
         between the gate's expiry check and this UPDATE still cannot spend.
+        The ``NOT EXISTS`` sibling check extends the single spend to the
+        whole idempotency key: a second nonce approved for the same key
+        cannot spend while the first spend's deadline has not passed.
         """
         factory = get_session_factory()
         async with factory() as session:
+            # The key never changes after insert, so reading it first is safe;
+            # the NOT EXISTS check itself still runs inside the UPDATE below.
+            key_result = await session.execute(
+                select(ApprovalGateModel).where(
+                    cast(ColumnElement[bool], ApprovalGateModel.nonce == nonce)
+                )
+            )
+            key_model = key_result.scalar_one_or_none()
+            if key_model is None:
+                return None
+            key = key_model.idempotency_key
+            sibling_consumed = (
+                select(1)
+                .where(
+                    cast(
+                        ColumnElement[bool],
+                        ApprovalGateModel.idempotency_key == key,
+                    ),
+                    cast(ColumnElement[bool], ApprovalGateModel.nonce != nonce),
+                    cast(
+                        ColumnElement[bool],
+                        cast(Any, ApprovalGateModel.consumed_at).is_not(None),
+                    ),
+                    cast(
+                        ColumnElement[bool],
+                        ApprovalGateModel.deadline > _to_db_time(now),
+                    ),
+                )
+                .exists()
+            )
             updated = await session.execute(
                 update(ApprovalGateModel)
                 .where(
@@ -512,8 +661,9 @@ async def aconsume(self, nonce: str, *, now: float) -> ApprovalRecord | None:
                         ColumnElement[bool],
                         ApprovalGateModel.deadline > _to_db_time(now),
                     ),
+                    cast(ColumnElement[bool], ~sibling_consumed),
                 )
-                .values(consumed_at=utc_now())
+                .values(consumed_at=_to_db_time(now))
             )
             await session.commit()
             if cast(Any, updated).rowcount != 1:
@@ -564,6 +714,16 @@ async def amark_expired(self, nonce: str, *, now: float) -> ApprovalRecord | Non
 
 
 def _log_receipt_sink(event: str, record: dict[str, Any]) -> None:
+    """Default gate receipt route: one structured log line per event.
+
+    Deliberately not the durable ReceiptService pipeline: create_receipt is
+    an async database write that needs permit/wallet/ledger identity the gate
+    does not have, and the sync evaluate() path cannot safely touch the
+    loop-bound engine. Every gate event (denied, timeout, tampered, ...,
+    consumed) flows through this single sink, so none is silently more or
+    less durable than the others; callers that need durability inject a
+    receipt_sink.
+    """
     logger.info(
         "approval_receipt event=%s record=%s", event, json.dumps(record, sort_keys=True)
     )
@@ -578,6 +738,33 @@ class ApprovalDecision:
     nonce: str | None = None
 
 
+def _seal_pause_state(
+    secret: bytes,
+    *,
+    args_hash: str,
+    caller: str,
+    tool: str,
+    session_id: str,
+    idempotency_key: str,
+    deadline: float,
+    nonce: str,
+) -> str:
+    """Seal the retry state for one pause, fresh or reused."""
+    return _seal_state(
+        secret,
+        {
+            "v": STATE_VERSION,
+            "argsHash": args_hash,
+            "principal": caller,
+            "tool": tool,
+            "sessionId": session_id,
+            "idempotencyKey": idempotency_key,
+            "deadline": deadline,
+            "nonce": nonce,
+        },
+    )
+
+
 def _pause_payload(tool: str, nonce: str, request_state: str) -> dict[str, Any]:
     """Build the SEP-2322 ``input_required`` pause payload.
 
@@ -814,10 +1001,13 @@ def _check_record(
 
     @staticmethod
     def _is_expired(now: float, sealed_deadline: Any, record: ApprovalRecord) -> bool:
+        # Expired when now >= deadline, matching the SQL conditions
+        # (deadline > now admits only the live). The in-memory store already
+        # uses deadline <= now / deadline > now, so all three agree.
         return (
             not isinstance(sealed_deadline, (int, float))
-            or now > sealed_deadline
-            or now > record.deadline
+            or now >= sealed_deadline
+            or now >= record.deadline
             or record.status == "expired"
         )
 
@@ -842,6 +1032,25 @@ def _pause_new_call(
             return ApprovalDecision(action="REJECT", reason="unsupported_client")
         secret = self._require_secret()
         now = self.clock()
+        existing = self.store.find_active_by_key(idempotency_key, now=now)
+        if existing is not None:
+            # Same logical call already paused: reissue its pause instead of
+            # minting a second spendable approval for one idempotency key.
+            state = _seal_pause_state(
+                secret,
+                args_hash=args_hash,
+                caller=caller,
+                tool=tool,
+                session_id=session_id,
+                idempotency_key=idempotency_key,
+                deadline=existing.deadline,
+                nonce=existing.nonce,
+            )
+            return ApprovalDecision(
+                action="PAUSE",
+                nonce=existing.nonce,
+                payload=_pause_payload(tool, existing.nonce, state),
+            )
         nonce = secrets.token_urlsafe(16)
         deadline = now + self._ttl
         self.store.save(
@@ -854,20 +1063,18 @@ def _pause_new_call(
                 session_id=session_id,
                 created_at=now,
                 deadline=deadline,
-            )
+            ),
+            now=now,
         )
-        state = _seal_state(
+        state = _seal_pause_state(
             secret,
-            {
-                "v": STATE_VERSION,
-                "argsHash": args_hash,
-                "principal": caller,
-                "tool": tool,
-                "sessionId": session_id,
-                "idempotencyKey": idempotency_key,
-                "deadline": deadline,
-                "nonce": nonce,
-            },
+            args_hash=args_hash,
+            caller=caller,
+            tool=tool,
+            session_id=session_id,
+            idempotency_key=idempotency_key,
+            deadline=deadline,
+            nonce=nonce,
         )
         return ApprovalDecision(
             action="PAUSE",
@@ -896,6 +1103,25 @@ async def _apause_new_call(
             return ApprovalDecision(action="REJECT", reason="unsupported_client")
         secret = self._require_secret()
         now = self.clock()
+        existing = await self.store.afind_active_by_key(idempotency_key, now=now)
+        if existing is not None:
+            # Same logical call already paused: reissue its pause instead of
+            # minting a second spendable approval for one idempotency key.
+            state = _seal_pause_state(
+                secret,
+                args_hash=args_hash,
+                caller=caller,
+                tool=tool,
+                session_id=session_id,
+                idempotency_key=idempotency_key,
+                deadline=existing.deadline,
+                nonce=existing.nonce,
+            )
+            return ApprovalDecision(
+                action="PAUSE",
+                nonce=existing.nonce,
+                payload=_pause_payload(tool, existing.nonce, state),
+            )
         nonce = secrets.token_urlsafe(16)
         deadline = now + self._ttl
         await self.store.asave(
@@ -908,20 +1134,18 @@ async def _apause_new_call(
                 session_id=session_id,
                 created_at=now,
                 deadline=deadline,
-            )
+            ),
+            now=now,
         )
-        state = _seal_state(
+        state = _seal_pause_state(
             secret,
-            {
-                "v": STATE_VERSION,
-                "argsHash": args_hash,
-                "principal": caller,
-                "tool": tool,
-                "sessionId": session_id,
-                "idempotencyKey": idempotency_key,
-                "deadline": deadline,
-                "nonce": nonce,
-            },
+            args_hash=args_hash,
+            caller=caller,
+            tool=tool,
+            session_id=session_id,
+            idempotency_key=idempotency_key,
+            deadline=deadline,
+            nonce=nonce,
         )
         return ApprovalDecision(
             action="PAUSE",
diff --git a/gateway/scripts/amw_approvals.py b/gateway/scripts/amw_approvals.py
index 4ee597fc..1a743b47 100644
--- a/gateway/scripts/amw_approvals.py
+++ b/gateway/scripts/amw_approvals.py
@@ -15,9 +15,11 @@
 from __future__ import annotations
 
 import argparse
+import ipaddress
 import os
 import sys
 from datetime import datetime, timezone
+from urllib.parse import urlparse
 
 import httpx
 
@@ -29,6 +31,43 @@
 API_KEY_ENV = "VALID_API_KEYS"
 API_KEY_HEADER = "X-API-Key"
 
+_LOOPBACK_V4 = ipaddress.ip_network("127.0.0.0/8")
+_LOOPBACK_V6 = ipaddress.ip_address("::1")
+
+
+def _refusal_reason(base_url: str) -> str | None:
+    """Why the API key must not be sent to base_url; None when allowed.
+
+    Allowed: any https URL, or http only on loopback (localhost, 127.0.0.0/8,
+    ::1). The host comes from parsed.hostname, so userinfo tricks such as
+    http://127.0.0.1@evil resolve to the real host (evil), and look-alikes
+    such as localhost.example or 127.0.0.1.nip.io fail both the exact
+    localhost match and the IP parse. Unknown schemes and empty hosts refuse.
+    """
+    try:
+        parsed = urlparse(base_url)
+        host = parsed.hostname or ""
+    except ValueError:
+        return "unparseable URL"
+    scheme = parsed.scheme.lower()
+    if scheme not in ("http", "https"):
+        if parsed.scheme:
+            return f"unsupported scheme {parsed.scheme!r}"
+        return "missing URL scheme"
+    if not host:
+        return "missing host"
+    if scheme == "https":
+        return None
+    if host == "localhost":
+        return None
+    try:
+        addr = ipaddress.ip_address(host)
+    except ValueError:
+        return f"non-loopback http host {host!r}"
+    if addr in _LOOPBACK_V4 or addr == _LOOPBACK_V6:
+        return None
+    return f"non-loopback http host {host!r}"
+
 
 def _read_api_key() -> str:
     """First key from API_KEY_ENV; empty when unset or blank."""
@@ -39,6 +78,9 @@ def _read_api_key() -> str:
 
 
 def _client(base_url: str, api_key: str) -> httpx.Client:
+    reason = _refusal_reason(base_url)
+    if reason is not None:
+        raise ValueError(f"refusing to send the API key: {reason}")
     return httpx.Client(
         base_url=base_url, timeout=10.0, headers={API_KEY_HEADER: api_key}
     )
@@ -120,6 +162,17 @@ def build_parser() -> argparse.ArgumentParser:
 
 def main(argv: list[str] | None = None) -> int:
     args = build_parser().parse_args(argv)
+    # Validate the destination before the key is read, so a refusal never
+    # attaches the bootstrap admin key to a request.
+    reason = _refusal_reason(args.base_url)
+    if reason is not None:
+        print(
+            "error: refusing to send the API key: "
+            f"{reason}; use an https URL or http on loopback "
+            "(localhost, 127.0.0.0/8, ::1)",
+            file=sys.stderr,
+        )
+        return 1
     api_key = _read_api_key()
     if not api_key:
         print(
diff --git a/gateway/tests/test_amw_approvals_cli.py b/gateway/tests/test_amw_approvals_cli.py
index a8e908da..bce1d49f 100644
--- a/gateway/tests/test_amw_approvals_cli.py
+++ b/gateway/tests/test_amw_approvals_cli.py
@@ -8,6 +8,7 @@
 
 import importlib.util
 import secrets
+import uuid
 from pathlib import Path
 
 import pytest
@@ -116,3 +117,181 @@ def test_unreachable_gateway_error_hides_key(
     out = capsys.readouterr()
     assert PLACEHOLDER_KEY not in out.out
     assert PLACEHOLDER_KEY not in out.err
+
+
+def _random_port() -> int:
+    return 1024 + secrets.randbelow(65535 - 1024 + 1)
+
+
+def _random_10_host() -> str:
+    return (
+        f"10.{secrets.randbelow(256)}.{secrets.randbelow(256)}"
+        f".{1 + secrets.randbelow(254)}"
+    )
+
+
+def _random_192_168_host() -> str:
+    return f"192.168.{secrets.randbelow(256)}.{1 + secrets.randbelow(254)}"
+
+
+def _random_invalid_host() -> str:
+    return f"{uuid.uuid4().hex}.invalid"
+
+
+def _random_127_host() -> str:
+    return (
+        f"127.{secrets.randbelow(256)}.{secrets.randbelow(256)}"
+        f".{secrets.randbelow(256)}"
+    )
+
+
+def _allowed_url(case: str) -> str:
+    # Deterministic labels only; random hosts/ports are built in the body so
+    # xdist workers collecting independently see identical test ids.
+    if case == "https-any-host":
+        return f"https://{_random_invalid_host()}:{_random_port()}"
+    if case == "https-no-port":
+        return f"https://{_random_invalid_host()}"
+    if case == "http-localhost":
+        return f"http://localhost:{_random_port()}"
+    if case == "http-127-1":
+        return f"http://127.0.0.1:{_random_port()}"
+    if case == "http-127-range":
+        return f"http://{_random_127_host()}:{_random_port()}"
+    if case == "http-ipv6-loopback":
+        return f"http://[::1]:{_random_port()}"
+    if case == "http-uppercase-localhost":
+        return f"HTTP://localhost:{_random_port()}"
+    raise AssertionError(f"unknown allowed case: {case}")
+
+
+def _refused_url(case: str) -> str:
+    if case == "http-rfc1918-10":
+        return f"http://{_random_10_host()}:{_random_port()}"
+    if case == "http-rfc1918-192-168":
+        return f"http://{_random_192_168_host()}:{_random_port()}"
+    if case == "http-public-hostname":
+        return f"http://{_random_invalid_host()}:{_random_port()}"
+    if case == "http-userinfo-trick":
+        return f"http://127.0.0.1@{_random_invalid_host()}:{_random_port()}"
+    if case == "http-localhost-userinfo-trick":
+        return f"http://localhost@{_random_invalid_host()}:{_random_port()}/"
+    if case == "http-localhost-lookalike":
+        return f"http://localhost.{uuid.uuid4().hex}.invalid:{_random_port()}"
+    if case == "http-ip-lookalike":
+        return f"http://127.0.0.1.{uuid.uuid4().hex}.invalid:{_random_port()}"
+    if case == "ftp-loopback":
+        return f"ftp://localhost:{_random_port()}/"
+    if case == "file-scheme":
+        return "file:///etc/hosts"
+    if case == "missing-host":
+        return "http:///no-host"
+    if case == "empty":
+        return ""
+    if case == "no-scheme":
+        return f"localhost:{_random_port()}"
+    raise AssertionError(f"unknown refused case: {case}")
+
+
+@pytest.mark.parametrize(
+    "case",
+    [
+        "https-any-host",
+        "https-no-port",
+        "http-localhost",
+        "http-127-1",
+        "http-127-range",
+        "http-ipv6-loopback",
+        "http-uppercase-localhost",
+    ],
+)
+def test_allowed_base_url_sends_request(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], case: str
+) -> None:
+    sent: list[str] = []
+
+    class _RecordingClient:
+        def __init__(self, *args: object, **kwargs: object) -> None:
+            pass
+
+        def get(self, path: str) -> _StubResponse:
+            sent.append(path)
+            return _StubResponse({"approvals": []})
+
+    monkeypatch.setattr(amw_approvals.httpx, "Client", _RecordingClient)
+    monkeypatch.setenv("VALID_API_KEYS", PLACEHOLDER_KEY)
+    assert amw_approvals.main(["--base-url", _allowed_url(case), "list"]) == 0
+    assert sent == ["/v1/approvals/pending"]
+    out = capsys.readouterr()
+    assert PLACEHOLDER_KEY not in out.out
+    assert PLACEHOLDER_KEY not in out.err
+
+
+class _ExplodingClient:
+    def __init__(self, *args: object, **kwargs: object) -> None:
+        raise AssertionError("refused URL must not open an HTTP client")
+
+
+@pytest.mark.parametrize(
+    "case",
+    [
+        "http-rfc1918-10",
+        "http-rfc1918-192-168",
+        "http-public-hostname",
+        "http-userinfo-trick",
+        "http-localhost-userinfo-trick",
+        "http-localhost-lookalike",
+        "http-ip-lookalike",
+        "ftp-loopback",
+        "file-scheme",
+        "missing-host",
+        "empty",
+        "no-scheme",
+    ],
+)
+def test_refused_base_url_makes_no_request(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], case: str
+) -> None:
+    monkeypatch.setattr(amw_approvals.httpx, "Client", _ExplodingClient)
+    monkeypatch.setenv("VALID_API_KEYS", PLACEHOLDER_KEY)
+    base_url = _refused_url(case)
+    nonce = uuid.uuid4().hex
+    for argv in (["list"], ["approve", nonce], ["deny", nonce]):
+        assert amw_approvals.main(["--base-url", base_url, *argv]) == 1
+        err = capsys.readouterr().err
+        assert "refusing to send the API key" in err
+        assert PLACEHOLDER_KEY not in err
+
+
+def test_refusal_happens_before_key_check(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    monkeypatch.setattr(amw_approvals.httpx, "Client", _ExplodingClient)
+    monkeypatch.delenv("VALID_API_KEYS", raising=False)
+    url = f"http://{_random_10_host()}:{_random_port()}"
+    assert amw_approvals.main(["--base-url", url, "list"]) == 1
+    err = capsys.readouterr().err
+    assert "refusing to send the API key" in err
+    assert "VALID_API_KEYS" not in err
+
+
+def test_refused_env_base_url_makes_no_request(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    monkeypatch.setattr(amw_approvals.httpx, "Client", _ExplodingClient)
+    monkeypatch.setenv("VALID_API_KEYS", PLACEHOLDER_KEY)
+    monkeypatch.setenv(
+        amw_approvals.BASE_URL_ENV,
+        f"http://{_random_invalid_host()}:{_random_port()}",
+    )
+    assert amw_approvals.main(["list"]) == 1
+    out = capsys.readouterr()
+    assert "refusing to send the API key" in out.err
+    assert PLACEHOLDER_KEY not in out.out
+    assert PLACEHOLDER_KEY not in out.err
+
+
+def test_client_refuses_disallowed_url_directly() -> None:
+    url = f"http://{_random_10_host()}:{_random_port()}"
+    with pytest.raises(ValueError, match="refusing to send the API key"):
+        amw_approvals._client(url, PLACEHOLDER_KEY)
diff --git a/gateway/tests/test_approval_gate.py b/gateway/tests/test_approval_gate.py
index 7a45b990..e26b57a5 100644
--- a/gateway/tests/test_approval_gate.py
+++ b/gateway/tests/test_approval_gate.py
@@ -7,6 +7,7 @@
 
 import asyncio
 import json
+import logging
 import secrets
 import sys
 import uuid
@@ -26,7 +27,7 @@
 
 import pytest  # noqa: E402
 from goose_amw import canonical as goose_canonical  # noqa: E402
-from sqlalchemy import delete  # noqa: E402
+from sqlalchemy import delete, select  # noqa: E402
 
 import app.services.approval_gate as gate_module  # noqa: E402
 from app.db.approval_gate_models import ApprovalGateModel  # noqa: E402
@@ -731,3 +732,262 @@ async def test_db_alist_pending_excludes_expired(
     assert second.nonce is not None
     pending = await DbApprovalStore().alist_pending(now=clock())
     assert [record.nonce for record in pending] == [second.nonce]
+
+
+def test_exact_deadline_pending_refuses_gate_and_store(
+    gate: ApprovalGate, clock: _Clock, receipts: list[tuple[str, dict[str, Any]]]
+) -> None:
+    """At now == deadline the record is expired for both gate and store."""
+    decision = _pause(gate)
+    assert decision.nonce is not None
+    assert decision.payload is not None
+    record = gate.store.get(decision.nonce)
+    assert record is not None
+    clock.now = record.deadline
+    assert gate.store.decide(decision.nonce, True, now=clock()) is None
+    rejected = _retry(gate, decision.payload["requestState"], decision.nonce)
+    assert rejected.action == "REJECT"
+    assert rejected.reason == "timeout"
+    assert gate.store.get(decision.nonce) is not None
+    assert gate.store.get(decision.nonce).status == "expired"  # type: ignore[union-attr]
+    assert [event for event, _ in receipts] == ["timeout"]
+
+
+def test_exact_deadline_approved_refuses_gate_and_store(
+    gate: ApprovalGate, clock: _Clock, receipts: list[tuple[str, dict[str, Any]]]
+) -> None:
+    """An approval cannot spend at its exact deadline instant."""
+    decision = _pause(gate)
+    assert decision.nonce is not None
+    assert decision.payload is not None
+    assert gate.store.decide(decision.nonce, True, now=clock()) is not None
+    record = gate.store.get(decision.nonce)
+    assert record is not None
+    clock.now = record.deadline
+    assert gate.store.consume(decision.nonce, now=clock()) is None
+    assert gate.store.get(decision.nonce) is not None
+    assert gate.store.get(decision.nonce).consumed is False  # type: ignore[union-attr]
+    rejected = _retry(gate, decision.payload["requestState"], decision.nonce)
+    assert rejected.action == "REJECT"
+    assert rejected.reason == "timeout"
+    assert [event for event, _ in receipts] == ["timeout"]
+
+
+async def test_db_exact_deadline_approved_refuses_gate_and_store(
+    db_gate: ApprovalGate,
+    clock: _Clock,
+    receipts: list[tuple[str, dict[str, Any]]],
+    clean_database: Any,
+    purge_approval_gate_table: Any,
+) -> None:
+    """The atomic consume UPDATE and the gate agree at now == deadline."""
+    decision = await _apause(db_gate)
+    assert decision.nonce is not None
+    assert decision.payload is not None
+    store = DbApprovalStore()
+    assert await store.adecide(decision.nonce, True, now=clock()) is not None
+    record = await store.aget(decision.nonce)
+    assert record is not None
+    clock.now = record.deadline
+    assert await store.aconsume(decision.nonce, now=clock()) is None
+    rejected = await _aretry(db_gate, decision.payload["requestState"], decision.nonce)
+    assert rejected.action == "REJECT"
+    assert rejected.reason == "timeout"
+    assert [event for event, _ in receipts] == ["timeout"]
+
+
+async def test_db_consume_stamps_passed_now(
+    db_gate: ApprovalGate,
+    clock: _Clock,
+    clean_database: Any,
+    purge_approval_gate_table: Any,
+) -> None:
+    """decided_at/consumed_at come from the gate clock, not wall clock."""
+    decision = await _apause(db_gate)
+    assert decision.nonce is not None
+    store = DbApprovalStore()
+    assert await store.adecide(decision.nonce, True, now=clock()) is not None
+    assert await store.aconsume(decision.nonce, now=clock()) is not None
+    expected = gate_module._to_db_time(clock())
+    factory = get_session_factory()
+    async with factory() as session:
+        result = await session.execute(
+            select(ApprovalGateModel).where(ApprovalGateModel.nonce == decision.nonce)
+        )
+        model = result.scalar_one()
+        assert model.decided_at == expected
+        assert model.consumed_at == expected
+
+
+def test_second_first_call_reuses_pending_record(gate: ApprovalGate) -> None:
+    """A repeat first call for one key reissues the same pause."""
+    first = _pause(gate)
+    second = gate.evaluate(CALLER, TOOL, ARGS, SESSION, True)
+    assert second.action == "PAUSE"
+    assert second.nonce == first.nonce
+    assert len(gate.store.list_pending()) == 1
+
+
+def test_first_call_reuses_approved_unconsumed_record(
+    gate: ApprovalGate,
+) -> None:
+    """Reuse covers approved-but-unspent records, and the pause still spends."""
+    first = _pause(gate)
+    assert first.nonce is not None
+    assert gate.store.decide(first.nonce, True, now=gate.clock()) is not None
+    second = gate.evaluate(CALLER, TOOL, ARGS, SESSION, True)
+    assert second.action == "PAUSE"
+    assert second.nonce == first.nonce
+    assert second.payload is not None
+    executed = _retry(gate, second.payload["requestState"], second.nonce)
+    assert executed.action == "EXECUTE"
+
+
+@pytest.mark.parametrize("terminal", ["denied", "consumed", "expired"])
+def test_first_call_mints_after_terminal_record(
+    gate: ApprovalGate, clock: _Clock, terminal: str
+) -> None:
+    """Denied, spent, and expired records are not reused for a fresh call."""
+    first = _pause(gate)
+    assert first.nonce is not None
+    assert first.payload is not None
+    if terminal == "denied":
+        assert gate.store.decide(first.nonce, False, now=gate.clock()) is not None
+    elif terminal == "consumed":
+        assert gate.store.decide(first.nonce, True, now=gate.clock()) is not None
+        executed = _retry(gate, first.payload["requestState"], first.nonce)
+        assert executed.action == "EXECUTE"
+    else:
+        clock.now += 3600 + 1
+    second = gate.evaluate(CALLER, TOOL, ARGS, SESSION, True)
+    assert second.action == "PAUSE"
+    assert second.nonce != first.nonce
+
+
+async def test_async_second_first_call_reuses_active_record(
+    gate: ApprovalGate,
+) -> None:
+    first = await gate.aevaluate(CALLER, TOOL, ARGS, SESSION, True)
+    second = await gate.aevaluate(CALLER, TOOL, ARGS, SESSION, True)
+    assert second.action == "PAUSE"
+    assert second.nonce == first.nonce
+
+
+async def test_db_second_first_call_reuses_active_record(
+    db_gate: ApprovalGate,
+    clock: _Clock,
+    clean_database: Any,
+    purge_approval_gate_table: Any,
+) -> None:
+    first = await _apause(db_gate)
+    second = await _apause(db_gate)
+    assert second.action == "PAUSE"
+    assert second.nonce == first.nonce
+    pending = await DbApprovalStore().alist_pending(now=clock())
+    assert [record.nonce for record in pending] == [first.nonce]
+
+
+def _seal_duplicate(nonce: str, deadline: float) -> str:
+    """Seal retry state for a duplicate nonce the way the pause path does."""
+    return gate_module._seal_pause_state(
+        SECRET.encode("utf-8"),
+        args_hash=goose_canonical.args_hash(ARGS),
+        caller=CALLER,
+        tool=TOOL,
+        session_id=SESSION,
+        idempotency_key=goose_canonical.idempotency_key(
+            caller=CALLER, tool=TOOL, arguments=ARGS, session_id=SESSION
+        ),
+        deadline=deadline,
+        nonce=nonce,
+    )
+
+
+def _duplicate_nonce(first_nonce: str) -> str:
+    nonce = secrets.token_urlsafe(16)
+    while nonce == first_nonce:
+        nonce = secrets.token_urlsafe(16)
+    return nonce
+
+
+def test_two_approved_pauses_same_key_execute_exactly_once(
+    gate: ApprovalGate, receipts: list[tuple[str, dict[str, Any]]]
+) -> None:
+    """Two approved nonces for one key spend exactly once between them."""
+    first = _pause(gate)
+    assert first.nonce is not None
+    assert first.payload is not None
+    record = gate.store.get(first.nonce)
+    assert record is not None
+    # A duplicate pause for the same key (minted before reuse, or raced in):
+    # same key fields, fresh nonce, saved directly.
+    duplicate = _duplicate_nonce(first.nonce)
+    gate.store.save(replace(record, nonce=duplicate), now=gate.clock())
+    assert gate.store.decide(first.nonce, True, now=gate.clock()) is not None
+    assert gate.store.decide(duplicate, True, now=gate.clock()) is not None
+    executed = _retry(gate, first.payload["requestState"], first.nonce)
+    assert executed.action == "EXECUTE"
+    replayed = _retry(gate, _seal_duplicate(duplicate, record.deadline), duplicate)
+    assert replayed.action == "REJECT"
+    assert replayed.reason == "replayed"
+    assert [event for event, _ in receipts] == ["consumed", "replayed"]
+
+
+async def test_db_two_approved_pauses_same_key_execute_exactly_once(
+    db_gate: ApprovalGate,
+    clock: _Clock,
+    receipts: list[tuple[str, dict[str, Any]]],
+    clean_database: Any,
+    purge_approval_gate_table: Any,
+) -> None:
+    """The NOT EXISTS consume guard holds on the DB path, either order."""
+    decision = await _apause(db_gate)
+    assert decision.nonce is not None
+    assert decision.payload is not None
+    store = DbApprovalStore()
+    record = await store.aget(decision.nonce)
+    assert record is not None
+    duplicate = _duplicate_nonce(decision.nonce)
+    await store.asave(replace(record, nonce=duplicate), now=clock())
+    assert await store.adecide(decision.nonce, True, now=clock()) is not None
+    assert await store.adecide(duplicate, True, now=clock()) is not None
+    # Spend the duplicate first: the original must then refuse.
+    executed = await _aretry(
+        db_gate, _seal_duplicate(duplicate, record.deadline), duplicate
+    )
+    assert executed.action == "EXECUTE"
+    replayed = await _aretry(db_gate, decision.payload["requestState"], decision.nonce)
+    assert replayed.action == "REJECT"
+    assert replayed.reason == "replayed"
+    assert [event for event, _ in receipts] == ["consumed", "replayed"]
+
+
+def test_default_sink_routes_consumed_like_other_events(
+    clock: _Clock, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Without an injected sink, consumed logs exactly like other events."""
+    gate = ApprovalGate(secret=SECRET, store=InMemoryApprovalStore(), clock=clock)
+    with caplog.at_level(logging.INFO, logger="app.services.approval_gate"):
+        first = gate.evaluate(CALLER, TOOL, ARGS, SESSION, True)
+        assert first.nonce is not None
+        assert first.payload is not None
+        assert gate.store.decide(first.nonce, True, now=gate.clock()) is not None
+        executed = _retry(gate, first.payload["requestState"], first.nonce)
+        assert executed.action == "EXECUTE"
+        # The spent record is terminal, so this mints a fresh pause to deny.
+        second = gate.evaluate(CALLER, TOOL, ARGS, SESSION, True)
+        assert second.nonce is not None
+        assert second.nonce != first.nonce
+        assert second.payload is not None
+        assert gate.store.decide(second.nonce, False, now=gate.clock()) is not None
+        denied = _retry(gate, second.payload["requestState"], second.nonce)
+        assert denied.action == "REJECT"
+        assert denied.reason == "denied"
+    lines = [
+        record.getMessage()
+        for record in caplog.records
+        if record.name == "app.services.approval_gate"
+    ]
+    assert any("event=consumed" in line for line in lines)
+    assert any("event=denied" in line for line in lines)
+    assert all(line.startswith("approval_receipt event=") for line in lines)

From b93b631b5487a184c9016aa75b3f8ffd4bcbadea Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Sat, 24 Oct 2026 21:17:22 -0700
Subject: [PATCH 2/5] fix(gateway): refuse post-spend approvals and use
 reserved test endpoints

- refuse a new call at creation when a live spent record exists for the key
- a consume refused for a spent sibling moves the record to denied so it never revives
- CLI tests use fixed RFC 2606/6761/5737/3849 values instead of generated ones

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 gateway/app/services/approval_gate.py   | 170 ++++++++++++++++++---
 gateway/tests/test_amw_approvals_cli.py |  83 +++++------
 gateway/tests/test_approval_gate.py     | 187 +++++++++++++++++++++++-
 3 files changed, 367 insertions(+), 73 deletions(-)

diff --git a/gateway/app/services/approval_gate.py b/gateway/app/services/approval_gate.py
index 6daa1061..8844a90c 100644
--- a/gateway/app/services/approval_gate.py
+++ b/gateway/app/services/approval_gate.py
@@ -178,6 +178,16 @@ def find_active_by_key(
         """
         raise NotImplementedError
 
+    def find_consumed_by_key(
+        self, idempotency_key: str, *, now: float
+    ) -> ApprovalRecord | None:
+        """Live spent record for a key: consumed with deadline after now.
+
+        Used at creation to refuse a fresh mint while a prior spend's
+        deadline is still live; None when no such record exists.
+        """
+        raise NotImplementedError
+
     def list_pending(self, *, now: float | None = None) -> list[ApprovalRecord]:
         """Pending records, excluding ones past deadline when now is given."""
         raise NotImplementedError
@@ -218,6 +228,12 @@ async def afind_active_by_key(
         """Async ``find_active_by_key`` for handlers already on the app loop."""
         raise NotImplementedError
 
+    async def afind_consumed_by_key(
+        self, idempotency_key: str, *, now: float
+    ) -> ApprovalRecord | None:
+        """Async ``find_consumed_by_key`` for handlers on the app loop."""
+        raise NotImplementedError
+
     async def alist_pending(self, *, now: float | None = None) -> list[ApprovalRecord]:
         """Async ``list_pending`` for handlers already on the app event loop."""
         raise NotImplementedError
@@ -272,6 +288,21 @@ def find_active_by_key(
                 return None
             return replace(min(candidates, key=lambda r: (r.created_at, r.nonce)))
 
+    def find_consumed_by_key(
+        self, idempotency_key: str, *, now: float
+    ) -> ApprovalRecord | None:
+        with self._lock:
+            candidates = [
+                record
+                for record in self._records.values()
+                if record.idempotency_key == idempotency_key
+                and record.consumed
+                and record.deadline > now
+            ]
+            if not candidates:
+                return None
+            return replace(min(candidates, key=lambda r: (r.created_at, r.nonce)))
+
     def list_pending(self, *, now: float | None = None) -> list[ApprovalRecord]:
         with self._lock:
             return [
@@ -311,6 +342,12 @@ def consume(self, nonce: str, *, now: float) -> ApprovalRecord | None:
                     and other.consumed
                     and other.deadline > now
                 ):
+                    # Spent sibling blocks this spend: terminalize to denied
+                    # under the same lock so the approval can never revive
+                    # past the sibling's deadline.
+                    self._records[nonce] = replace(
+                        record, status="denied", decided_at=now
+                    )
                     return None
             updated = replace(record, consumed=True)
             self._records[nonce] = updated
@@ -337,6 +374,11 @@ async def afind_active_by_key(
     ) -> ApprovalRecord | None:
         return self.find_active_by_key(idempotency_key, now=now)
 
+    async def afind_consumed_by_key(
+        self, idempotency_key: str, *, now: float
+    ) -> ApprovalRecord | None:
+        return self.find_consumed_by_key(idempotency_key, now=now)
+
     async def alist_pending(self, *, now: float | None = None) -> list[ApprovalRecord]:
         return self.list_pending(now=now)
 
@@ -523,6 +565,41 @@ async def afind_active_by_key(
             model = result.scalar_one_or_none()
             return _record_from_model(model) if model is not None else None
 
+    def find_consumed_by_key(
+        self, idempotency_key: str, *, now: float
+    ) -> ApprovalRecord | None:
+        return _run_sync(self.afind_consumed_by_key(idempotency_key, now=now))
+
+    async def afind_consumed_by_key(
+        self, idempotency_key: str, *, now: float
+    ) -> ApprovalRecord | None:
+        factory = get_session_factory()
+        async with factory() as session:
+            result = await session.execute(
+                select(ApprovalGateModel)
+                .where(
+                    cast(
+                        ColumnElement[bool],
+                        ApprovalGateModel.idempotency_key == idempotency_key,
+                    ),
+                    cast(
+                        ColumnElement[bool],
+                        cast(Any, ApprovalGateModel.consumed_at).is_not(None),
+                    ),
+                    cast(
+                        ColumnElement[bool],
+                        ApprovalGateModel.deadline > _to_db_time(now),
+                    ),
+                )
+                .order_by(
+                    col(ApprovalGateModel.created_at).asc(),
+                    col(ApprovalGateModel.nonce).asc(),
+                )
+                .limit(1)
+            )
+            model = result.scalar_one_or_none()
+            return _record_from_model(model) if model is not None else None
+
     def list_pending(self, *, now: float | None = None) -> list[ApprovalRecord]:
         return _run_sync(self.alist_pending(now=now))
 
@@ -611,7 +688,10 @@ async def aconsume(self, nonce: str, *, now: float) -> ApprovalRecord | None:
         between the gate's expiry check and this UPDATE still cannot spend.
         The ``NOT EXISTS`` sibling check extends the single spend to the
         whole idempotency key: a second nonce approved for the same key
-        cannot spend while the first spend's deadline has not passed.
+        cannot spend while the first spend's deadline has not passed. When
+        the sibling check is what blocks the spend, a second conditional
+        UPDATE terminalizes this nonce to denied so it can never revive
+        past the sibling's deadline.
         """
         factory = get_session_factory()
         async with factory() as session:
@@ -626,26 +706,62 @@ async def aconsume(self, nonce: str, *, now: float) -> ApprovalRecord | None:
             if key_model is None:
                 return None
             key = key_model.idempotency_key
-            sibling_consumed = (
-                select(1)
+
+            def _sibling_live() -> Any:
+                return (
+                    select(1)
+                    .where(
+                        cast(
+                            ColumnElement[bool],
+                            ApprovalGateModel.idempotency_key == key,
+                        ),
+                        cast(ColumnElement[bool], ApprovalGateModel.nonce != nonce),
+                        cast(
+                            ColumnElement[bool],
+                            cast(Any, ApprovalGateModel.consumed_at).is_not(None),
+                        ),
+                        cast(
+                            ColumnElement[bool],
+                            ApprovalGateModel.deadline > _to_db_time(now),
+                        ),
+                    )
+                    .exists()
+                )
+
+            updated = await session.execute(
+                update(ApprovalGateModel)
                 .where(
+                    cast(ColumnElement[bool], ApprovalGateModel.nonce == nonce),
                     cast(
                         ColumnElement[bool],
-                        ApprovalGateModel.idempotency_key == key,
+                        ApprovalGateModel.status == "approved",
                     ),
-                    cast(ColumnElement[bool], ApprovalGateModel.nonce != nonce),
                     cast(
                         ColumnElement[bool],
-                        cast(Any, ApprovalGateModel.consumed_at).is_not(None),
+                        cast(Any, ApprovalGateModel.consumed_at).is_(None),
                     ),
                     cast(
                         ColumnElement[bool],
                         ApprovalGateModel.deadline > _to_db_time(now),
                     ),
+                    cast(ColumnElement[bool], ~_sibling_live()),
                 )
-                .exists()
+                .values(consumed_at=_to_db_time(now))
             )
-            updated = await session.execute(
+            if cast(Any, updated).rowcount == 1:
+                await session.commit()
+                result = await session.execute(
+                    select(ApprovalGateModel).where(
+                        cast(ColumnElement[bool], ApprovalGateModel.nonce == nonce)
+                    )
+                )
+                model = result.scalar_one_or_none()
+                return _record_from_model(model) if model is not None else None
+            # Spend lost: when a live spent sibling is the blocker, move this
+            # nonce to denied in one conditional UPDATE so it stays refused
+            # even after the sibling's deadline passes. Other losses
+            # (missing, denied, consumed, expired) match no row here.
+            await session.execute(
                 update(ApprovalGateModel)
                 .where(
                     cast(ColumnElement[bool], ApprovalGateModel.nonce == nonce),
@@ -661,20 +777,12 @@ async def aconsume(self, nonce: str, *, now: float) -> ApprovalRecord | None:
                         ColumnElement[bool],
                         ApprovalGateModel.deadline > _to_db_time(now),
                     ),
-                    cast(ColumnElement[bool], ~sibling_consumed),
+                    cast(ColumnElement[bool], _sibling_live()),
                 )
-                .values(consumed_at=_to_db_time(now))
+                .values(status="denied", decided_at=_to_db_time(now))
             )
             await session.commit()
-            if cast(Any, updated).rowcount != 1:
-                return None
-            result = await session.execute(
-                select(ApprovalGateModel).where(
-                    cast(ColumnElement[bool], ApprovalGateModel.nonce == nonce)
-                )
-            )
-            model = result.scalar_one_or_none()
-            return _record_from_model(model) if model is not None else None
+            return None
 
     def mark_expired(self, nonce: str, *, now: float) -> ApprovalRecord | None:
         return _run_sync(self.amark_expired(nonce, now=now))
@@ -1051,6 +1159,18 @@ def _pause_new_call(
                 nonce=existing.nonce,
                 payload=_pause_payload(tool, existing.nonce, state),
             )
+        spent = self.store.find_consumed_by_key(idempotency_key, now=now)
+        if spent is not None:
+            # Key already spent and the spend window is still live: refuse
+            # without minting so no second spendable nonce exists.
+            self._emit(
+                "replayed",
+                nonce=None,
+                tool=tool,
+                principal=caller,
+                session_id=session_id,
+            )
+            return ApprovalDecision(action="REJECT", reason="replayed")
         nonce = secrets.token_urlsafe(16)
         deadline = now + self._ttl
         self.store.save(
@@ -1122,6 +1242,18 @@ async def _apause_new_call(
                 nonce=existing.nonce,
                 payload=_pause_payload(tool, existing.nonce, state),
             )
+        spent = await self.store.afind_consumed_by_key(idempotency_key, now=now)
+        if spent is not None:
+            # Key already spent and the spend window is still live: refuse
+            # without minting so no second spendable nonce exists.
+            self._emit(
+                "replayed",
+                nonce=None,
+                tool=tool,
+                principal=caller,
+                session_id=session_id,
+            )
+            return ApprovalDecision(action="REJECT", reason="replayed")
         nonce = secrets.token_urlsafe(16)
         deadline = now + self._ttl
         await self.store.asave(
diff --git a/gateway/tests/test_amw_approvals_cli.py b/gateway/tests/test_amw_approvals_cli.py
index bce1d49f..27d5ce42 100644
--- a/gateway/tests/test_amw_approvals_cli.py
+++ b/gateway/tests/test_amw_approvals_cli.py
@@ -8,7 +8,6 @@
 
 import importlib.util
 import secrets
-import uuid
 from pathlib import Path
 
 import pytest
@@ -119,69 +118,59 @@ def test_unreachable_gateway_error_hides_key(
     assert PLACEHOLDER_KEY not in out.err
 
 
-def _random_port() -> int:
-    return 1024 + secrets.randbelow(65535 - 1024 + 1)
-
-
-def _random_10_host() -> str:
-    return (
-        f"10.{secrets.randbelow(256)}.{secrets.randbelow(256)}"
-        f".{1 + secrets.randbelow(254)}"
-    )
-
-
-def _random_192_168_host() -> str:
-    return f"192.168.{secrets.randbelow(256)}.{1 + secrets.randbelow(254)}"
-
-
-def _random_invalid_host() -> str:
-    return f"{uuid.uuid4().hex}.invalid"
-
-
-def _random_127_host() -> str:
-    return (
-        f"127.{secrets.randbelow(256)}.{secrets.randbelow(256)}"
-        f".{secrets.randbelow(256)}"
-    )
+# Fixed documentation-only fixtures. Every host below is reserved and
+# never routes to a real peer; no test in this file opens a real socket
+# for refused URLs (see _ExplodingClient).
+EXAMPLE_COM = "example.com"  # RFC 2606, reserved example domain
+EXAMPLE_INVALID = "example.invalid"  # RFC 2606, invalid TLD never resolves
+LOCALHOST_NAME = "localhost"  # RFC 6761, special-use loopback name
+TEST_NET_1 = "192.0.2.1"  # RFC 5737, TEST-NET-1 documentation-only
+TEST_NET_2 = "198.51.100.7"  # RFC 5737, TEST-NET-2 documentation-only
+DOC_IPV6 = "2001:db8::1"  # RFC 3849, documentation-only IPv6
+LOOPBACK_V4 = "127.0.0.1"  # RFC 1122, IPv4 loopback
+LOOPBACK_V4_OTHER = "127.0.0.2"  # RFC 1122, same 127/8 loopback range
+LOOPBACK_V6 = "::1"  # RFC 4291, IPv6 loopback
+PRIVATE_10 = "10.0.0.1"  # RFC 1918, private-use address
+PRIVATE_192_168 = "192.168.0.1"  # RFC 1918, private-use address
+HTTPS_TEST_PORT = 8443  # fixed test port in the user range (RFC 6335)
+HTTP_TEST_PORT = 8080  # fixed test port in the user range (RFC 6335)
 
 
 def _allowed_url(case: str) -> str:
-    # Deterministic labels only; random hosts/ports are built in the body so
-    # xdist workers collecting independently see identical test ids.
     if case == "https-any-host":
-        return f"https://{_random_invalid_host()}:{_random_port()}"
+        return f"https://[{DOC_IPV6}]:{HTTPS_TEST_PORT}"
     if case == "https-no-port":
-        return f"https://{_random_invalid_host()}"
+        return f"https://{EXAMPLE_COM}"
     if case == "http-localhost":
-        return f"http://localhost:{_random_port()}"
+        return f"http://{LOCALHOST_NAME}:{HTTP_TEST_PORT}"
     if case == "http-127-1":
-        return f"http://127.0.0.1:{_random_port()}"
+        return f"http://{LOOPBACK_V4}:{HTTP_TEST_PORT}"
     if case == "http-127-range":
-        return f"http://{_random_127_host()}:{_random_port()}"
+        return f"http://{LOOPBACK_V4_OTHER}:{HTTP_TEST_PORT}"
     if case == "http-ipv6-loopback":
-        return f"http://[::1]:{_random_port()}"
+        return f"http://[{LOOPBACK_V6}]:{HTTP_TEST_PORT}"
     if case == "http-uppercase-localhost":
-        return f"HTTP://localhost:{_random_port()}"
+        return f"HTTP://{LOCALHOST_NAME}:{HTTP_TEST_PORT}"
     raise AssertionError(f"unknown allowed case: {case}")
 
 
 def _refused_url(case: str) -> str:
     if case == "http-rfc1918-10":
-        return f"http://{_random_10_host()}:{_random_port()}"
+        return f"http://{PRIVATE_10}:{HTTP_TEST_PORT}"
     if case == "http-rfc1918-192-168":
-        return f"http://{_random_192_168_host()}:{_random_port()}"
+        return f"http://{PRIVATE_192_168}:{HTTP_TEST_PORT}"
     if case == "http-public-hostname":
-        return f"http://{_random_invalid_host()}:{_random_port()}"
+        return f"http://{EXAMPLE_COM}:{HTTP_TEST_PORT}"
     if case == "http-userinfo-trick":
-        return f"http://127.0.0.1@{_random_invalid_host()}:{_random_port()}"
+        return f"http://{LOOPBACK_V4}@{EXAMPLE_INVALID}:{HTTP_TEST_PORT}"
     if case == "http-localhost-userinfo-trick":
-        return f"http://localhost@{_random_invalid_host()}:{_random_port()}/"
+        return f"http://{LOCALHOST_NAME}@{EXAMPLE_INVALID}:{HTTP_TEST_PORT}/"
     if case == "http-localhost-lookalike":
-        return f"http://localhost.{uuid.uuid4().hex}.invalid:{_random_port()}"
+        return f"http://{LOCALHOST_NAME}.{EXAMPLE_INVALID}:{HTTP_TEST_PORT}"
     if case == "http-ip-lookalike":
-        return f"http://127.0.0.1.{uuid.uuid4().hex}.invalid:{_random_port()}"
+        return f"http://{LOOPBACK_V4}.{EXAMPLE_INVALID}:{HTTP_TEST_PORT}"
     if case == "ftp-loopback":
-        return f"ftp://localhost:{_random_port()}/"
+        return f"ftp://{LOCALHOST_NAME}:{HTTP_TEST_PORT}/"
     if case == "file-scheme":
         return "file:///etc/hosts"
     if case == "missing-host":
@@ -189,7 +178,7 @@ def _refused_url(case: str) -> str:
     if case == "empty":
         return ""
     if case == "no-scheme":
-        return f"localhost:{_random_port()}"
+        return f"{LOCALHOST_NAME}:{HTTP_TEST_PORT}"
     raise AssertionError(f"unknown refused case: {case}")
 
 
@@ -255,7 +244,7 @@ def test_refused_base_url_makes_no_request(
     monkeypatch.setattr(amw_approvals.httpx, "Client", _ExplodingClient)
     monkeypatch.setenv("VALID_API_KEYS", PLACEHOLDER_KEY)
     base_url = _refused_url(case)
-    nonce = uuid.uuid4().hex
+    nonce = "test-nonce"
     for argv in (["list"], ["approve", nonce], ["deny", nonce]):
         assert amw_approvals.main(["--base-url", base_url, *argv]) == 1
         err = capsys.readouterr().err
@@ -268,7 +257,7 @@ def test_refusal_happens_before_key_check(
 ) -> None:
     monkeypatch.setattr(amw_approvals.httpx, "Client", _ExplodingClient)
     monkeypatch.delenv("VALID_API_KEYS", raising=False)
-    url = f"http://{_random_10_host()}:{_random_port()}"
+    url = f"http://{TEST_NET_1}:{HTTP_TEST_PORT}"
     assert amw_approvals.main(["--base-url", url, "list"]) == 1
     err = capsys.readouterr().err
     assert "refusing to send the API key" in err
@@ -282,7 +271,7 @@ def test_refused_env_base_url_makes_no_request(
     monkeypatch.setenv("VALID_API_KEYS", PLACEHOLDER_KEY)
     monkeypatch.setenv(
         amw_approvals.BASE_URL_ENV,
-        f"http://{_random_invalid_host()}:{_random_port()}",
+        f"http://{EXAMPLE_INVALID}:{HTTP_TEST_PORT}",
     )
     assert amw_approvals.main(["list"]) == 1
     out = capsys.readouterr()
@@ -292,6 +281,6 @@ def test_refused_env_base_url_makes_no_request(
 
 
 def test_client_refuses_disallowed_url_directly() -> None:
-    url = f"http://{_random_10_host()}:{_random_port()}"
+    url = f"http://{TEST_NET_2}:{HTTP_TEST_PORT}"
     with pytest.raises(ValueError, match="refusing to send the API key"):
         amw_approvals._client(url, PLACEHOLDER_KEY)
diff --git a/gateway/tests/test_approval_gate.py b/gateway/tests/test_approval_gate.py
index e26b57a5..6d01629e 100644
--- a/gateway/tests/test_approval_gate.py
+++ b/gateway/tests/test_approval_gate.py
@@ -843,20 +843,16 @@ def test_first_call_reuses_approved_unconsumed_record(
     assert executed.action == "EXECUTE"
 
 
-@pytest.mark.parametrize("terminal", ["denied", "consumed", "expired"])
+@pytest.mark.parametrize("terminal", ["denied", "expired"])
 def test_first_call_mints_after_terminal_record(
     gate: ApprovalGate, clock: _Clock, terminal: str
 ) -> None:
-    """Denied, spent, and expired records are not reused for a fresh call."""
+    """Denied and expired records are not reused for a fresh call."""
     first = _pause(gate)
     assert first.nonce is not None
     assert first.payload is not None
     if terminal == "denied":
         assert gate.store.decide(first.nonce, False, now=gate.clock()) is not None
-    elif terminal == "consumed":
-        assert gate.store.decide(first.nonce, True, now=gate.clock()) is not None
-        executed = _retry(gate, first.payload["requestState"], first.nonce)
-        assert executed.action == "EXECUTE"
     else:
         clock.now += 3600 + 1
     second = gate.evaluate(CALLER, TOOL, ARGS, SESSION, True)
@@ -974,7 +970,10 @@ def test_default_sink_routes_consumed_like_other_events(
         assert gate.store.decide(first.nonce, True, now=gate.clock()) is not None
         executed = _retry(gate, first.payload["requestState"], first.nonce)
         assert executed.action == "EXECUTE"
-        # The spent record is terminal, so this mints a fresh pause to deny.
+        # Past the spend window a fresh pause mints normally to deny.
+        spent = gate.store.get(first.nonce)
+        assert spent is not None
+        clock.now = spent.deadline + 1
         second = gate.evaluate(CALLER, TOOL, ARGS, SESSION, True)
         assert second.nonce is not None
         assert second.nonce != first.nonce
@@ -991,3 +990,177 @@ def test_default_sink_routes_consumed_like_other_events(
     assert any("event=consumed" in line for line in lines)
     assert any("event=denied" in line for line in lines)
     assert all(line.startswith("approval_receipt event=") for line in lines)
+
+
+def test_first_call_refuses_while_spent_live(
+    gate: ApprovalGate, clock: _Clock, receipts: list[tuple[str, dict[str, Any]]]
+) -> None:
+    """Within the spend window a fresh call refuses without minting."""
+    first = _pause(gate)
+    assert first.nonce is not None
+    assert first.payload is not None
+    assert gate.store.decide(first.nonce, True, now=gate.clock()) is not None
+    executed = _retry(gate, first.payload["requestState"], first.nonce)
+    assert executed.action == "EXECUTE"
+    second = gate.evaluate(CALLER, TOOL, ARGS, SESSION, True)
+    assert second.action == "REJECT"
+    assert second.reason == "replayed"
+    assert second.nonce is None
+    assert gate.store.list_pending(now=clock()) == []
+    assert [event for event, _ in receipts] == ["consumed", "replayed"]
+
+
+async def test_async_first_call_refuses_while_spent_live(
+    gate: ApprovalGate, clock: _Clock, receipts: list[tuple[str, dict[str, Any]]]
+) -> None:
+    """Async pause path refuses a fresh mint while the spend is live."""
+    first = await _apause(gate)
+    assert first.nonce is not None
+    assert first.payload is not None
+    assert await gate.store.adecide(first.nonce, True, now=gate.clock()) is not None
+    executed = await _aretry(gate, first.payload["requestState"], first.nonce)
+    assert executed.action == "EXECUTE"
+    second = await gate.aevaluate(CALLER, TOOL, ARGS, SESSION, True)
+    assert second.action == "REJECT"
+    assert second.reason == "replayed"
+    assert second.nonce is None
+    assert await gate.store.alist_pending(now=clock()) == []
+    assert [event for event, _ in receipts] == ["consumed", "replayed"]
+
+
+async def test_db_first_call_refuses_while_spent_live(
+    db_gate: ApprovalGate,
+    clock: _Clock,
+    receipts: list[tuple[str, dict[str, Any]]],
+    clean_database: Any,
+    purge_approval_gate_table: Any,
+) -> None:
+    """DB pause path refuses a fresh mint while the spend is live."""
+    first = await _apause(db_gate)
+    assert first.nonce is not None
+    assert first.payload is not None
+    store = DbApprovalStore()
+    assert await store.adecide(first.nonce, True, now=clock()) is not None
+    executed = await _aretry(db_gate, first.payload["requestState"], first.nonce)
+    assert executed.action == "EXECUTE"
+    second = await db_gate.aevaluate(CALLER, TOOL, ARGS, SESSION, True)
+    assert second.action == "REJECT"
+    assert second.reason == "replayed"
+    assert second.nonce is None
+    assert await store.alist_pending(now=clock()) == []
+    assert [event for event, _ in receipts] == ["consumed", "replayed"]
+
+
+def test_sibling_refused_stays_refused_past_spent_deadline(
+    gate: ApprovalGate, clock: _Clock
+) -> None:
+    """A sibling blocked by a live spend stays refused past that deadline."""
+    first = _pause(gate)
+    assert first.nonce is not None
+    assert first.payload is not None
+    record = gate.store.get(first.nonce)
+    assert record is not None
+    duplicate = _duplicate_nonce(first.nonce)
+    gate.store.save(
+        replace(record, nonce=duplicate, deadline=record.deadline + 500),
+        now=gate.clock(),
+    )
+    assert gate.store.decide(first.nonce, True, now=gate.clock()) is not None
+    assert gate.store.decide(duplicate, True, now=gate.clock()) is not None
+    executed = _retry(gate, first.payload["requestState"], first.nonce)
+    assert executed.action == "EXECUTE"
+    assert gate.store.consume(duplicate, now=clock()) is None
+    blocked = gate.store.get(duplicate)
+    assert blocked is not None
+    assert blocked.status == "denied"
+    spent = gate.store.get(first.nonce)
+    assert spent is not None
+    clock.now = spent.deadline + 1
+    assert clock() < blocked.deadline
+    assert gate.store.consume(duplicate, now=clock()) is None
+    assert gate.store.get(duplicate) is not None
+    assert gate.store.get(duplicate).status == "denied"  # type: ignore[union-attr]
+
+
+async def test_db_sibling_refused_stays_refused_past_spent_deadline(
+    db_gate: ApprovalGate,
+    clock: _Clock,
+    clean_database: Any,
+    purge_approval_gate_table: Any,
+) -> None:
+    """DB sibling blocked by a live spend stays refused past that deadline."""
+    first = await _apause(db_gate)
+    assert first.nonce is not None
+    assert first.payload is not None
+    store = DbApprovalStore()
+    record = await store.aget(first.nonce)
+    assert record is not None
+    duplicate = _duplicate_nonce(first.nonce)
+    await store.asave(
+        replace(record, nonce=duplicate, deadline=record.deadline + 500),
+        now=clock(),
+    )
+    assert await store.adecide(first.nonce, True, now=clock()) is not None
+    assert await store.adecide(duplicate, True, now=clock()) is not None
+    executed = await _aretry(db_gate, first.payload["requestState"], first.nonce)
+    assert executed.action == "EXECUTE"
+    assert await store.aconsume(duplicate, now=clock()) is None
+    blocked = await store.aget(duplicate)
+    assert blocked is not None
+    assert blocked.status == "denied"
+    spent = await store.aget(first.nonce)
+    assert spent is not None
+    clock.now = spent.deadline + 1
+    assert clock() < blocked.deadline
+    assert await store.aconsume(duplicate, now=clock()) is None
+    refused = await store.aget(duplicate)
+    assert refused is not None
+    assert refused.status == "denied"
+
+
+def test_first_call_mints_after_spent_window(gate: ApprovalGate, clock: _Clock) -> None:
+    """Past the spend window a brand-new call mints and spends normally."""
+    first = _pause(gate)
+    assert first.nonce is not None
+    assert first.payload is not None
+    assert gate.store.decide(first.nonce, True, now=gate.clock()) is not None
+    executed = _retry(gate, first.payload["requestState"], first.nonce)
+    assert executed.action == "EXECUTE"
+    spent = gate.store.get(first.nonce)
+    assert spent is not None
+    clock.now = spent.deadline + 1
+    second = gate.evaluate(CALLER, TOOL, ARGS, SESSION, True)
+    assert second.action == "PAUSE"
+    assert second.nonce is not None
+    assert second.nonce != first.nonce
+    assert second.payload is not None
+    assert gate.store.decide(second.nonce, True, now=gate.clock()) is not None
+    replay = _retry(gate, second.payload["requestState"], second.nonce)
+    assert replay.action == "EXECUTE"
+
+
+async def test_db_first_call_mints_after_spent_window(
+    db_gate: ApprovalGate,
+    clock: _Clock,
+    clean_database: Any,
+    purge_approval_gate_table: Any,
+) -> None:
+    """DB path mints and spends normally past the spend window."""
+    first = await _apause(db_gate)
+    assert first.nonce is not None
+    assert first.payload is not None
+    store = DbApprovalStore()
+    assert await store.adecide(first.nonce, True, now=clock()) is not None
+    executed = await _aretry(db_gate, first.payload["requestState"], first.nonce)
+    assert executed.action == "EXECUTE"
+    spent = await store.aget(first.nonce)
+    assert spent is not None
+    clock.now = spent.deadline + 1
+    second = await db_gate.aevaluate(CALLER, TOOL, ARGS, SESSION, True)
+    assert second.action == "PAUSE"
+    assert second.nonce is not None
+    assert second.nonce != first.nonce
+    assert second.payload is not None
+    assert await store.adecide(second.nonce, True, now=clock()) is not None
+    replay = await _aretry(db_gate, second.payload["requestState"], second.nonce)
+    assert replay.action == "EXECUTE"

From fd261f8cd5be6ab7087955acc430708088eaed18 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Sat, 24 Oct 2026 21:29:57 -0700
Subject: [PATCH 3/5] fix(gateway): harden approvals CLI and check spent keys
 first

- CLI httpx client uses trust_env=False so proxy env vars never see the admin key
- CLI refuses URLs carrying userinfo; malformed IPv6 refused cleanly
- first-call path checks for a live spent record before reusing an active one

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 gateway/app/services/approval_gate.py   | 48 +++++++-------
 gateway/scripts/amw_approvals.py        | 13 +++-
 gateway/tests/test_amw_approvals_cli.py | 39 +++++++++++
 gateway/tests/test_approval_gate.py     | 86 +++++++++++++++++++++++++
 4 files changed, 160 insertions(+), 26 deletions(-)

diff --git a/gateway/app/services/approval_gate.py b/gateway/app/services/approval_gate.py
index 8844a90c..a4bc6f6a 100644
--- a/gateway/app/services/approval_gate.py
+++ b/gateway/app/services/approval_gate.py
@@ -1140,6 +1140,18 @@ def _pause_new_call(
             return ApprovalDecision(action="REJECT", reason="unsupported_client")
         secret = self._require_secret()
         now = self.clock()
+        spent = self.store.find_consumed_by_key(idempotency_key, now=now)
+        if spent is not None:
+            # Key already spent and the spend window is still live: refuse
+            # without minting so no second spendable nonce exists.
+            self._emit(
+                "replayed",
+                nonce=None,
+                tool=tool,
+                principal=caller,
+                session_id=session_id,
+            )
+            return ApprovalDecision(action="REJECT", reason="replayed")
         existing = self.store.find_active_by_key(idempotency_key, now=now)
         if existing is not None:
             # Same logical call already paused: reissue its pause instead of
@@ -1159,18 +1171,6 @@ def _pause_new_call(
                 nonce=existing.nonce,
                 payload=_pause_payload(tool, existing.nonce, state),
             )
-        spent = self.store.find_consumed_by_key(idempotency_key, now=now)
-        if spent is not None:
-            # Key already spent and the spend window is still live: refuse
-            # without minting so no second spendable nonce exists.
-            self._emit(
-                "replayed",
-                nonce=None,
-                tool=tool,
-                principal=caller,
-                session_id=session_id,
-            )
-            return ApprovalDecision(action="REJECT", reason="replayed")
         nonce = secrets.token_urlsafe(16)
         deadline = now + self._ttl
         self.store.save(
@@ -1223,6 +1223,18 @@ async def _apause_new_call(
             return ApprovalDecision(action="REJECT", reason="unsupported_client")
         secret = self._require_secret()
         now = self.clock()
+        spent = await self.store.afind_consumed_by_key(idempotency_key, now=now)
+        if spent is not None:
+            # Key already spent and the spend window is still live: refuse
+            # without minting so no second spendable nonce exists.
+            self._emit(
+                "replayed",
+                nonce=None,
+                tool=tool,
+                principal=caller,
+                session_id=session_id,
+            )
+            return ApprovalDecision(action="REJECT", reason="replayed")
         existing = await self.store.afind_active_by_key(idempotency_key, now=now)
         if existing is not None:
             # Same logical call already paused: reissue its pause instead of
@@ -1242,18 +1254,6 @@ async def _apause_new_call(
                 nonce=existing.nonce,
                 payload=_pause_payload(tool, existing.nonce, state),
             )
-        spent = await self.store.afind_consumed_by_key(idempotency_key, now=now)
-        if spent is not None:
-            # Key already spent and the spend window is still live: refuse
-            # without minting so no second spendable nonce exists.
-            self._emit(
-                "replayed",
-                nonce=None,
-                tool=tool,
-                principal=caller,
-                session_id=session_id,
-            )
-            return ApprovalDecision(action="REJECT", reason="replayed")
         nonce = secrets.token_urlsafe(16)
         deadline = now + self._ttl
         await self.store.asave(
diff --git a/gateway/scripts/amw_approvals.py b/gateway/scripts/amw_approvals.py
index 1a743b47..3c2ee71f 100644
--- a/gateway/scripts/amw_approvals.py
+++ b/gateway/scripts/amw_approvals.py
@@ -39,7 +39,9 @@ def _refusal_reason(base_url: str) -> str | None:
     """Why the API key must not be sent to base_url; None when allowed.
 
     Allowed: any https URL, or http only on loopback (localhost, 127.0.0.0/8,
-    ::1). The host comes from parsed.hostname, so userinfo tricks such as
+    ::1). Any userinfo (username or password) refuses before the allowed
+    returns, so http://guest@localhost still refuses. Otherwise the host
+    comes from parsed.hostname, so userinfo tricks such as
     http://127.0.0.1@evil resolve to the real host (evil), and look-alikes
     such as localhost.example or 127.0.0.1.nip.io fail both the exact
     localhost match and the IP parse. Unknown schemes and empty hosts refuse.
@@ -56,6 +58,8 @@ def _refusal_reason(base_url: str) -> str | None:
         return "missing URL scheme"
     if not host:
         return "missing host"
+    if parsed.username is not None or parsed.password is not None:
+        return "userinfo not allowed"
     if scheme == "https":
         return None
     if host == "localhost":
@@ -81,8 +85,13 @@ def _client(base_url: str, api_key: str) -> httpx.Client:
     reason = _refusal_reason(base_url)
     if reason is not None:
         raise ValueError(f"refusing to send the API key: {reason}")
+    # trust_env=False: httpx honors ambient HTTP_PROXY/HTTPS_PROXY/ALL_PROXY
+    # by default, which would hand the bootstrap admin key to a proxy.
     return httpx.Client(
-        base_url=base_url, timeout=10.0, headers={API_KEY_HEADER: api_key}
+        base_url=base_url,
+        timeout=10.0,
+        headers={API_KEY_HEADER: api_key},
+        trust_env=False,
     )
 
 
diff --git a/gateway/tests/test_amw_approvals_cli.py b/gateway/tests/test_amw_approvals_cli.py
index 27d5ce42..519a845f 100644
--- a/gateway/tests/test_amw_approvals_cli.py
+++ b/gateway/tests/test_amw_approvals_cli.py
@@ -84,7 +84,9 @@ def __init__(
             base_url: str = "",
             timeout: float | None = None,
             headers: dict[str, str] | None = None,
+            trust_env: bool = True,
         ) -> None:
+            assert trust_env is False
             self._headers = dict(headers or {})
 
         def get(self, path: str) -> _StubResponse:
@@ -165,6 +167,15 @@ def _refused_url(case: str) -> str:
         return f"http://{LOOPBACK_V4}@{EXAMPLE_INVALID}:{HTTP_TEST_PORT}"
     if case == "http-localhost-userinfo-trick":
         return f"http://{LOCALHOST_NAME}@{EXAMPLE_INVALID}:{HTTP_TEST_PORT}/"
+    if case == "http-userinfo-localhost":
+        # RFC 6761 localhost, RFC 6335 fixed port; userinfo refuses outright.
+        return f"http://guest@{LOCALHOST_NAME}:{HTTP_TEST_PORT}"
+    if case == "https-userinfo-example":
+        # RFC 2606 example.com; userinfo refuses even on allowed https.
+        return f"https://guest@{EXAMPLE_COM}"
+    if case == "http-malformed-ipv6-bracket":
+        # RFC 3849 documentation IPv6, missing closing bracket: ValueError.
+        return f"http://[{DOC_IPV6}"
     if case == "http-localhost-lookalike":
         return f"http://{LOCALHOST_NAME}.{EXAMPLE_INVALID}:{HTTP_TEST_PORT}"
     if case == "http-ip-lookalike":
@@ -229,6 +240,9 @@ def __init__(self, *args: object, **kwargs: object) -> None:
         "http-public-hostname",
         "http-userinfo-trick",
         "http-localhost-userinfo-trick",
+        "http-userinfo-localhost",
+        "https-userinfo-example",
+        "http-malformed-ipv6-bracket",
         "http-localhost-lookalike",
         "http-ip-lookalike",
         "ftp-loopback",
@@ -284,3 +298,28 @@ def test_client_refuses_disallowed_url_directly() -> None:
     url = f"http://{TEST_NET_2}:{HTTP_TEST_PORT}"
     with pytest.raises(ValueError, match="refusing to send the API key"):
         amw_approvals._client(url, PLACEHOLDER_KEY)
+
+
+def test_client_disables_trust_env_and_ignores_proxies(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # RFC 5737 TEST-NET-1, RFC 6335 fixed port: proxy decoy, never dialed.
+    monkeypatch.setenv("HTTP_PROXY", f"http://{TEST_NET_1}:{HTTP_TEST_PORT}")
+    # RFC 5737 TEST-NET-2, RFC 6335 fixed port: proxy decoy, never dialed.
+    monkeypatch.setenv("ALL_PROXY", f"http://{TEST_NET_2}:{HTTP_TEST_PORT}")
+    # RFC 6761 localhost, RFC 6335 fixed port: allowed http loopback URL.
+    client = amw_approvals._client(
+        f"http://{LOCALHOST_NAME}:{HTTP_TEST_PORT}", PLACEHOLDER_KEY
+    )
+    try:
+        assert client.trust_env is False
+        assert dict(client._mounts) == {}
+        # Same env with trust_env=True mounts proxies, proving the decoys
+        # above are live env httpx would otherwise honor. No request is sent.
+        proving = amw_approvals.httpx.Client(trust_env=True)
+        try:
+            assert len(proving._mounts) > 0
+        finally:
+            proving.close()
+    finally:
+        client.close()
diff --git a/gateway/tests/test_approval_gate.py b/gateway/tests/test_approval_gate.py
index 6d01629e..ea414bf4 100644
--- a/gateway/tests/test_approval_gate.py
+++ b/gateway/tests/test_approval_gate.py
@@ -1164,3 +1164,89 @@ async def test_db_first_call_mints_after_spent_window(
     assert await store.adecide(second.nonce, True, now=clock()) is not None
     replay = await _aretry(db_gate, second.payload["requestState"], second.nonce)
     assert replay.action == "EXECUTE"
+
+
+def test_first_call_prefers_spent_over_live_active_sibling(
+    gate: ApprovalGate, clock: _Clock, receipts: list[tuple[str, dict[str, Any]]]
+) -> None:
+    """A live spend beats a live approved sibling on a fresh call."""
+    first = _pause(gate)
+    assert first.nonce is not None
+    assert first.payload is not None
+    record = gate.store.get(first.nonce)
+    assert record is not None
+    duplicate = _duplicate_nonce(first.nonce)
+    gate.store.save(replace(record, nonce=duplicate), now=gate.clock())
+    assert gate.store.decide(first.nonce, True, now=gate.clock()) is not None
+    assert gate.store.decide(duplicate, True, now=gate.clock()) is not None
+    executed = _retry(gate, first.payload["requestState"], first.nonce)
+    assert executed.action == "EXECUTE"
+    key = goose_canonical.idempotency_key(
+        caller=CALLER, tool=TOOL, arguments=ARGS, session_id=SESSION
+    )
+    assert gate.store.find_consumed_by_key(key, now=clock()) is not None
+    assert gate.store.find_active_by_key(key, now=clock()) is not None
+    second = gate.evaluate(CALLER, TOOL, ARGS, SESSION, True)
+    assert second.action == "REJECT"
+    assert second.reason == "replayed"
+    assert second.nonce is None
+    assert [event for event, _ in receipts] == ["consumed", "replayed"]
+
+
+async def test_async_first_call_prefers_spent_over_live_active_sibling(
+    gate: ApprovalGate, clock: _Clock, receipts: list[tuple[str, dict[str, Any]]]
+) -> None:
+    """Async pause path also refuses when a live spend and sibling coexist."""
+    first = await _apause(gate)
+    assert first.nonce is not None
+    assert first.payload is not None
+    record = await gate.store.aget(first.nonce)
+    assert record is not None
+    duplicate = _duplicate_nonce(first.nonce)
+    await gate.store.asave(replace(record, nonce=duplicate), now=gate.clock())
+    assert await gate.store.adecide(first.nonce, True, now=gate.clock()) is not None
+    assert await gate.store.adecide(duplicate, True, now=gate.clock()) is not None
+    executed = await _aretry(gate, first.payload["requestState"], first.nonce)
+    assert executed.action == "EXECUTE"
+    key = goose_canonical.idempotency_key(
+        caller=CALLER, tool=TOOL, arguments=ARGS, session_id=SESSION
+    )
+    assert await gate.store.afind_consumed_by_key(key, now=clock()) is not None
+    assert await gate.store.afind_active_by_key(key, now=clock()) is not None
+    second = await gate.aevaluate(CALLER, TOOL, ARGS, SESSION, True)
+    assert second.action == "REJECT"
+    assert second.reason == "replayed"
+    assert second.nonce is None
+    assert [event for event, _ in receipts] == ["consumed", "replayed"]
+
+
+async def test_db_first_call_prefers_spent_over_live_active_sibling(
+    db_gate: ApprovalGate,
+    clock: _Clock,
+    receipts: list[tuple[str, dict[str, Any]]],
+    clean_database: Any,
+    purge_approval_gate_table: Any,
+) -> None:
+    """DB pause path refuses when a live spend and sibling coexist."""
+    first = await _apause(db_gate)
+    assert first.nonce is not None
+    assert first.payload is not None
+    store = DbApprovalStore()
+    record = await store.aget(first.nonce)
+    assert record is not None
+    duplicate = _duplicate_nonce(first.nonce)
+    await store.asave(replace(record, nonce=duplicate), now=clock())
+    assert await store.adecide(first.nonce, True, now=clock()) is not None
+    assert await store.adecide(duplicate, True, now=clock()) is not None
+    executed = await _aretry(db_gate, first.payload["requestState"], first.nonce)
+    assert executed.action == "EXECUTE"
+    key = goose_canonical.idempotency_key(
+        caller=CALLER, tool=TOOL, arguments=ARGS, session_id=SESSION
+    )
+    assert await store.afind_consumed_by_key(key, now=clock()) is not None
+    assert await store.afind_active_by_key(key, now=clock()) is not None
+    second = await db_gate.aevaluate(CALLER, TOOL, ARGS, SESSION, True)
+    assert second.action == "REJECT"
+    assert second.reason == "replayed"
+    assert second.nonce is None
+    assert [event for event, _ in receipts] == ["consumed", "replayed"]

From a4431dae9a211d7311543dc36f3f90c73664e7e3 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Sat, 24 Oct 2026 21:44:48 -0700
Subject: [PATCH 4/5] fix(gateway): reject malformed approvals CLI URLs cleanly

Refuse control characters and whitespace, validate with httpx.URL, require a numeric port, and turn httpx.InvalidURL into a clean error that never prints the key.

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 gateway/scripts/amw_approvals.py        | 84 ++++++++++++++++++++-----
 gateway/tests/test_amw_approvals_cli.py | 21 ++++++-
 2 files changed, 88 insertions(+), 17 deletions(-)

diff --git a/gateway/scripts/amw_approvals.py b/gateway/scripts/amw_approvals.py
index 3c2ee71f..5bc85733 100644
--- a/gateway/scripts/amw_approvals.py
+++ b/gateway/scripts/amw_approvals.py
@@ -19,7 +19,6 @@
 import os
 import sys
 from datetime import datetime, timezone
-from urllib.parse import urlparse
 
 import httpx
 
@@ -35,31 +34,73 @@
 _LOOPBACK_V6 = ipaddress.ip_address("::1")
 
 
+def _raw_authority(base_url: str) -> str | None:
+    """Authority after '://' up to / ? #; None when there is no '://'."""
+    sep = base_url.find("://")
+    if sep == -1:
+        return None
+    rest = base_url[sep + 3 :]
+    for i, ch in enumerate(rest):
+        if ch in "/?#":
+            return rest[:i]
+    return rest
+
+
+def _raw_port(authority: str) -> str | None:
+    """Port substring from a raw authority; None when absent or malformed."""
+    hostport = authority.rsplit("@", 1)[-1]
+    if hostport.startswith("["):
+        end = hostport.find("]")
+        if end == -1:
+            return None
+        after = hostport[end + 1 :]
+        if not after:
+            return None
+        if after.startswith(":"):
+            return after[1:]
+        return None
+    if ":" not in hostport:
+        return None
+    return hostport.rsplit(":", 1)[-1]
+
+
 def _refusal_reason(base_url: str) -> str | None:
     """Why the API key must not be sent to base_url; None when allowed.
 
     Allowed: any https URL, or http only on loopback (localhost, 127.0.0.0/8,
-    ::1). Any userinfo (username or password) refuses before the allowed
-    returns, so http://guest@localhost still refuses. Otherwise the host
-    comes from parsed.hostname, so userinfo tricks such as
+    ::1). Control characters, whitespace, and non-numeric ports refuse
+    before the allowed returns. Any userinfo (username or password) refuses
+    before the allowed returns, so http://guest@localhost still refuses.
+    Otherwise the host comes from httpx.URL.host, so userinfo tricks such as
     http://127.0.0.1@evil resolve to the real host (evil), and look-alikes
     such as localhost.example or 127.0.0.1.nip.io fail both the exact
     localhost match and the IP parse. Unknown schemes and empty hosts refuse.
+    The check uses httpx.URL, the same parser as the request.
     """
+    for ch in base_url:
+        if ord(ch) < 32 or ord(ch) == 127 or ch.isspace():
+            return "invalid characters in URL"
     try:
-        parsed = urlparse(base_url)
-        host = parsed.hostname or ""
-    except ValueError:
+        parsed = httpx.URL(base_url)
+    except httpx.InvalidURL:
         return "unparseable URL"
     scheme = parsed.scheme.lower()
     if scheme not in ("http", "https"):
         if parsed.scheme:
             return f"unsupported scheme {parsed.scheme!r}"
         return "missing URL scheme"
+    host = parsed.host or ""
     if not host:
         return "missing host"
-    if parsed.username is not None or parsed.password is not None:
+    if parsed.username or parsed.password or parsed.userinfo:
+        return "userinfo not allowed"
+    authority = _raw_authority(base_url)
+    if authority is not None and "@" in authority:
         return "userinfo not allowed"
+    if authority is not None:
+        port_str = _raw_port(authority)
+        if port_str and not (port_str.isascii() and port_str.isdigit()):
+            return f"invalid port {port_str!r}"
     if scheme == "https":
         return None
     if host == "localhost":
@@ -87,12 +128,15 @@ def _client(base_url: str, api_key: str) -> httpx.Client:
         raise ValueError(f"refusing to send the API key: {reason}")
     # trust_env=False: httpx honors ambient HTTP_PROXY/HTTPS_PROXY/ALL_PROXY
     # by default, which would hand the bootstrap admin key to a proxy.
-    return httpx.Client(
-        base_url=base_url,
-        timeout=10.0,
-        headers={API_KEY_HEADER: api_key},
-        trust_env=False,
-    )
+    try:
+        return httpx.Client(
+            base_url=base_url,
+            timeout=10.0,
+            headers={API_KEY_HEADER: api_key},
+            trust_env=False,
+        )
+    except httpx.InvalidURL:
+        raise ValueError("refusing to send the API key: unparseable URL") from None
 
 
 def _format_time(value: float) -> str:
@@ -105,6 +149,12 @@ def cmd_list(args: argparse.Namespace) -> int:
     try:
         response = _client(args.base_url, args.api_key).get("/v1/approvals/pending")
         response.raise_for_status()
+    except ValueError as exc:
+        print(f"error: {exc}", file=sys.stderr)
+        return 1
+    except httpx.InvalidURL as exc:
+        print(f"error: invalid gateway URL: {exc}", file=sys.stderr)
+        return 1
     except httpx.HTTPError as exc:
         print(
             f"error: gateway not reachable at {args.base_url}: {exc}", file=sys.stderr
@@ -127,6 +177,12 @@ def _decide(args: argparse.Namespace, action: str) -> int:
         response = _client(args.base_url, args.api_key).post(
             f"/v1/approvals/{args.nonce}/{action}"
         )
+    except ValueError as exc:
+        print(f"error: {exc}", file=sys.stderr)
+        return 1
+    except httpx.InvalidURL as exc:
+        print(f"error: invalid gateway URL: {exc}", file=sys.stderr)
+        return 1
     except httpx.HTTPError as exc:
         print(
             f"error: gateway not reachable at {args.base_url}: {exc}", file=sys.stderr
diff --git a/gateway/tests/test_amw_approvals_cli.py b/gateway/tests/test_amw_approvals_cli.py
index 519a845f..b35c1892 100644
--- a/gateway/tests/test_amw_approvals_cli.py
+++ b/gateway/tests/test_amw_approvals_cli.py
@@ -190,6 +190,15 @@ def _refused_url(case: str) -> str:
         return ""
     if case == "no-scheme":
         return f"{LOCALHOST_NAME}:{HTTP_TEST_PORT}"
+    if case == "http-tab-in-host":
+        # RFC 6761 localhost variant with a tab; control char refuses.
+        return f"http://local\thost:{HTTP_TEST_PORT}"
+    if case == "http-non-numeric-port":
+        # RFC 6761 localhost; non-numeric port refuses.
+        return f"http://{LOCALHOST_NAME}:abc"
+    if case == "https-newline":
+        # RFC 2606 example.com; newline refuses even on https.
+        return f"https://{EXAMPLE_COM}\n/"
     raise AssertionError(f"unknown refused case: {case}")
 
 
@@ -250,6 +259,9 @@ def __init__(self, *args: object, **kwargs: object) -> None:
         "missing-host",
         "empty",
         "no-scheme",
+        "http-tab-in-host",
+        "http-non-numeric-port",
+        "https-newline",
     ],
 )
 def test_refused_base_url_makes_no_request(
@@ -261,9 +273,12 @@ def test_refused_base_url_makes_no_request(
     nonce = "test-nonce"
     for argv in (["list"], ["approve", nonce], ["deny", nonce]):
         assert amw_approvals.main(["--base-url", base_url, *argv]) == 1
-        err = capsys.readouterr().err
-        assert "refusing to send the API key" in err
-        assert PLACEHOLDER_KEY not in err
+        out = capsys.readouterr()
+        assert "refusing to send the API key" in out.err
+        assert "Traceback" not in out.err
+        assert "Traceback" not in out.out
+        assert PLACEHOLDER_KEY not in out.err
+        assert PLACEHOLDER_KEY not in out.out
 
 
 def test_refusal_happens_before_key_check(

From 0e6f4003491df470dc52fc93682ade6ff58bc168 Mon Sep 17 00:00:00 2001
From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
Date: Sat, 24 Oct 2026 21:52:53 -0700
Subject: [PATCH 5/5] fix(gateway): refuse malformed IDNA hosts in approvals
 CLI

Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com>
---
 gateway/scripts/amw_approvals.py        | 19 ++++++++++++++-----
 gateway/tests/test_amw_approvals_cli.py | 16 +++++++++++++---
 2 files changed, 27 insertions(+), 8 deletions(-)

diff --git a/gateway/scripts/amw_approvals.py b/gateway/scripts/amw_approvals.py
index 5bc85733..09e93cd5 100644
--- a/gateway/scripts/amw_approvals.py
+++ b/gateway/scripts/amw_approvals.py
@@ -75,24 +75,31 @@ def _refusal_reason(base_url: str) -> str | None:
     http://127.0.0.1@evil resolve to the real host (evil), and look-alikes
     such as localhost.example or 127.0.0.1.nip.io fail both the exact
     localhost match and the IP parse. Unknown schemes and empty hosts refuse.
-    The check uses httpx.URL, the same parser as the request.
+    The check uses httpx.URL, the same parser as the request. IDNA decode
+    errors on host access (malformed punycode) refuse as unparseable.
     """
     for ch in base_url:
         if ord(ch) < 32 or ord(ch) == 127 or ch.isspace():
             return "invalid characters in URL"
     try:
         parsed = httpx.URL(base_url)
-    except httpx.InvalidURL:
+    except (httpx.InvalidURL, UnicodeError, ValueError):
         return "unparseable URL"
     scheme = parsed.scheme.lower()
     if scheme not in ("http", "https"):
         if parsed.scheme:
             return f"unsupported scheme {parsed.scheme!r}"
         return "missing URL scheme"
-    host = parsed.host or ""
+    try:
+        host = parsed.host or ""
+        has_userinfo = bool(parsed.username or parsed.password or parsed.userinfo)
+    except (UnicodeError, ValueError):
+        # httpx decodes IDNA on .host access, so malformed punycode
+        # (e.g. http://xn--) raises here, not at URL construction.
+        return "unparseable URL"
     if not host:
         return "missing host"
-    if parsed.username or parsed.password or parsed.userinfo:
+    if has_userinfo:
         return "userinfo not allowed"
     authority = _raw_authority(base_url)
     if authority is not None and "@" in authority:
@@ -135,7 +142,9 @@ def _client(base_url: str, api_key: str) -> httpx.Client:
             headers={API_KEY_HEADER: api_key},
             trust_env=False,
         )
-    except httpx.InvalidURL:
+    except (httpx.InvalidURL, UnicodeError, ValueError):
+        # All other arguments are constants, so any decode error here is
+        # URL-driven and refuses cleanly without ever printing the key.
         raise ValueError("refusing to send the API key: unparseable URL") from None
 
 
diff --git a/gateway/tests/test_amw_approvals_cli.py b/gateway/tests/test_amw_approvals_cli.py
index b35c1892..3a2f7227 100644
--- a/gateway/tests/test_amw_approvals_cli.py
+++ b/gateway/tests/test_amw_approvals_cli.py
@@ -199,6 +199,9 @@ def _refused_url(case: str) -> str:
     if case == "https-newline":
         # RFC 2606 example.com; newline refuses even on https.
         return f"https://{EXAMPLE_COM}\n/"
+    if case == "http-malformed-idna-label":
+        # RFC 3492/5891 punycode, malformed on purpose; .invalid is RFC 2606.
+        return f"http://xn--.{EXAMPLE_INVALID}:{HTTP_TEST_PORT}"
     raise AssertionError(f"unknown refused case: {case}")
 
 
@@ -262,6 +265,7 @@ def __init__(self, *args: object, **kwargs: object) -> None:
         "http-tab-in-host",
         "http-non-numeric-port",
         "https-newline",
+        "http-malformed-idna-label",
     ],
 )
 def test_refused_base_url_makes_no_request(
@@ -310,9 +314,15 @@ def test_refused_env_base_url_makes_no_request(
 
 
 def test_client_refuses_disallowed_url_directly() -> None:
-    url = f"http://{TEST_NET_2}:{HTTP_TEST_PORT}"
-    with pytest.raises(ValueError, match="refusing to send the API key"):
-        amw_approvals._client(url, PLACEHOLDER_KEY)
+    urls = [
+        f"http://{TEST_NET_2}:{HTTP_TEST_PORT}",
+        # RFC 3492/5891 punycode, malformed on purpose; .invalid is RFC 2606.
+        f"http://xn--.{EXAMPLE_INVALID}:{HTTP_TEST_PORT}",
+    ]
+    for url in urls:
+        with pytest.raises(ValueError, match="refusing to send the API key") as exc:
+            amw_approvals._client(url, PLACEHOLDER_KEY)
+        assert PLACEHOLDER_KEY not in str(exc.value)
 
 
 def test_client_disables_trust_env_and_ignores_proxies(