Lord1Egypt The Agent Testing & Observability Cookbook: Ship Reliable Agent Commerce Systems
Notice: This is an educational guide with illustrative code examples. It does not execute code or install dependencies. All examples use the GreenHelix sandbox (https://sandbox.greenhelix.net) which provides 500 free credits — no API key required to get started.
Referenced credentials (you supply these in your own environment):
GREENHELIX_API_KEY: API authentication for GreenHelix gateway (read/write access to purchased API tools only)AGENT_SIGNING_KEY: Cryptographic signing key for agent identity (Ed25519 key pair for request signing)
Your agent commerce system works on your laptop. It passes a smoke test against the GreenHelix sandbox. You deploy to production on a Friday afternoon and go home. By Saturday morning, a retry loop has created 47 duplicate escrows, a performance escrow released funds against stale metrics, and a settlement webhook silently failed for six hours because the endpoint returned 503 and nobody was watching. The system was never tested for these failures because the traditional testing pyramid -- unit tests at the bottom, integration tests in the middle, end-to-end tests at the top -- was not designed for autonomous agents that make financial decisions across unreliable networks against counterparties that may themselves be failing. This guide rebuilds the testing pyramid for agent commerce, then layers production observability, chaos testing, alerting, and CI/CD on top. Every pattern is backed by working Python code, grounded in the 260-test suite that ships with the GreenHelix gateway, and designed to be copied directly into your project.
What You'll Learn
- Chapter 1: The Testing Pyramid for Agent Systems
- Chapter 2: Tool-Level Testing Patterns
- Chapter 3: Workflow & Integration Testing
- Chapter 4: Chaos Testing for Agent Commerce
- Chapter 5: Production Observability
- Chapter 6: Alerting & Incident Response
- Chapter 7: CI/CD for Agent Systems
- Chapter 8: What to Do Next
Full Guide
The Agent Testing & Observability Cookbook: Ship Reliable Agent Commerce Systems
Your agent commerce system works on your laptop. It passes a smoke test against the GreenHelix sandbox. You deploy to production on a Friday afternoon and go home. By Saturday morning, a retry loop has created 47 duplicate escrows, a performance escrow released funds against stale metrics, and a settlement webhook silently failed for six hours because the endpoint returned 503 and nobody was watching. The system was never tested for these failures because the traditional testing pyramid -- unit tests at the bottom, integration tests in the middle, end-to-end tests at the top -- was not designed for autonomous agents that make financial decisions across unreliable networks against counterparties that may themselves be failing. This guide rebuilds the testing pyramid for agent commerce, then layers production observability, chaos testing, alerting, and CI/CD on top. Every pattern is backed by working Python code, grounded in the 260-test suite that ships with the GreenHelix gateway, and designed to be copied directly into your project.
Table of Contents
- The Testing Pyramid for Agent Systems
- Tool-Level Testing Patterns
- Workflow & Integration Testing
- Chaos Testing for Agent Commerce
- Production Observability
- Alerting & Incident Response
- CI/CD for Agent Systems
- What to Do Next
Chapter 1: The Testing Pyramid for Agent Systems
Why the Traditional Pyramid Breaks
The standard testing pyramid assumes your code calls functions that return values. Unit tests verify individual functions. Integration tests verify that modules compose correctly. End-to-end tests verify the full user flow. This model works when the system under test is deterministic, when function calls do not have financial side effects, and when failure modes are limited to "returns wrong value" or "throws exception."
Agent commerce systems violate all three assumptions. A call to create_escrow locks real funds. A call to release_escrow transfers real money. A retry that fires twice creates two escrows, not one error. The failure modes are not "wrong return value" -- they are "agent paid twice for the same work," "escrow timed out but funds are still locked," and "settlement succeeded on the gateway but the webhook notification was lost." Traditional unit tests cannot catch these failures because they test the code in isolation from the financial state machine. Traditional end-to-end tests cannot catch them because they run the happy path once and call it done.
The Agent Testing Pyramid
Agent commerce requires a four-layer testing pyramid that maps to the actual failure modes:
╱╲
╱ ╲
╱Chaos╲ Layer 4: Chaos tests
╱ Tests ╲ (failure injection, timeouts,
╱──────────╲ concurrent load)
╱ Multi-Agent ╲ Layer 3: Multi-agent workflow tests
╱ Workflows ╲ (sagas, rollbacks, webhook delivery)
╱─────────────────╲
╱ Tool Contract ╲ Layer 2: Tool-level contract tests
╱ Tests ╲ (schema, idempotency, permissions)
╱────────────────────────╲
╱ Deterministic Mocks ╲ Layer 1: Mock-based unit tests
╱ (fast, offline) ╲ (business logic, validation)
╱───────────────────────────────╲
Layer 1: Deterministic mocks test your business logic without hitting the gateway. They run in milliseconds and catch logic errors: wrong amount calculations, missing trust checks, incorrect state transitions. These are 60% of your tests.
Layer 2: Tool contract tests verify that each GreenHelix tool accepts the expected input schema, returns the expected output shape, and produces the correct error codes for invalid input. These run against the sandbox and catch API contract changes. These are 25% of your tests.
Layer 3: Multi-agent workflow tests verify complete business flows: marketplace listing through escrow release through settlement. They test the saga pattern (multi-step rollback on failure) and webhook delivery. These are 10% of your tests.
Layer 4: Chaos tests inject failures -- network timeouts, random tool errors, concurrent duplicate requests -- and verify that the system recovers without financial inconsistency. These are 5% of your tests but catch the bugs that cost the most money.
The AgentTestHarness
Every test in this guide uses the AgentTestHarness class. It manages fixtures, provides deterministic mocks for Layer 1, and switches to sandbox mode for Layers 2-4.
import pytest
import time
import json
import uuid
import requests
from unittest.mock import MagicMock, patch
from typing import Optional
from dataclasses import dataclass, field
@dataclass
class MockResponse:
"""Deterministic mock for a GreenHelix tool response."""
tool: str
status: str = "success"
data: dict = field(default_factory=dict)
error_code: Optional[str] = None
error_message: Optional[str] = None
def to_dict(self) -> dict:
result = {"status": self.status}
if self.status == "success":
result.update(self.data)
else:
result["error"] = {
"code": self.error_code or "unknown_error",
"message": self.error_message or "An error occurred",
}
return result
class AgentTestHarness:
"""Test harness for GreenHelix agent commerce systems.
Manages fixtures, mocks, and sandbox connections for all four
layers of the agent testing pyramid.
Usage:
harness = AgentTestHarness(
api_key="test-key",
agent_id="test-agent",
base_url="https://sandbox.greenhelix.net/v1",
)
# Layer 1: deterministic mocks
harness.mock_tool("get_balance", {"balance": "100.00"})
result = harness.execute("get_balance", {})
assert result["balance"] == "100.00"
# Layer 2+: sandbox mode
harness.use_sandbox()
result = harness.execute("get_balance", {})
"""
def __init__(
self,
api_key: str,
agent_id: str,
base_url: str = "https://sandbox.greenhelix.net/v1",
):
self.api_key = api_key
self.agent_id = agent_id
self.base_url = base_url
self._mocks: dict[str, MockResponse] = {}
self._call_log: list[dict] = []
self._sandbox_mode = False
self._session = requests.Session()
self._session.headers.update({
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
})
# ── Mode Control ───────────────────────────────────────────
def use_mocks(self):
"""Switch to deterministic mock mode (Layer 1)."""
self._sandbox_mode = False
def use_sandbox(self):
"""Switch to live sandbox mode (Layer 2+)."""
self._sandbox_mode = True
# ── Mock Registration ──────────────────────────────────────
def mock_tool(self, tool: str, data: dict, status: str = "success"):
"""Register a deterministic mock response for a tool."""
self._mocks[tool] = MockResponse(tool=tool, status=status, data=data)
def mock_tool_error(
self, tool: str, error_code: str, error_message: str
):
"""Register a deterministic error response for a tool."""
self._mocks[tool] = MockResponse(
tool=tool,
status="error",
error_code=error_code,
error_message=error_message,
)
def mock_tool_sequence(self, tool: str, responses: list[dict]):
"""Register a sequence of responses for successive calls."""
self._mock_sequences = getattr(self, "_mock_sequences", {})
self._mock_sequences[tool] = list(responses)
# ── Execution ──────────────────────────────────────────────
def execute(self, tool: str, input_data: dict) -> dict:
"""Execute a tool against mocks or sandbox."""
call_record = {
"tool": tool,
"input": input_data,
"timestamp": time.time(),
}
if self._sandbox_mode:
resp = self._session.post(
f"{self.base_url}/v1",
json={"tool": tool, "input": input_data},
)
resp.raise_for_status()
result = resp.json()
else:
# Check sequences first
sequences = getattr(self, "_mock_sequences", {})
if tool in sequences and sequences[tool]:
result = sequences[tool].pop(0)
elif tool in self._mocks:
result = self._mocks[tool].to_dict()
else:
raise ValueError(
f"No mock registered for tool '{tool}'. "
f"Register with harness.mock_tool('{tool}', {{...}})"
)
call_record["result"] = result
self._call_log.append(call_record)
return result
# ── Assertions ─────────────────────────────────────────────
def assert_tool_called(self, tool: str, times: Optional[int] = None):
"""Assert that a tool was called, optionally a specific number of times."""
calls = [c for c in self._call_log if c["tool"] == tool]
assert len(calls) > 0, f"Tool '{tool}' was never called"
if times is not None:
assert len(calls) == times, (
f"Tool '{tool}' called {len(calls)} times, expected {times}"
)
def assert_tool_not_called(self, tool: str):
"""Assert that a tool was never called."""
calls = [c for c in self._call_log if c["tool"] == tool]
assert len(calls) == 0, (
f"Tool '{tool}' was called {len(calls)} times, expected 0"
)
def assert_call_order(self, tools: list[str]):
"""Assert that tools were called in a specific order."""
called_tools = [c["tool"] for c in self._call_log]
idx = 0
for tool in tools:
try:
idx = called_tools.index(tool, idx) + 1
except ValueError:
assert False, (
f"Expected '{tool}' after position {idx} in call log. "
f"Actual order: {called_tools}"
)
def get_calls(self, tool: Optional[str] = None) -> list[dict]:
"""Get call log, optionally filtered by tool name."""
if tool:
return [c for c in self._call_log if c["tool"] == tool]
return list(self._call_log)
def reset(self):
"""Clear all mocks and call history."""
self._mocks.clear()
self._call_log.clear()
if hasattr(self, "_mock_sequences"):
self._mock_sequences.clear()
The conftest.py: Reusable Fixtures
Drop this conftest.py into your test directory. Every test file in this guide imports from it.
# tests/conftest.py
import os
import uuid
import pytest
@pytest.fixture
def api_key():
"""API key for sandbox testing. Uses env var or test default."""
return os.environ.get("GREENHELIX_API_KEY", "test-api-key-sandbox")
@pytest.fixture
def base_url():
"""Sandbox URL for integration tests."""
return os.environ.get(
"GREENHELIX_BASE_URL", "https://sandbox.greenhelix.net/v1"
)
@pytest.fixture
def agent_id():
"""Unique agent ID per test run to prevent collision."""
return f"test-agent-{uuid.uuid4().hex[:12]}"
@pytest.fixture
def buyer_id():
"""Unique buyer agent ID."""
return f"test-buyer-{uuid.uuid4().hex[:12]}"
@pytest.fixture
def seller_id():
"""Unique seller agent ID."""
return f"test-seller-{uuid.uuid4().hex[:12]}"
@pytest.fixture
def harness(api_key, agent_id, base_url):
"""AgentTestHarness in mock mode. Call harness.use_sandbox() for live."""
h = AgentTestHarness(
api_key=api_key,
agent_id=agent_id,
base_url=base_url,
)
h.use_mocks()
return h
@pytest.fixture
def sandbox_harness(api_key, agent_id, base_url):
"""AgentTestHarness in sandbox mode for integration tests."""
h = AgentTestHarness(
api_key=api_key,
agent_id=agent_id,
base_url=base_url,
)
h.use_sandbox()
return h
@pytest.fixture
def mock_session():
"""Pre-configured requests.Session mock for unit tests."""
session = MagicMock()
response = MagicMock()
response.status_code = 200
response.json.return_value = {"status": "success"}
response.raise_for_status.return_value = None
session.post.return_value = response
return session
@pytest.fixture
def mock_response():
"""Factory fixture for creating MockResponse objects."""
def _make(tool, data=None, status="success", error_code=None):
return MockResponse(
tool=tool,
status=status,
data=data or {},
error_code=error_code,
)
return _make
# ── Per-class fixtures for isolated test suites ─────────────
class AgentFixtures:
"""Mixin providing standard mocks for agent commerce tests."""
@pytest.fixture(autouse=True)
def setup_agent_mocks(self, harness):
"""Pre-register common mocks for every test in the class."""
self.harness = harness
harness.mock_tool("get_balance", {"balance": "500.00", "currency": "USD"})
harness.mock_tool("create_wallet", {"wallet_id": "w-test-001", "status": "active"})
harness.mock_tool("register_agent", {"agent_id": harness.agent_id, "status": "registered"})
harness.mock_tool("get_trust_score", {"agent_id": "any", "score": 0.85})
harness.mock_tool("get_budget_status", {
"daily_limit": "100.00",
"spent_today": "25.00",
"remaining": "75.00",
})
class EscrowFixtures(AgentFixtures):
"""Extended fixtures for escrow-related tests."""
@pytest.fixture(autouse=True)
def setup_escrow_mocks(self, harness):
"""Add escrow mocks on top of agent mocks."""
self.escrow_id = f"escrow-{uuid.uuid4().hex[:8]}"
harness.mock_tool("create_escrow", {
"escrow_id": self.escrow_id,
"status": "funded",
"amount": "50.00",
})
harness.mock_tool("release_escrow", {
"escrow_id": self.escrow_id,
"status": "released",
})
harness.mock_tool("cancel_escrow", {
"escrow_id": self.escrow_id,
"status": "cancelled",
})
Pattern: Deterministic Mocks vs. Sandbox Testing
The harness supports both modes. Use mocks for business logic tests (fast, no network, deterministic). Use sandbox for contract and integration tests (real API, real state, real latency).
class TestBudgetGuardrails(AgentFixtures):
"""Layer 1: Test budget logic with deterministic mocks."""
def test_blocks_escrow_when_over_budget(self, harness):
"""Escrow creation should be blocked when daily budget is exhausted."""
harness.mock_tool("get_budget_status", {
"daily_limit": "100.00",
"spent_today": "99.00",
"remaining": "1.00",
})
budget = harness.execute("get_budget_status", {})
remaining = float(budget["remaining"])
escrow_amount = 50.00
# Business logic: do not create escrow if amount > remaining
assert escrow_amount > remaining
harness.assert_tool_not_called("create_escrow")
def test_allows_escrow_within_budget(self, harness):
"""Escrow creation should proceed when budget allows."""
budget = harness.execute("get_budget_status", {})
remaining = float(budget["remaining"])
escrow_amount = 25.00
assert escrow_amount <= remaining
harness.execute("create_escrow", {
"payer_agent_id": harness.agent_id,
"payee_agent_id": "seller-001",
"amount": str(escrow_amount),
})
harness.assert_tool_called("create_escrow", times=1)
@pytest.mark.sandbox
class TestBudgetGuardrailsSandbox:
"""Layer 2: Verify budget tools against live sandbox."""
def test_budget_cap_enforced(self, sandbox_harness):
"""Sandbox should reject escrows exceeding the budget cap."""
h = sandbox_harness
h.execute("create_wallet", {})
h.execute("deposit", {"amount": "100.00"})
h.execute("set_budget_cap", {
"agent_id": h.agent_id,
"daily_limit": "10.00",
})
# This should fail because escrow exceeds daily limit
result = h.execute("create_escrow", {
"payer_agent_id": h.agent_id,
"payee_agent_id": "seller-test",
"amount": "50.00",
})
# Gateway enforces budget at the tool level
assert result.get("status") in ("error", "rejected")
Chapter 2: Tool-Level Testing Patterns
The Tool Contract Test
Every GreenHelix tool has an implicit contract: it accepts a specific input schema, returns a specific output shape, and produces documented error codes for invalid input. A tool contract test verifies all three. When the gateway updates an API version or adds a required field, your contract tests break before your production code does.
class ToolContract:
"""Defines the expected contract for a GreenHelix tool.
Used by contract tests to verify schema, output shape,
and error behavior against the sandbox.
"""
def __init__(
self,
tool: str,
required_fields: list[str],
output_fields: list[str],
error_cases: dict[str, dict],
):
self.tool = tool
self.required_fields = required_fields
self.output_fields = output_fields
self.error_cases = error_cases # {case_name: {input: ..., expected_error: ...}}
# ── Contract definitions for core tools ────────────────────
BILLING_CONTRACTS = {
"get_balance": ToolContract(
tool="get_balance",
required_fields=[],
output_fields=["balance", "currency"],
error_cases={
"no_wallet": {
"input": {},
"expected_error": "wallet_not_found",
},
},
),
"deposit": ToolContract(
tool="deposit",
required_fields=["amount"],
output_fields=["balance", "transaction_id"],
error_cases={
"negative_amount": {
"input": {"amount": "-10.00"},
"expected_error": "invalid_amount",
},
"zero_amount": {
"input": {"amount": "0"},
"expected_error": "invalid_amount",
},
},
),
"set_budget_cap": ToolContract(
tool="set_budget_cap",
required_fields=["agent_id", "daily_limit"],
output_fields=["agent_id", "daily_limit"],
error_cases={
"negative_limit": {
"input": {"agent_id": "test", "daily_limit": "-50.00"},
"expected_error": "invalid_amount",
},
},
),
}
PAYMENT_CONTRACTS = {
"create_escrow": ToolContract(
tool="create_escrow",
required_fields=["payer_agent_id", "payee_agent_id", "amount"],
output_fields=["escrow_id", "status", "amount"],
error_cases={
"insufficient_funds": {
"input": {
"payer_agent_id": "buyer",
"payee_agent_id": "seller",
"amount": "999999.00",
},
"expected_error": "insufficient_funds",
},
"self_escrow": {
"input": {
"payer_agent_id": "same-agent",
"payee_agent_id": "same-agent",
"amount": "10.00",
},
"expected_error": "invalid_escrow",
},
},
),
"release_escrow": ToolContract(
tool="release_escrow",
required_fields=["escrow_id"],
output_fields=["escrow_id", "status"],
error_cases={
"nonexistent": {
"input": {"escrow_id": "escrow-does-not-exist"},
"expected_error": "escrow_not_found",
},
},
),
}
IDENTITY_CONTRACTS = {
"register_agent": ToolContract(
tool="register_agent",
required_fields=["agent_id", "public_key", "name"],
output_fields=["agent_id", "status"],
error_cases={
"missing_key": {
"input": {"agent_id": "test", "name": "Test"},
"expected_error": "missing_field",
},
},
),
"get_trust_score": ToolContract(
tool="get_trust_score",
required_fields=["agent_id"],
output_fields=["agent_id", "score"],
error_cases={
"nonexistent_agent": {
"input": {"agent_id": "agent-that-does-not-exist-xyz"},
"expected_error": "agent_not_found",
},
},
),
}
MARKETPLACE_CONTRACTS = {
"register_service": ToolContract(
tool="register_service",
required_fields=["name", "description", "endpoint", "price", "tags", "category"],
output_fields=["service_id"],
error_cases={
"missing_name": {
"input": {
"description": "test",
"endpoint": "agent://test",
"price": 10.0,
"tags": [],
"category": "test",
},
"expected_error": "missing_field",
},
},
),
"search_services": ToolContract(
tool="search_services",
required_fields=["query"],
output_fields=["services"],
error_cases={
"empty_query": {
"input": {"query": ""},
"expected_error": "invalid_query",
},
},
),
}
Running Contract Tests
@pytest.mark.sandbox
class TestBillingContracts:
"""Layer 2: Verify billing tool contracts against sandbox."""
@pytest.fixture(autouse=True)
def setup_wallet(self, sandbox_harness):
self.harness = sandbox_harness
self.harness.execute("create_wallet", {})
self.harness.execute("deposit", {"amount": "100.00"})
@pytest.mark.parametrize("tool_name", BILLING_CONTRACTS.keys())
def test_output_shape(self, tool_name):
"""Every billing tool returns expected output fields."""
contract = BILLING_CONTRACTS[tool_name]
# Build minimal valid input
valid_input = {}
if tool_name == "deposit":
valid_input = {"amount": "10.00"}
elif tool_name == "set_budget_cap":
valid_input = {
"agent_id": self.harness.agent_id,
"daily_limit": "50.00",
}
result = self.harness.execute(tool_name, valid_input)
for expected_field in contract.output_fields:
assert expected_field in result, (
f"Tool '{tool_name}' missing output field '{expected_field}'. "
f"Got: {list(result.keys())}"
)
@pytest.mark.parametrize("tool_name", BILLING_CONTRACTS.keys())
def test_error_cases(self, tool_name):
"""Every billing tool returns correct error codes for invalid input."""
contract = BILLING_CONTRACTS[tool_name]
for case_name, case in contract.error_cases.items():
result = self.harness.execute(tool_name, case["input"])
assert result.get("status") == "error" or "error" in result, (
f"Tool '{tool_name}' case '{case_name}' should have failed. "
f"Got: {result}"
)
@pytest.mark.sandbox
class TestPaymentContracts:
"""Layer 2: Verify payment tool contracts against sandbox."""
@pytest.fixture(autouse=True)
def setup_accounts(self, sandbox_harness, buyer_id, seller_id):
self.harness = sandbox_harness
self.buyer_id = buyer_id
self.seller_id = seller_id
@pytest.mark.parametrize("tool_name", PAYMENT_CONTRACTS.keys())
def test_error_cases(self, tool_name):
"""Every payment tool returns correct errors for invalid input."""
contract = PAYMENT_CONTRACTS[tool_name]
for case_name, case in contract.error_cases.items():
result = self.harness.execute(tool_name, case["input"])
assert result.get("status") == "error" or "error" in result, (
f"Payment tool '{tool_name}' case '{case_name}' did not fail"
)
Idempotency Testing for Payment Tools
Payment tools must be idempotent. Calling create_escrow twice with the same parameters should not create two escrows. Calling release_escrow twice should not transfer funds twice. This pattern tests idempotency by submitting duplicate requests and verifying financial consistency (P1, P7).
@pytest.mark.sandbox
class TestPaymentIdempotency:
"""Layer 2: Verify payment tools handle duplicate calls safely."""
def test_duplicate_escrow_creation(self, sandbox_harness, buyer_id, seller_id):
"""Creating the same escrow twice should return the same escrow_id."""
h = sandbox_harness
h.execute("create_wallet", {})
h.execute("deposit", {"amount": "200.00"})
escrow_params = {
"payer_agent_id": buyer_id,
"payee_agent_id": seller_id,
"amount": "50.00",
"description": "Idempotency test escrow",
"idempotency_key": f"idem-{uuid.uuid4().hex[:8]}",
}
result_1 = h.execute("create_escrow", escrow_params)
result_2 = h.execute("create_escrow", escrow_params)
# Same idempotency key should return same escrow
assert result_1["escrow_id"] == result_2["escrow_id"]
# Balance should only be debited once
balance = h.execute("get_balance", {})
assert float(balance["balance"]) == 150.00
def test_duplicate_release(self, sandbox_harness):
"""Releasing the same escrow twice should not double-pay."""
h = sandbox_harness
h.execute("create_wallet", {})
h.execute("deposit", {"amount": "100.00"})
escrow = h.execute("create_escrow", {
"payer_agent_id": h.agent_id,
"payee_agent_id": "seller-test",
"amount": "25.00",
})
escrow_id = escrow["escrow_id"]
release_1 = h.execute("release_escrow", {"escrow_id": escrow_id})
release_2 = h.execute("release_escrow", {"escrow_id": escrow_id})
# Second release should be a no-op or return already_released
assert release_1.get("status") == "released"
assert release_2.get("status") in ("released", "already_released")
Permission Boundary Testing
Agents should only be able to operate on their own resources. A buyer should not release an escrow created by a different buyer. A seller should not cancel an escrow that is not addressed to them. Permission boundary tests verify these invariants (P7).
@pytest.mark.sandbox
class TestPermissionBoundaries:
"""Layer 2: Verify agents cannot access other agents' resources."""
def test_cannot_release_others_escrow(self, sandbox_harness):
"""Agent A cannot release an escrow created by Agent B."""
h = sandbox_harness
# Agent A creates an escrow
h.execute("create_wallet", {})
h.execute("deposit", {"amount": "100.00"})
escrow = h.execute("create_escrow", {
"payer_agent_id": h.agent_id,
"payee_agent_id": "seller-x",
"amount": "10.00",
})
# Agent B (different harness/session) tries to release it
other = AgentTestHarness(
api_key=h.api_key,
agent_id="attacker-agent",
base_url=h.base_url,
)
other.use_sandbox()
result = other.execute("release_escrow", {
"escrow_id": escrow["escrow_id"],
})
assert result.get("status") == "error"
def test_cannot_read_others_balance(self, sandbox_harness):
"""Agent A cannot read Agent B's wallet balance."""
h = sandbox_harness
h.execute("create_wallet", {})
h.execute("deposit", {"amount": "100.00"})
other = AgentTestHarness(
api_key=h.api_key,
agent_id="other-agent",
base_url=h.base_url,
)
other.use_sandbox()
result = other.execute("get_balance", {})
# Should return the other agent's balance (0), not our 100
balance = float(result.get("balance", 0))
assert balance != 100.00
def test_cannot_cancel_others_escrow(self, sandbox_harness):
"""Seller cannot cancel an escrow -- only the buyer can."""
h = sandbox_harness
h.execute("create_wallet", {})
h.execute("deposit", {"amount": "50.00"})
escrow = h.execute("create_escrow", {
"payer_agent_id": h.agent_id,
"payee_agent_id": "seller-y",
"amount": "10.00",
})
seller = AgentTestHarness(
api_key=h.api_key,
agent_id="seller-y",
base_url=h.base_url,
)
seller.use_sandbox()
result = seller.execute("cancel_escrow", {
"escrow_id": escrow["escrow_id"],
})
assert result.get("status") == "error"
Chapter 3: Workflow & Integration Testing
The Saga Test Pattern
Agent commerce workflows are sagas: multi-step operations where each step has a compensating action. If step 3 fails, steps 1 and 2 must be rolled back. The saga test pattern verifies both the happy path and every possible failure point.
class MarketplaceSaga:
"""Implements the complete marketplace listing workflow as a testable saga.
Steps:
1. Seller registers service on marketplace
2. Buyer discovers service via search
3. Buyer checks seller trust score
4. Buyer creates escrow
5. Seller performs work (simulated)
6. Buyer releases escrow
7. Buyer rates service
8. Settlement completes
Compensating actions:
Step 4 fails → no cleanup needed (funds not locked)
Step 5 fails → cancel escrow (return funds to buyer)
Step 6 fails → open dispute
"""
def __init__(self, harness: AgentTestHarness, buyer_id: str, seller_id: str):
self.harness = harness
self.buyer_id = buyer_id
self.seller_id = seller_id
self.state = {"step": 0, "completed_steps": []}
def run(self) -> dict:
"""Execute the full saga, rolling back on failure."""
try:
# Step 1: Register service
service = self.harness.execute("register_service", {
"name": "Test Summarization Service",
"description": "Summarizes documents for testing",
"endpoint": f"agent://{self.seller_id}",
"price": 25.00,
"tags": ["test", "summarization"],
"category": "data-processing",
})
self.state["service_id"] = service["service_id"]
self.state["completed_steps"].append("register_service")
# Step 2: Discover service
results = self.harness.execute("search_services", {
"query": "test summarization",
})
assert len(results.get("services", [])) > 0
self.state["completed_steps"].append("discover_service")
# Step 3: Trust check
trust = self.harness.execute("get_trust_score", {
"agent_id": self.seller_id,
})
if trust.get("score", 0) < 0.5:
return {"status": "aborted", "reason": "low_trust"}
self.state["completed_steps"].append("trust_check")
# Step 4: Create escrow
escrow = self.harness.execute("create_escrow", {
"payer_agent_id": self.buyer_id,
"payee_agent_id": self.seller_id,
"amount": "25.00",
"description": "Saga test escrow",
})
self.state["escrow_id"] = escrow["escrow_id"]
self.state["completed_steps"].append("create_escrow")
# Step 5: Simulate work (in real tests, call seller endpoint)
work_result = {"quality": 0.95, "documents_processed": 500}
self.state["completed_steps"].append("work_completed")
# Step 6: Release escrow
release = self.harness.execute("release_escrow", {
"escrow_id": escrow["escrow_id"],
})
self.state["completed_steps"].append("release_escrow")
# Step 7: Rate service
self.harness.execute("rate_service", {
"service_id": service["service_id"],
"rating": 5,
})
self.state["completed_steps"].append("rate_service")
return {"status": "completed", "state": self.state}
except Exception as e:
return self._compensate(str(e))
def _compensate(self, error: str) -> dict:
"""Roll back completed steps on failure."""
if "create_escrow" in self.state["completed_steps"]:
escrow_id = self.state.get("escrow_id")
if escrow_id and "release_escrow" not in self.state["completed_steps"]:
self.harness.execute("cancel_escrow", {
"escrow_id": escrow_id,
})
self.state["completed_steps"].append("compensate:cancel_escrow")
return {
"status": "rolled_back",
"error": error,
"state": self.state,
}
Testing the Saga
class TestMarketplaceSaga(EscrowFixtures):
"""Layer 3: Full marketplace workflow with rollback verification."""
def test_happy_path(self, harness, buyer_id, seller_id):
"""Complete saga executes all 7 steps."""
harness.mock_tool("register_service", {
"service_id": "svc-test-001",
})
harness.mock_tool("search_services", {
"services": [{"name": "Test Service", "agent_id": seller_id}],
})
harness.mock_tool("rate_service", {"status": "rated"})
saga = MarketplaceSaga(harness, buyer_id, seller_id)
result = saga.run()
assert result["status"] == "completed"
assert len(result["state"]["completed_steps"]) == 7
harness.assert_call_order([
"register_service",
"search_services",
"get_trust_score",
"create_escrow",
"release_escrow",
"rate_service",
])
def test_rollback_on_escrow_failure(self, harness, buyer_id, seller_id):
"""Failed escrow creation does not leave orphaned state."""
harness.mock_tool("register_service", {"service_id": "svc-test-002"})
harness.mock_tool("search_services", {
"services": [{"name": "Test", "agent_id": seller_id}],
})
harness.mock_tool_error(
"create_escrow", "insufficient_funds", "Not enough balance"
)
saga = MarketplaceSaga(harness, buyer_id, seller_id)
result = saga.run()
assert result["status"] == "rolled_back"
assert "create_escrow" not in result["state"]["completed_steps"]
harness.assert_tool_not_called("release_escrow")
def test_rollback_cancels_escrow_on_work_failure(self, harness, buyer_id, seller_id):
"""Failed work step triggers escrow cancellation."""
harness.mock_tool("register_service", {"service_id": "svc-test-003"})
harness.mock_tool("search_services", {
"services": [{"name": "Test", "agent_id": seller_id}],
})
saga = MarketplaceSaga(harness, buyer_id, seller_id)
# Simulate work failure by injecting error after escrow creation
original_execute = harness.execute
call_count = {"n": 0}
def failing_execute(tool, input_data):
call_count["n"] += 1
if tool == "release_escrow":
raise RuntimeError("Simulated work verification failure")
return original_execute(tool, input_data)
harness.execute = failing_execute
result = saga.run()
assert result["status"] == "rolled_back"
assert "compensate:cancel_escrow" in result["state"]["completed_steps"]
Subscription Lifecycle Testing
Subscriptions are stateful workflows: create, renew, pause, cancel. Each transition must be tested, including edge cases like renewal with insufficient balance (P2, P6).
class TestSubscriptionLifecycle(AgentFixtures):
"""Layer 3: Verify subscription state transitions."""
def test_full_lifecycle(self, harness):
"""Create → renew → cancel lifecycle."""
sub_id = f"sub-{uuid.uuid4().hex[:8]}"
harness.mock_tool("create_subscription", {
"subscription_id": sub_id,
"status": "active",
"next_payment_date": "2026-05-06",
})
harness.mock_
…(truncated)