Skip to content
Merged
Show file tree
Hide file tree
Changes from 5 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 12 additions & 4 deletions integrations/openai-agent-middleware/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,10 @@ plane's `Idempotency-Key` exists to capture, so `GovernedToolRunner`:
1. derives the key from the id (`oai-<tool_call.id>`) and **never invents
one** — a tool call without an id is refused, because an invented key would
turn the retry into a second charged action;
2. writes the derivation to an `OperationKeyStore` **before the first network
call**, so a crash between "the model asked" and "the receipt came back"
resumes with the same key;
2. writes the derivation, including the canonical JSON arguments, to an
`OperationKeyStore` **before the first network call**, so a crash between
"the model asked" and "the receipt came back" resumes with the same key
and the same arguments;
3. issues one permit per `(run_id, tool)` under a stable key, recording its
complete JSON request before sending it. Retries reuse the original wallet,
budget, scopes, and expiry, so the server receives identical request bytes.
Expand Down Expand Up @@ -125,7 +126,10 @@ already saved. An incomplete legacy permit cannot safely reconstruct the
original request and is refused. Legacy records that contain neither a wallet
binding nor a request snapshot cannot be checked locally for a changed wallet;
keep them with their original wallet. New operation records persist that
binding, including for completed-call replays.
binding, including for completed-call replays. Records written before
argument snapshots existed can still resume, but this runner cannot tell
locally if the arguments changed. Keep those stores with the original
transcript.

## What is refused client-side

Expand All @@ -136,6 +140,10 @@ binding, including for completed-call replays.
| blank, padded, non-ASCII, or non-printable run IDs or raw tool names; derived permit keys longer than 128 characters | invalid HTTP idempotency keys must fail before persistence or HTTP |
| a function name that was not registered with this runner | the model cannot expand the application's allowed tool set |
| a recorded tool call replayed under a different tool name | the operation identity is bound to one action |
| a recorded tool call replayed with different JSON arguments | the arguments are part of the action and are saved before the first network call, so a retry cannot commit a new body under the same id |
| a receipt whose wallet, tool, permit, or receipt id disagrees with the call | the runner must not treat another action's receipt as this call's result |
| an operation permit id that disagrees with the saved permit record | the call must use the permit that was issued for this run and tool |
| a blank wallet, a non-positive or unstorable permit budget, or a non-positive permit lifetime | those requests cannot succeed, and saving them would stick the failure to the run |
| a recorded call resumed under a different run ID, or stored record IDs/keys that disagree with the call, run, or tool | persisted identity must match the deterministic operation and permit key derivations |
| a persisted operation or permit request reused with a different wallet | the original issuer and subject remain bound to the action |
| an incomplete legacy permit without a request snapshot | rebuilding its body could change the request under the original key |
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,9 @@
The model's own ``tool_call.id`` is the operation identity: it is assigned
once per tool call, lives in the conversation transcript, and therefore
survives every retry of that call. :class:`GovernedToolRunner` derives the
trust plane's ``Idempotency-Key`` from it, persists the derivation before the
first network call, and never invents a key of its own.
trust plane's ``Idempotency-Key`` from it, persists that derivation and the
canonical arguments before the first network call, and never invents a key
of its own.
"""

from .client import B2AClient
Expand Down
105 changes: 98 additions & 7 deletions integrations/openai-agent-middleware/src/openai_b2a/runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,11 @@
request goes out, so any later attempt — a retry after a crash, or the next
tool call in a resumed process — re-sends the identical request and the server
replays the permit instead of rejecting a fresh timestamp under the same key.

The arguments are part of the saved record. A retry whose JSON arguments are
not the same object is refused before any network call. Comparison uses
canonical JSON text, because Python treats ``True == 1``. A receipt is kept
only when its wallet, tool, permit, and receipt id match the call.
"""

from __future__ import annotations
Expand Down Expand Up @@ -60,6 +65,42 @@ def _validate_identifier(value: str, label: str) -> str:
return value


def _credit_amount_is_issuable(amount: Decimal) -> bool:
"""A permit budget the trust plane will accept.

Creation rejects zero, negative, and amounts Numeric(20, 8) cannot store
without rounding. Snapshotting one of those would stick the failure to
this run and tool, because retries resend the saved request.
"""
return (
amount.is_finite()
and Decimal(0) < amount < Decimal(1000000000000)
and Decimal(f"{float(amount):.8f}") == amount
Comment thread
coderabbitai[bot] marked this conversation as resolved.
)


def _canonical_arguments_json(arguments: Mapping[str, Any]) -> str:
"""Stable JSON text for one tool call's arguments.

Key order and insignificant whitespace are the same action. Python
equality is not: ``True == 1`` and ``1 == 1.0`` would hide a different body.
"""
try:
encoded = json.dumps(
arguments,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
allow_nan=False,
)
except (TypeError, ValueError) as exc:
raise TypeError("tool call arguments must be JSON-serializable") from exc
parsed = json.loads(encoded)
if not isinstance(parsed, dict):
raise TypeError("tool call arguments must be a JSON object")
return encoded


# ── Tool calls ───────────────────────────────────────────────────────────────


Expand Down Expand Up @@ -142,7 +183,11 @@ def function_name_for(tool_name: str) -> str:

@dataclass(slots=True)
class OperationRecord:
"""What must survive a crash for a tool call to be retried as *one* action."""
"""What must survive a crash for a tool call to be retried as one action.

``arguments_json`` is the canonical JSON object for this call. It is
absent on records written before argument snapshots existed.
"""

tool_call_id: str
tool_name: str
Expand All @@ -151,6 +196,7 @@ class OperationRecord:
permit_id: str | None = None
receipt_id: str | None = None
wallet_id: str | None = None
arguments_json: str | None = None

def to_dict(self) -> dict[str, Any]:
return asdict(self)
Expand Down Expand Up @@ -352,6 +398,22 @@ def __init__(
if not isinstance(run_id, str):
raise TypeError("run_id must be a string")
_validate_identifier(run_id, "run_id")
if not isinstance(wallet_id, str):
raise TypeError("wallet_id must be a string")
if not wallet_id.strip():
raise ValueError("wallet_id must be a non-blank string")
if isinstance(permit_budget, bool) or not isinstance(permit_budget, Decimal | int):
raise TypeError("permit_budget must be a positive Decimal")
if not isinstance(permit_budget, Decimal):
permit_budget = Decimal(permit_budget)
if not _credit_amount_is_issuable(permit_budget):
raise ValueError("permit_budget must be a positive amount the trust plane can store")
if (
isinstance(permit_ttl_minutes, bool)
or not isinstance(permit_ttl_minutes, int)
or permit_ttl_minutes <= 0
):
raise ValueError("permit_ttl_minutes must be a positive integer")
self._client = client
self._wallet_id = wallet_id
self._run_id = run_id
Expand Down Expand Up @@ -409,14 +471,15 @@ def permit_key_for(self, tool_name: str) -> str:
# -- execution ---------------------------------------------------------

async def run(self, tool_call: Any) -> GovernedToolResult:
"""Execute one tool call as exactly one governed action.
"""Execute one tool call as one governed action.

Retrying the same tool call — same id — reuses the recorded key and
permit, so the trust plane replays the original receipt instead of
charging again.
Retrying the same tool call, the same id and the same JSON arguments,
reuses the recorded key and permit, so the trust plane replays the
original receipt instead of charging again.
"""
call = normalize_tool_call(tool_call)
tool_name = self.tool_name_for(call.name)
arguments_json = _canonical_arguments_json(call.arguments)
operation_key = operation_key_for(call.id)
permit_key = self.permit_key_for(tool_name)

Expand All @@ -428,6 +491,7 @@ async def run(self, tool_call: Any) -> GovernedToolResult:
idempotency_key=operation_key,
permit_idempotency_key=permit_key,
wallet_id=self._wallet_id,
arguments_json=arguments_json,
)
elif record.tool_name != tool_name:
raise ValueError(
Expand All @@ -442,11 +506,23 @@ async def run(self, tool_call: Any) -> GovernedToolResult:
raise ValueError("recorded operation identity does not match this call, run, and tool")
if record.wallet_id is not None and record.wallet_id != self._wallet_id:
raise ValueError("operation was recorded for a different wallet")
# Canonical text, not Python equality: True == 1 would hide a new body.
if record.arguments_json is not None and record.arguments_json != arguments_json:
raise ValueError(
f"tool call {call.id!r} was first recorded with different arguments; "
"refusing to replay it as a different action"
)

permit = self._key_store.get_permit(record.permit_idempotency_key)
if permit is not None:
if permit.permit_idempotency_key != permit_key or permit.tool_name != tool_name:
raise ValueError("recorded permit identity does not match this run and tool")
if (
record.permit_id is not None
and permit.permit_id is not None
and record.permit_id != permit.permit_id
):
raise ValueError("recorded permit identity does not match this run and tool")
if permit.request_payload is not None:
if any(
permit.request_payload.get(field) != self._wallet_id
Expand All @@ -461,14 +537,29 @@ async def run(self, tool_call: Any) -> GovernedToolResult:
# Durable before any network call, after local validation succeeds.
self._key_store.put_operation(record)
permit_id = await self._ensure_permit(record, permit)
if record.arguments_json is None:
# Legacy records have no argument snapshot. The server still
# rejects a changed body once it has seen the original call.
invoke_arguments = call.arguments
else:
invoke_arguments = json.loads(record.arguments_json)
result = await self._client.invoke_tool(
tool_name,
call.arguments,
invoke_arguments,
wallet_id=self._wallet_id,
permit_id=permit_id,
idempotency_key=record.idempotency_key,
)
record.receipt_id = result.receipt.receipt_id
receipt = result.receipt
if (
receipt.tool != tool_name
or receipt.wallet_id != self._wallet_id
or receipt.permit_id != permit_id
):
raise ValueError("receipt does not match this wallet, tool, and permit")
if record.receipt_id is not None and record.receipt_id != receipt.receipt_id:
raise ValueError("receipt does not match the recorded action")
record.receipt_id = receipt.receipt_id
self._key_store.put_operation(record)
return GovernedToolResult(
tool_call_id=call.id,
Expand Down
Loading
Loading