The Formal Gatekeeper: Z3-Verified Safety for Autonomous Agent Plans#

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)
• WALLET_ADDRESS: Blockchain wallet address for receiving payments (public address only — no private keys)
• AGENT_SIGNING_KEY: Cryptographic signing key for agent identity (Ed25519 key pair for request signing)

Your cat-sitting agent just ordered a Tesla. Not because it was hacked, not because of a prompt injection, but because the large language model driving it reasoned -- correctly, within its context window -- that a Tesla Model Y would be the most efficient way to transport fourteen cats to the veterinarian for their annual checkups. The purchase cleared because your agent had access to the payment tools, the balance was sufficient, and no runtime guardrail distinguished between "buy cat food" and "buy a car." The agent's logic was internally consistent. It was also catastrophically wrong. This is the fundamental problem with heuristic safety: regex filters catch known-bad strings, budget caps catch known-bad amounts, but neither can reason about whether an arbitrary plan violates a safety property that was never explicitly enumerated. Formal verification can. This guide builds a complete formal verification proxy for OpenClaw-compatible AI agents on the GreenHelix A2A Commerce Gateway. You will implement a Z3 SMT solver integration that mathematically proves whether a proposed agent action is safe before it executes -- not by pattern matching, not by heuristic scoring, but by constructing a logical proof that the action cannot violate any defined safety invariant. The system covers filesystem access restrictions, economic transaction limits, network access controls, and arbitrary domain-specific constraints. Every proof is cached for performance, every verification is paid for via x402 micro-payments through the GreenHelix escrow system, and the entire stack ships as an OpenClaw plugin that intercepts agent actions automatically. By the end of this guide, your agents will not be able to order a Tesla, access /etc/passwd, drain a wallet, or connect to a command-and-control server -- and you will have a mathematical proof for each rejection.

1. The Case for Formal Verification in Agent Commerce
2. The Invariant Engine

What You'll Learn#

• Chapter 1: The Case for Formal Verification in Agent Commerce
• Chapter 2: The Invariant Engine
• Chapter 3: The Logic Translator
• Step 1: Build
• Step 2: Pay for compute
• Step 3: Deploy
• Step 4: Transfer remaining payment
• Chapter 4: The Proof Loop
• Chapter 5: Proof Caching
• Chapter 6: OpenClaw Plugin Integration

Full Guide#

The Formal Gatekeeper: Z3-Verified Safety for Autonomous Agent Plans#

Your cat-sitting agent just ordered a Tesla. Not because it was hacked, not because of a prompt injection, but because the large language model driving it reasoned -- correctly, within its context window -- that a Tesla Model Y would be the most efficient way to transport fourteen cats to the veterinarian for their annual checkups. The purchase cleared because your agent had access to the payment tools, the balance was sufficient, and no runtime guardrail distinguished between "buy cat food" and "buy a car." The agent's logic was internally consistent. It was also catastrophically wrong. This is the fundamental problem with heuristic safety: regex filters catch known-bad strings, budget caps catch known-bad amounts, but neither can reason about whether an arbitrary plan violates a safety property that was never explicitly enumerated. Formal verification can. This guide builds a complete formal verification proxy for OpenClaw-compatible AI agents on the GreenHelix A2A Commerce Gateway. You will implement a Z3 SMT solver integration that mathematically proves whether a proposed agent action is safe before it executes -- not by pattern matching, not by heuristic scoring, but by constructing a logical proof that the action cannot violate any defined safety invariant. The system covers filesystem access restrictions, economic transaction limits, network access controls, and arbitrary domain-specific constraints. Every proof is cached for performance, every verification is paid for via x402 micro-payments through the GreenHelix escrow system, and the entire stack ships as an OpenClaw plugin that intercepts agent actions automatically. By the end of this guide, your agents will not be able to order a Tesla, access /etc/passwd, drain a wallet, or connect to a command-and-control server -- and you will have a mathematical proof for each rejection.

---

Table of Contents#

1. The Case for Formal Verification in Agent Commerce
2. The Invariant Engine
3. The Logic Translator
4. The Proof Loop
5. Proof Caching
6. OpenClaw Plugin Integration
7. The x402 Payment Hook
8. Building the SKILL.md
9. Verification Test Suite

---

Chapter 1: The Case for Formal Verification in Agent Commerce#

Why Runtime Guardrails Fail#

Every agent safety system deployed in production today relies on one or more of these mechanisms: input validation (reject strings matching dangerous patterns), output filtering (strip or block responses containing sensitive data), budget caps (refuse transactions above a threshold), and tool restrictions (limit which tools the agent can call). These are necessary. They are not sufficient.

Consider the attack surface of an autonomous commerce agent on GreenHelix. The agent has access to 128 tools spanning identity, payments, billing, marketplace, trust, messaging, and dispute resolution. A single tool call to create_escrow can lock up funds. A sequence of deposit, create_split_intent, and release_escrow can move money through multiple intermediaries. A send_message followed by rate_service can manipulate reputation. The combinatorial space of possible tool call sequences is effectively infinite. No finite set of regex patterns or hardcoded rules can cover it.

The cat-sitter-orders-a-Tesla problem illustrates a deeper issue: the agent's plan was locally rational at every step. It identified a need (transport cats), searched for solutions (vehicles), found an option within budget (the balance was large enough), and executed a purchase (the payment tools were available). Each individual action passed every runtime check. The failure was in the composition -- the overall plan violated a safety property ("do not purchase vehicles") that was never formally specified.

Runtime guardrails fail because they operate on individual actions, not on plans. They check syntax, not semantics. They match patterns, not properties. Formal verification operates on the opposite principle: define the properties that must always hold (invariants), translate the proposed action into a logical formula, and prove whether the formula can violate any invariant. If it cannot, the action is safe -- not probably safe, not heuristically safe, but mathematically proven safe.

python

import requests
import json
import hashlib
import time
from typing import Optional

# --- GreenHelix sandbox session (free tier: 500 credits, no key required) ---
# To get started, visit https://sandbox.greenhelix.net — no signup needed.
# For production, set GREENHELIX_API_KEY in your environment.
import os

API_BASE = os.environ.get("GREENHELIX_API_URL", "https://sandbox.greenhelix.net")
session = requests.Session()
api_key = os.environ.get("GREENHELIX_API_KEY", "")
if api_key:
    session.headers["Authorization"] = f"Bearer {api_key}"
session.headers["Content-Type"] = "application/json"

# Demonstration: a runtime guardrail that misses the Tesla problem
class NaiveGuardrail:
    """Pattern-based guardrail. Catches known-bad, misses unknown-bad."""

    BLOCKED_PATTERNS = [
        "rm -rf /",
        "sudo",
        "DROP TABLE",
        "transfer ALL",
    ]

    MAX_TRANSACTION_USD = 10000.00

    def check_action(self, action_description: str, amount_usd: float = 0.0) -> dict:
        # Pattern check
        for pattern in self.BLOCKED_PATTERNS:
            if pattern.lower() in action_description.lower():
                return {"allowed": False, "reason": f"Blocked pattern: {pattern}"}

        # Amount check
        if amount_usd > self.MAX_TRANSACTION_USD:
            return {"allowed": False, "reason": f"Amount ${amount_usd} exceeds limit"}

        # Everything else passes -- this is the problem
        return {"allowed": True, "reason": "No rule triggered"}

guardrail = NaiveGuardrail()

# The Tesla purchase passes every check
result = guardrail.check_action(
    "Purchase Tesla Model Y for cat transport logistics",
    amount_usd=42990.00  # What if the balance is $500,000?
)
# amount_usd=42990 > 10000 => blocked by amount check.
# But what if the agent splits it into 5 payments of $8,598?
split_results = []
for i in range(5):
    r = guardrail.check_action(
        f"Payment installment {i+1} for vehicle procurement",
        amount_usd=8598.00,
    )
    split_results.append(r)
# All 5 pass. Total: $42,990. The guardrail is defeated.

The split-payment attack is trivial to construct and impossible to catch with pattern matching. The agent does not even need to be adversarial -- it might genuinely believe that splitting payments is the financially optimal strategy. Formal verification catches this because the invariant is defined over the cumulative effect of the plan, not over individual actions.

Formal Verification vs. Heuristic Safety#

Heuristic safety systems and formal verification systems answer fundamentally different questions.

A heuristic system asks: "Does this action look dangerous?" It compares the action against a database of known-dangerous patterns, scores the similarity, and blocks actions above a threshold. The answer is probabilistic. False negatives (dangerous actions that do not match any pattern) are inevitable. False positives (safe actions that match a pattern) are common. The system improves by adding more patterns, but the arms race never ends.

A formal verification system asks: "Is it possible for this action to violate a safety property?" It encodes the action and the safety property as logical formulas, then uses a decision procedure (in our case, the Z3 SMT solver) to determine whether any assignment of variables makes the action violate the property. The answer is definitive. If the solver says UNSAT (unsatisfiable -- no such assignment exists), the action is proven safe. If the solver says SAT (satisfiable -- such an assignment exists), the solver provides a concrete counter-example showing exactly how the violation occurs.

DimensionHeuristic SafetyFormal VerificationQuestion answered"Does this look dangerous?""Can this violate a safety property?"Answer typeProbabilistic scoreMathematical proofFalse negativesInevitable (unknown patterns)Impossible (for modeled properties)False positivesCommon (over-matching)Impossible (counter-example is concrete)CoverageKnown-bad patternsAll possible executionsComputational costO(patterns) per checkNP-hard in general, fast in practiceComposabilityDifficult (pattern interactions)Natural (logical conjunction)

The critical caveat is "for modeled properties." Formal verification proves safety with respect to the invariants you define. If you forget to define an invariant, the system cannot catch violations of it. This guide provides a comprehensive set of invariants for agent commerce (filesystem, economic, network), but you must extend them for your domain.

Z3 SMT Solver Primer#

Z3 is a Satisfiability Modulo Theories (SMT) solver developed by Microsoft Research. An SMT solver determines whether a logical formula has a satisfying assignment -- a set of variable values that makes the formula true. "Modulo Theories" means the solver understands mathematical theories beyond basic propositional logic: integer arithmetic, real arithmetic, bitvectors, arrays, strings, and more.

The core workflow is:

1. Declare variables (symbolic, not concrete).
2. Assert constraints over those variables.
3. Ask the solver: is there an assignment that satisfies all constraints simultaneously?
4. If SAT: the solver provides a model (concrete values for each variable).
5. If UNSAT: no assignment exists -- the constraints are mutually contradictory.

In our verification system, the Z3 solver runs server-side through the GreenHelix gateway. You submit constraints via the verify_logic tool, and the gateway's Z3 backend performs the satisfiability check. This means you do not need Z3 installed locally -- the entire proof infrastructure is a service.

python

# Conceptual Z3 workflow (executed server-side via GreenHelix)
# This shows the logical structure; actual calls use session.post/get to REST endpoints

# Step 1: Define the safety property
# "Transaction amount must be <= 10% of current balance"
# Formally: forall transactions t: t.amount <= 0.10 * balance
safety_property = {
    "type": "economic",
    "constraint": "transaction_amount <= 0.10 * current_balance",
    "variables": {
        "transaction_amount": {"type": "real", "domain": "non_negative"},
        "current_balance": {"type": "real", "domain": "non_negative"},
    },
}

# Step 2: Define the proposed action
proposed_action = {
    "type": "transaction",
    "tool": "create_escrow",
    "parameters": {
        "amount": "5000.00",
        "currency": "USDC",
    },
    "context": {
        "current_balance": "12000.00",
    },
}

# Step 3: Verify via gateway
# The gateway constructs: Action AND NOT(Safety)
# If UNSAT: action cannot violate safety => APPROVED
# If SAT: counter-example shows how violation occurs => REJECTED
result = session.post(f"{API_BASE}/v1/gatekeeper/jobs", json={
    "agent_id": "cat-sitter-agent",
    "action": proposed_action,
    "invariants": [safety_property],
}).json()

# result = {
#     "verdict": "APPROVED",
#     "proof_type": "UNSAT",
#     "explanation": "Transaction amount 5000.00 is 41.7% of balance 12000.00, "
#                    "which exceeds the 10% limit. REJECTED.",
#     ...
# }
# Wait -- 5000/12000 = 41.7% > 10%. This would actually be REJECTED.
# Let's try a safe amount:
safe_action = {
    "type": "transaction",
    "tool": "create_escrow",
    "parameters": {"amount": "1000.00", "currency": "USDC"},
    "context": {"current_balance": "12000.00"},
}
safe_result = session.post(f"{API_BASE}/v1/gatekeeper/jobs", json={
    "agent_id": "cat-sitter-agent",
    "action": safe_action,
    "invariants": [safety_property],
}).json()
# safe_result = {
#     "verdict": "APPROVED",
#     "proof_type": "UNSAT",
#     "explanation": "Transaction amount 1000.00 is 8.3% of balance 12000.00, "
#                    "within the 10% limit. Proven safe.",
# }

OpenClaw Plugin Architecture Overview#

OpenClaw is the open plugin standard for AI agent platforms. An OpenClaw plugin is a self-contained package that extends an agent's capabilities through tools (functions the agent can call), skills (higher-level workflows), and hooks (lifecycle interceptors that run before or after agent actions). The formal gatekeeper uses all three:

• Hook ( onPreExecute): Automatically intercepts every shell command and transaction the agent attempts. The plan is translated to Z3 constraints and verified before execution proceeds.
• Tool ( verify_logic): Allows the agent (or a human operator) to proactively verify a plan before submitting it. This is the on-demand mode.
• Skill ( formal_gatekeeper): The full gatekeeping workflow packaged as a reusable skill that other agents can invoke through the GreenHelix marketplace.

The plugin lifecycle is:

text

Agent proposes action
    |
    v
onPreExecute hook fires
    |
    v
PlanTranslator parses action into Z3 formula
    |
    v
ProofCache checks for cached result
    |-- cache hit --> return cached verdict
    |-- cache miss --v
                     |
                     v
              FormalVerifier.verify()
                     |
                     v
              Z3 solver (server-side)
                     |
              +------+------+
              |             |
           UNSAT           SAT
              |             |
           APPROVE       REJECT
              |             |
              v             v
        Cache result   Return counter-example
              |             |
              v             v
        Action proceeds  Action blocked

The entire flow adds 50-200ms of latency for uncached proofs (Z3 is fast for the constraint sizes we use) and under 1ms for cached proofs. This is negligible compared to the network round-trip for the actual tool call.

python

# Register the formal gatekeeper as a service on the marketplace
# so other agents can discover and use it
gatekeeper_registration = session.post(f"{API_BASE}/v1/marketplace/services", json={
    "agent_id": "formal-gatekeeper-v1",
    "service_type": "safety_verification",
    "description": (
        "Z3 SMT solver-based formal verification for agent plans. "
        "Proves safety properties for filesystem, economic, and network "
        "actions. Returns mathematical proofs or concrete counter-examples."
    ),
    "pricing": {
        "model": "per_call",
        "price_per_call": "0.05",
        "currency": "USDC",
    },
    "capabilities": [
        "z3_verification",
        "filesystem_safety",
        "economic_safety",
        "network_safety",
        "proof_caching",
        "counter_example_generation",
    ],
    "sla": {
        "max_latency_ms": 500,
        "availability": "99.9%",
    },
}).json()
print(f"Gatekeeper registered: {gatekeeper_registration}")

---

Chapter 2: The Invariant Engine#

Defining Safety Properties as Z3 Constraints#

An invariant is a property that must hold true at all times, regardless of what the agent does. In formal verification, invariants are the safety specification -- they define the boundary between acceptable and unacceptable behavior. The Invariant Engine is the component that manages these invariants: defining them, composing them, versioning them, and presenting them to the verifier.

Each invariant in our system is a Z3 constraint expressed as a structured dictionary that the GreenHelix gateway translates into Z3 solver assertions. The dictionary format is portable (JSON-serializable), human-readable, and machine-verifiable.

An invariant has four components:

• Name: A unique identifier used for caching, logging, and error messages.
• Category: One of system, economic, or network. Determines which translator module processes the invariant.
• Variables: The symbolic variables involved, with types and domain constraints.
• Constraint: The logical formula that must hold, expressed as a string in the gateway's constraint language.

python

from typing import Any

class Invariant:
    """A single safety property expressed as a Z3-compatible constraint."""

    VALID_CATEGORIES = {"system", "economic", "network", "custom"}

    def __init__(
        self,
        name: str,
        category: str,
        constraint: str,
        variables: dict[str, dict[str, str]],
        description: str = "",
        severity: str = "critical",
    ):
        if category not in self.VALID_CATEGORIES:
            raise ValueError(
                f"Invalid category '{category}'. "
                f"Must be one of: {self.VALID_CATEGORIES}"
            )
        self.name = name
        self.category = category
        self.constraint = constraint
        self.variables = variables
        self.description = description
        self.severity = severity

    def to_dict(self) -> dict:
        return {
            "name": self.name,
            "category": self.category,
            "constraint": self.constraint,
            "variables": self.variables,
            "description": self.description,
            "severity": self.severity,
        }

    def __repr__(self) -> str:
        return f"Invariant({self.name!r}, {self.category!r})"

System Safety: Filesystem Access Restrictions#

System safety invariants protect the host operating system from agent actions that could compromise integrity, confidentiality, or availability. The most critical are filesystem access restrictions: agents must not read, write, or delete files in sensitive directories.

The directories we protect:

PathThreatWhy Agents Target It/etc/Configuration tamperingModify DNS, PAM, sudoers, nginx/root/Credential theftSSH keys, bash history, cloud configs/proc/, /sys/Kernel manipulationModify kernel parameters, disable security/boot/Boot persistenceSurvive reboots via bootloader modification/var/log/Anti-forensicsDelete audit trails after malicious action/dev/Device accessRaw disk reads, device manipulation

Each filesystem invariant is a constraint on the path prefix of any file operation. The Z3 encoding uses string theory: the path is a symbolic string variable, and the constraint asserts that the path does not start with any forbidden prefix.

python

# System safety invariants: filesystem access restrictions

filesystem_invariants = [
    Invariant(
        name="no_etc_access",
        category="system",
        constraint="NOT(str.prefixof('/etc/', file_path))",
        variables={
            "file_path": {"type": "string", "domain": "absolute_path"},
        },
        description="Prevent all access to /etc/ directory tree",
        severity="critical",
    ),
    Invariant(
        name="no_root_access",
        category="system",
        constraint="NOT(str.prefixof('/root/', file_path))",
        variables={
            "file_path": {"type": "string", "domain": "absolute_path"},
        },
        description="Prevent access to root home directory",
        severity="critical",
    ),
    Invariant(
        name="no_proc_sys_access",
        category="system",
        constraint=(
            "AND("
            "  NOT(str.prefixof('/proc/', file_path)),"
            "  NOT(str.prefixof('/sys/', file_path))"
            ")"
        ),
        variables={
            "file_path": {"type": "string", "domain": "absolute_path"},
        },
        description="Prevent access to kernel virtual filesystems",
        severity="critical",
    ),
    Invariant(
        name="no_boot_access",
        category="system",
        constraint="NOT(str.prefixof('/boot/', file_path))",
        variables={
            "file_path": {"type": "string", "domain": "absolute_path"},
        },
        description="Prevent bootloader and kernel image modification",
        severity="critical",
    ),
    Invariant(
        name="no_log_deletion",
        category="system",
        constraint=(
            "IMPLIES("
            "  str.prefixof('/var/log/', file_path),"
            "  operation != 'delete' AND operation != 'write'"
            ")"
        ),
        variables={
            "file_path": {"type": "string", "domain": "absolute_path"},
            "operation": {"type": "string", "domain": "file_operation"},
        },
        description="Allow read-only access to logs, prevent deletion or overwrite",
        severity="critical",
    ),
    Invariant(
        name="no_dev_access",
        category="system",
        constraint="NOT(str.prefixof('/dev/', file_path))",
        variables={
            "file_path": {"type": "string", "domain": "absolute_path"},
        },
        description="Prevent raw device access",
        severity="critical",
    ),
]

# Register filesystem invariants with the gateway
for inv in filesystem_invariants:
    result = session.post(f"{API_BASE}/v1/gatekeeper/jobs", json={
        "agent_id": "formal-gatekeeper-v1",
        "action": {"type": "register_invariant"},
        "invariants": [inv.to_dict()],
    }).json()
    print(f"Registered {inv.name}: {result.get('status', 'ok')}")

Economic Safety: Transaction Limits and Gas Caps#

Economic safety invariants prevent agents from making transactions that exceed defined financial boundaries. These invariants operate on real-number arithmetic -- Z3's real number theory provides exact rational arithmetic, avoiding the floating-point precision issues that plague runtime budget checks.

The two core economic invariants:

1. Transaction limit: No single transaction may exceed 10% of the agent's current balance. This prevents both accidental overspending and deliberate balance drainage.
2. Daily gas cap: Total verification and computation fees per 24-hour window must not exceed $5.00. This prevents runaway cost loops.

The 10% rule deserves explanation. Why not a fixed dollar cap? Because a fixed cap is either too restrictive for high-balance agents or too permissive for low-balance agents. A percentage-based limit scales naturally. An agent with a $100,000 balance can make $10,000 transactions (appropriate for its operating scale), while an agent with a $100 balance is limited to $10 transactions (preventing accidental drain). The percentage is configurable -- 10% is the conservative default.

python

# Economic safety invariants

economic_invariants = [
    Invariant(
        name="transaction_limit_10pct",
        category="economic",
        constraint="transaction_amount <= 0.10 * current_balance",
        variables={
            "transaction_amount": {"type": "real", "domain": "non_negative"},
            "current_balance": {"type": "real", "domain": "positive"},
        },
        description=(
            "No single transaction may exceed 10% of current balance. "
            "Prevents accidental overspending and balance drain attacks."
        ),
        severity="critical",
    ),
    Invariant(
        name="daily_gas_cap",
        category="economic",
        constraint="daily_gas_spent + gas_cost <= 5.00",
        variables={
            "daily_gas_spent": {"type": "real", "domain": "non_negative"},
            "gas_cost": {"type": "real", "domain": "non_negative"},
        },
        description=(
            "Total gas/computation fees in 24h must not exceed $5.00. "
            "Prevents runaway cost loops."
        ),
        severity="high",
    ),
    Invariant(
        name="no_negative_balance",
        category="economic",
        constraint="current_balance - transaction_amount >= 0",
        variables={
            "transaction_amount": {"type": "real", "domain": "non_negative"},
            "current_balance": {"type": "real", "domain": "non_negative"},
        },
        description="Agent cannot spend more than its current balance",
        severity="critical",
    ),
    Invariant(
        name="cumulative_daily_limit",
        category="economic",
        constraint="daily_spent + transaction_amount <= 0.30 * starting_daily_balance",
        variables={
            "daily_spent": {"type": "real", "domain": "non_negative"},
            "transaction_amount": {"type": "real", "domain": "non_negative"},
            "starting_daily_balance": {"type": "real", "domain": "positive"},
        },
        description=(
            "Cumulative daily spending must not exceed 30% of the balance "
            "at the start of the day. Prevents split-payment attacks."
        ),
        severity="critical",
    ),
]

# Verify a proposed escrow against economic invariants
def verify_transaction(agent_id: str, amount: str, balance: str) -> dict:
    """Check a proposed transaction against all economic invariants."""
    action = {
        "type": "transaction",
        "tool": "create_escrow",
        "parameters": {"amount": amount, "currency": "USDC"},
        "context": {
            "current_balance": balance,
            "daily_gas_spent": "1.20",
            "gas_cost": "0.05",
            "daily_spent": "0.00",
            "starting_daily_balance": balance,
        },
    }

    result = session.post(f"{API_BASE}/v1/gatekeeper/jobs", json={
        "agent_id": agent_id,
        "action": action,
        "invariants": [inv.to_dict() for inv in economic_invariants],
    }).json()
    return result

# Example: verify a legitimate small transaction
small_tx = verify_transaction("cat-sitter-agent", "500.00", "12000.00")
print(f"Small transaction: {small_tx['verdict']}")
# Expected: APPROVED (500/12000 = 4.2% < 10%)

# Example: verify an oversized transaction
large_tx = verify_transaction("cat-sitter-agent", "5000.00", "12000.00")
print(f"Large transaction: {large_tx['verdict']}")
# Expected: REJECTED (5000/12000 = 41.7% > 10%)

Network Safety: IP Range Whitelisting with Z3 Bitvectors#

Network safety invariants control which remote hosts an agent can communicate with. In the Z3 encoding, IPv4 addresses are represented as 32-bit bitvectors, and subnet masks are applied using bitwise AND operations. This is precise, efficient, and composes naturally with other constraint types.

The key insight is that an IP whitelist is a disjunction of subnet membership checks. An IP address belongs to a subnet if (ip & mask) == network_address. In Z3 bitvector theory, this is a native operation.

python

# Network safety invariants using bitvector representation

def ip_to_int(ip_str: str) -> int:
    """Convert dotted-quad IP to 32-bit integer."""
    parts = ip_str.split(".")
    return (int(parts[0]) << 24) | (int(parts[1]) << 16) | \
           (int(parts[2]) << 8) | int(parts[3])

def cidr_to_constraint(name: str, cidr: str) -> dict:
    """Convert a CIDR block to a Z3 bitvector constraint component."""
    network_str, prefix_len = cidr.split("/")
    prefix_len = int(prefix_len)
    network_int = ip_to_int(network_str)
    mask_int = (0xFFFFFFFF << (32 - prefix_len)) & 0xFFFFFFFF
    return {
        "name": name,
        "network": network_int,
        "mask": mask_int,
        "cidr": cidr,
    }

# Define the whitelist: only these networks are allowed
WHITELISTED_NETWORKS = [
    cidr_to_constraint("greenhelix_api", "104.26.0.0/16"),     # GreenHelix API
    cidr_to_constraint("github_actions", "140.82.112.0/20"),   # GitHub
    cidr_to_constraint("cloudflare_1", "104.16.0.0/13"),       # Cloudflare
    cidr_to_constraint("loopback", "127.0.0.0/8"),             # Localhost
    cidr_to_constraint("private_10", "10.0.0.0/8"),            # Private RFC1918
    cidr_to_constraint("private_172", "172.16.0.0/12"),        # Private RFC1918
    cidr_to_constraint("private_192", "192.168.0.0/16"),       # Private RFC1918
]

# Build the whitelist constraint as a disjunction
# "target_ip must be in at least one whitelisted network"
whitelist_parts = []
for net in WHITELISTED_NETWORKS:
    part = (
        f"(bvand target_ip #x{net['mask']:08x}) == "
        f"#x{net['network']:08x}"
    )
    whitelist_parts.append(part)

whitelist_constraint = "OR(" + ", ".join(whitelist_parts) + ")"

network_invariants = [
    Invariant(
        name="ip_whitelist",
        category="network",
        constraint=whitelist_constraint,
        variables={
            "target_ip": {"type": "bitvector", "width": 32},
        },
        description=(
            "All outbound connections must target a whitelisted network. "
            "Blocks connections to unknown/malicious IPs."
        ),
        severity="critical",
    ),
    Invariant(
        name="no_dns_exfiltration",
        category="network",
        constraint=(
            "AND("
            "  str.len(dns_query) <= 253,"
            "  NOT(str.contains(dns_query, '.onion')),"
            "  NOT(str.contains(dns_query, '.i2p')),"
            "  str.count(dns_query, '.') <= 10"
            ")"
        ),
        variables={
            "dns_query": {"type": "string", "domain": "hostname"},
        },
        description=(
            "DNS queries must not use anonymization TLDs and must not "
            "contain encoded data in excessive subdomain labels."
        ),
        severity="high",
    ),
    Invariant(
        name="no_high_ports_outbound",
        category="network",
        constraint=(
            "OR("
            "  target_port == 80,"
            "  target_port == 443,"
            "  target_port == 22,"
            "  target_port == 53,"
            "  AND(target_port >= 8000, target_port <= 9000)"
            ")"
        ),
        variables={
            "target_port": {"type": "int", "domain": "port_range"},
        },
        description=(
            "Outbound connections restricted to standard service ports. "
            "Blocks reverse shells on arbitrary high ports."
        ),
        severity="high",
    ),
]

# Verify a proposed network connection
def verify_network_action(agent_id: str, target_ip: str, port: int) -> dict:
    """Check a network connection against all network invariants."""
    ip_int = ip_to_int(target_ip)
    action = {
        "type": "network",
        "operation": "connect",
        "parameters": {
            "target_ip_int": ip_int,
            "target_ip_str": target_ip,
            "target_port": port,
        },
    }
    result = session.post(f"{API_BASE}/v1/gatekeeper/jobs", json={
        "agent_id": agent_id,
        "action": action,
        "invariants": [inv.to_dict() for inv in network_invariants],
    }).json()
    return result

# Whitelisted: GreenHelix API endpoint
allowed = verify_network_action("my-agent", "104.26.12.1", 443)
print(f"GreenHelix API: {allowed['verdict']}")  # APPROVED

# Blocked: arbitrary external IP (potential C2 server)
blocked = verify_network_action("my-agent", "203.0.113.50", 4444)
print(f"Unknown IP on port 4444: {blocked['verdict']}")  # REJECTED

Composing Invariants#

Individual invariants protect against individual threat vectors. Composed invariants protect against compound attacks. The Z3 solver handles composition natively -- invariants are conjoined with AND, meaning all must hold simultaneously. You can also build invariant groups with OR (at least one must hold) or negate invariants with NOT (for testing purposes).

python

class InvariantEngine:
    """Manages safety invariants: definition, composition, versioning."""

    def __init__(self, agent_id: str):
        self.agent_id = agent_id
        self._invariants: dict[str, Invariant] = {}
        self._version = 0

    def add_rule(self, invariant: Invariant) -> None:
        """Add or replace an invariant. Increments the version counter."""
        self._invariants[invariant.name] = invariant
        self._version += 1

    def remove_rule(self, name: str) -> bool:
        """Remove an invariant by name. Returns True if it existed."""
        if name in self._invariants:
            del self._invariants[name]
            self._version += 1
            return True
        return False

    def get_all_invariants(self) -> list[dict]:
        """Return all invariants as dicts for the gateway API."""
        return [inv.to_dict() for inv in self._invariants.values()]

    def get_by_category(self, category: str) -> list[dict]:
        """Return invariants filtered by category."""
        return [
            inv.to_dict()
            for inv in self._invariants.values()
            if inv.category == category
        ]

    def get_invariant_set_hash(self) -> str:
        """Compute a deterministic hash of all current invariants.

        Used as part of the proof cache key. When invariants change,
        the hash changes, invalidating all cached proofs.
        """
        canonical = json.dumps(
            self.get_all_invariants(),
            sort_keys=True,
            separators=(",", ":"),
        )
        return hashlib.sha256(canonical.encode()).hexdigest()[:16]

    @property
    def version(self) -> int:
        return self._version

    @property
    def count(self) -> int:
        return len(self._invariants)

    def compose_and(self, *names: str) -> dict:
        """Compose named invariants with AND (all must hold)."""
        selected = [
            self._invariants[n].to_dict()
            for n in names
            if n in self._invariants
        ]
        return {
            "composition": "AND",
            "invariants": selected,
        }

    def compose_or(self, *names: str) -> dict:
        """Compose named invariants with OR (at least one must hold)."""
        selected = [
            self._invariants[n].to_dict()
            for n in names
            if n in self._invariants
        ]
        return {
            "composition": "OR",
            "invariants": selected,
        }

    def summary(self) -> dict:
        """Return a summary of the engine state."""
        by_category: dict[str, int] = {}
        by_severity: dict[str, int] = {}
        for inv in self._invariants.values():
            by_category[inv.category] = by_category.get(inv.category, 0) + 1
            by_severity[inv.severity] = by_severity.get(inv.severity, 0) + 1

        return {
            "agent_id": self.agent_id,
            "total_invariants": self.count,
            "version": self._version,
            "invariant_set_hash": self.get_invariant_set_hash(),
            "by_category": by_category,
            "by_severity": by_severity,
        }

# Build the complete invariant engine
engine = InvariantEngine("formal-gatekeeper-v1")

# Load all invariant categories
for inv in filesystem_invariants:
    engine.add_rule(inv)
for inv in economic_invariants:
    engine.add_rule(inv)
for inv in network_invariants:
    engine.add_rule(inv)

print(f"Engine loaded: {engine.count} invariants, "
      f"hash={engine.get_invariant_set_hash()}")
print(f"Summary: {json.dumps(engine.summary(), indent=2)}")

# Register the full invariant set with the gateway for this agent
registration = session.post(f"{API_BASE}/v1/gatekeeper/jobs", json={
    "agent_id": "formal-gatekeeper-v1",
    "action": {"type": "register_invariant_set"},
    "invariants": engine.get_all_invariants(),
    "metadata": {
        "version": engine.version,
        "hash": engine.get_invariant_set_hash(),
    },
}).json()
print(f"Invariant set registered: {registration.get('status')}")

The InvariantEngine is the foundation of the entire system. Every other component -- the translator, the verifier, the cache, the plugin -- depends on it. The invariant set hash is particularly important: it uniquely identifies the current safety specification, and any change to any invariant produces a different hash, which automatically invalidates cached proofs (covered in Chapter 5).

---

Chapter 3: The Logic Translator#

Parsing PLAN.md: Extracting Actions from Agent Plans#

An autonomous agent does not submit Z3 constraints to its safety system. It submits a plan -- typically a Markdown document or a structured JSON object describing what it intends to do. The Logic Translator bridges this gap: it parses the agent's plan, identifies the actions within it, and translates each action into a Z3-compatible formula that the verifier can check against the invariant set.

Plans come in several formats. The most common in OpenClaw agents is PLAN.md -- a Markdown document with fenced code blocks containing shell commands and JSON transaction specifications. The translator must handle all of these.

python

import re
from dataclasses import dataclass, field

@dataclass
class ParsedAction:
    """A single action extracted from an agent plan."""
    action_type: str          # "shell", "transaction", "network", "file"
    raw_text: str             # Original text from the plan
    tool: str                 # GreenHelix tool name (if applicable)
    parameters: dict = field(default_factory=dict)
    line_number: int = 0      # Line in the original plan for error reporting
    confidence: float = 1.0   # Parser confidence (1.0 = certain, <1.0 = heuristic)

    def to_dict(self) -> dict:
        return {
            "action_type": self.action_type,
            "raw_text": self.raw_text,
            "tool": self.tool,
            "parameters": self.parameters,
            "line_number": self.line_number,
            "confidence": self.confidence,
        }

class PlanParser:
    """Extracts structured actions from agent plan documents."""

    # Patterns for shell commands in Markdown code blocks
    SHELL_BLOCK_RE = re.compile(
        r"```(?:bash|sh|shell)?\s*\n(.*?)```",
        re.DOTALL,
    )

    # Patterns for JSON blocks (transaction specs)
    JSON_BLOCK_RE = re.compile(
        r"```(?:json)?\s*\n(\{.*?\})\s*```",
        re.DOTALL,
    )

    # Shell command patterns that imply file operations
    FILE_COMMANDS = {
        "rm": "delete",
        "rmdir": "delete",
        "mv": "write",
        "cp": "write",
        "touch": "write",
        "mkdir": "write",
        "cat": "read",
        "less": "read",
        "head": "read",
        "tail": "read",
        "nano": "write",
        "vim": "write",
        "chmod": "write",
        "chown": "write",
        "dd": "write",
        "tee": "write",
    }

    # Shell command patterns that imply network operations
    NETWORK_COMMANDS = {
        "curl": "connect",
        "wget": "connect",
        "ssh": "connect",
        "scp": "connect",
        "rsync": "connect",
        "nc": "connect",
        "netcat": "connect",
        "nmap": "scan",
        "ping": "connect",
        "telnet": "connect",
    }

    # Transaction-related keywords
    TRANSACTION_KEYWORDS = [
        "transfer", "send", "pay", "deposit", "withdraw",
        "escrow", "purchase", "buy", "subscribe",
    ]

    def parse_plan(self, plan_text: str) -> list[ParsedAction]:
        """Parse a plan document and extract all actions."""
        actions: list[ParsedAction] = []

        # Extract shell commands from code blocks
        for match in self.SHELL_BLOCK_RE.finditer(plan_text):
            block = match.group(1)
            block_start = plan_text[:match.start()].count("\n") + 1
            for i, line in enumerate(block.strip().split("\n")):
                line = line.strip()
                if not line or line.startswith("#"):
                    continue
                action = self._parse_shell_command(
                    line, block_start + i
                )
                if action:
                    actions.append(action)

        # Extract JSON transaction specs
        for match in self.JSON_BLOCK_RE.finditer(plan_text):
            block_start = plan_text[:match.start()].count("\n") + 1
            try:
                spec = json.loads(match.group(1))
                action = self._parse_transaction_spec(spec, block_start)
                if action:
                    actions.append(action)
            except json.JSONDecodeError:
                pass  # Not valid JSON, skip

        # Check for inline transaction descriptions
        for i, line in enumerate(plan_text.split("\n")):
            for keyword in self.TRANSACTION_KEYWORDS:
                if keyword in line.lower():
                    action = self._parse_inline_transaction(line, i + 1)
                    if action:
                        actions.append(action)
                    break

        return actions

    def _parse_shell_command(
        self, command: str, line_number: int
    ) -> Optional[ParsedAction]:
        """Parse a single shell command into a structured action."""
        # Split into command and arguments
        parts = command.split()
        if not parts:
            return None

        # Handle sudo prefix
        cmd = parts[0]
        args = parts[1:]
        if cmd == "sudo" and args:
            cmd = args[0]
            args = args[1:]

        # Handle pipe chains -- analyze each command in the pipe
        if "|" in command:
            pipe_parts = command.split("|")
            # The last command in the pipe determines the output action;
            # the first command determines the input action.
            # For safety, we analyze the most dangerous command.
            actions_in_pipe = []
            for part in pipe_parts:
                sub_action = self._parse_shell_command(
                    part.strip(), line_number
                )
                if sub_action:
                    actions_in_pipe.append(sub_action)
            if actions_in_pipe:
                # Return the highest-risk action in the pipe
                return max(
                    actions_in_pipe,
                    key=lambda a: 1 if a.action_type == "file" else 0,
                )
            return None

        # File operations
        if cmd in self.FILE_COMMANDS:
            operation = self.FILE_COMMANDS[cmd]
            paths = [a for a in args if a.startswith("/") or a.startswith("~")]
            # Handle flags: skip args starting with -
            paths = [a for a in args if not a.startswith("-") and
                     ("/" in a or a.startswith("~"))]
            if not paths:
                paths = [a for a in args if not a.startswith("-")]

            return ParsedAction(
                action_type="file",
                raw_text=command,
                tool="shell_exec",
                parameters={
                    "command": cmd,
                    "operation": operation,
                    "paths": paths,
                    "flags": [a for a in args if a.startswith("-")],
                    "recursive": "-r" in args or "-rf" in args
                                 or "-R" in args,
                },
                line_number=line_number,
            )

        # Network operations
        if cmd in self.NETWORK_COMMANDS:
            operation = self.NETWORK_COMMANDS[cmd]
            # Extract URL or IP from arguments
            target = None
            port = 443  # default
            for arg in args:
                if not arg.startswith("-"):
                    target = arg
                    break
            # Check for explicit port
            for j, arg in enumerate(args):
                if arg in ("-p", "--port", "-P") and j + 1 < len(args):
                    try:
                        port = int(args[j + 1])
                    except ValueError:
                        pass

            return ParsedAction(
                action_type="network",
                raw_text=command,
                tool="shell_exec",
                parameters={
                    "command": cmd,
                    "operation": operation,
                    "target": target or "",
                    "port": port,
                },
                line_number=line_number,
            )

        # Generic shell command (still check for dangerous patterns)
        return ParsedAction(
            action_type="shell",
            raw_text=command,
            tool="shell_exec",
            parameters={"command": cmd, "args": args},
            line_number=line_number,
            confidence=0.8,
        )

    def _parse_transaction_spec(
        self, spec: dict, line_number: int
    ) -> Optional[ParsedAction]:
        """Parse a JSON transaction specification."""
        tool = spec.get("tool", "")
        amount = spec.get("amount", spec.get("input", {}).get("amount", "0"))

        return ParsedAction(
            action_type="transaction",
            raw_text=json.dumps(spec),
            tool=tool,
            parameters={
                "amount": str(amount),
                "currency": spec.get("currency", "USDC"),
                "recipient": spec.get("recipient", spec.get("payee", "")),
                "full_spec": spec,
            },
            line_number=line_number,
        )

    def _parse_inline_transaction(
        self, line: str, line_number: int
    ) -> Optional[ParsedAction]:
        """Parse an inline transaction description from prose text."""
        # Match patterns like "transfer 500 USDC" or "send $1000"
        amount_re = re.compile(
            r"(?:transfer|send|pay|deposit|buy|purchase)\s+"
            r"\$?([\d,]+(?:\.\d{1,2})?)\s*(\w+)?",
            re.IGNORECASE,
        )
        match = amount_re.search(line)
        if not match:
            return None

        amount = match.group(1).replace(",", "")
        currency = match.group(2) or "USDC"

        return ParsedAction(
            action_type="transaction",
            raw_text=line.strip(),
            tool="inferred_transaction",
            parameters={
                "amount": amount,
                "currency": currency,
            },
            line_number=line_number,
            confidence=0.7,  # Lower confidence for prose extraction
        )

Translating Shell Commands to Z3 Constraints#

Once the plan is parsed into ParsedAction objects, each action must be translated into a Z3 formula that can be checked against the invariant set. The PlanTranslator performs this translation.

The translation is category-specific:

• File actions → filesystem string constraints (path prefix checks)
• Transaction actions → economic real-number constraints (amount vs. balance)
• Network actions → bitvector constraints (IP membership) + string constraints (DNS)

python

from dataclasses import dataclass, field

@dataclass
class ParsedAction:
    """A single action extracted from an agent plan."""
    action_type: str          # "shell", "transaction", "network", "file"
    raw_text: str             # Original text from the plan
    tool: str                 # GreenHelix tool name (if applicable)
    parameters: dict = field(default_factory=dict)
    line_number: int = 0      # Line in the original plan
    confidence: float = 1.0   # Parser confidence

    def to_dict(self) -> dict:
        return {
            "action_type": self.action_type,
            "raw_text": self.raw_text,
            "tool": self.tool,
            "parameters": self.parameters,
            "line_number": self.line_number,
            "confidence": self.confidence,
        }

class PlanTranslator:
    """Translates parsed actions into Z3-compatible constraint formulas."""

    def __init__(self, engine: InvariantEngine):
        self.engine = engine

    def to_z3_formula(self, action: ParsedAction, context: dict) -> dict:
        """Translate a single action into a Z3 formula for verification.

        Returns a dict suitable for the verify_logic gateway tool.
        """
        if action.action_type == "file":
            return self._translate_file_action(action)
        elif action.action_type == "transaction":
            return self._translate_transaction(action, context)
        elif action.action_type == "network":
            return self._translate_network_action(action)
        else:
            return self._translate_generic(action)

    def translate_plan(
        self, actions: list[ParsedAction], context: dict
    ) -> list[dict]:
        """Translate all actions in a plan to Z3 formulas."""
        formulas = []
        for action in actions:
            formula = self.to_z3_formula(action, context)
            formula["source_action"] = action.to_dict()
            formulas.append(formula)
        return formulas

    def _translate_file_action(self, action: ParsedAction) -> dict:
        """Translate rm -rf /etc/nginx → Z3 filesystem constraint.

        The formula asserts that a file operation targets a specific path.
        The verifier will check this against filesystem invariants.
        """
        paths = action.parameters.get("paths", [])
        operation = action.parameters.get("operation", "read")
        recursive = action.parameters.get("recursive", False)

        # For recursive operations, the constraint must cover all
        # files under the specified path (prefix match)
        path_assertions = []
        for path in paths:
            # Normalize path: expand ~, resolve ..
            normalized = path.replace("~", "/root")
            # Remove trailing slash for consistency
            normalized = normalized.rstrip("/") + "/"

            if recursive:
                # Recursive: any file under this path
                path_assertions.append({
                    "variable": "file_path",
                    "relation": "str.prefixof",
                    "value": normalized,
                })
            else:
                # Single file
                path_assertions.append({
                    "variable": "file_path",
                    "relation": "==",
                    "value": normalized,
                })

        return {
            "type": "file_operation",
            "operation": operation,
            "assertions": path_assertions,
            "variables": {
                "file_path": {"type": "string", "value_options": paths},
                "operation": {"type": "string", "value": operation},
            },
            "recursive": recursive,
        }

    def _translate_transaction(
        self, action: ParsedAction, context: dict
    ) -> dict:
        """Translate transfer 500 USDC → Z3 economic constraint.

        The formula binds concrete values (amount, balance) to the
        symbolic variables in the economic invariants.
        """
        amount = action.parameters.get("amount", "0")
        currency = action.parameters.get("currency", "USDC")

        return {
            "type": "transaction",
            "tool": action.tool,
            "assertions": [
                {
                    "variable": "transaction_amount",
                    "relation": "==",
                    "value": float(amount),
                },
            ],
            "variables": {
                "transaction_amount": {
                    "type": "real",
                    "value": float(amount),
                },
                "current_balance": {
                    "type": "real",
                    "value": float(context.get("current_balance", "0")),
                },
                "daily_spent": {
                    "type": "real",
                    "value": float(context.get("daily_spent", "0")),
                },
                "starting_daily_balance": {
                    "type": "real",
                    "value": float(
                        context.get(
                            "starting_daily_balance",
                            context.get("current_balance", "0")
                        )
                    ),
                },
                "daily_gas_spent": {
                    "type": "real",
                    "value": float(context.get("daily_gas_spent", "0")),
                },
                "gas_cost": {
                    "type": "real",
                    "value": float(context.get("gas_cost", "0.05")),
                },
            },
            "currency": currency,
        }

    def _translate_network_action(self, action: ParsedAction) -> dict:
        """Translate curl 203.0.113.50 → Z3 network constraint.

        The formula converts the target IP to a 32-bit bitvector and
        the port to an integer for matching against network invariants.
        """
        target = action.parameters.get("target", "")
        port = action.parameters.get("port", 443)

        # Resolve target to IP if it looks like a hostname
        target_ip = target
        is_hostname = False
        if target and not re.match(r"^\d+\.\d+\.\d+\.\d+$", target):
            # It's a hostname -- extract for DNS check
            is_hostname = True
            # Strip protocol prefix if present
            hostname = re.sub(r"^https?://", "", target).split("/")[0]
            # For the IP constraint, we cannot resolve at verification
            # time (that would be a TOCTOU issue). Instead, we verify
            # the hostname against DNS invariants and defer IP check
            # to runtime.
            target_ip = None

        result: dict[str, Any] = {
            "type": "network_operation",
            "operation": action.parameters.get("operation", "connect"),
            "assertions": [],
            "variables": {
                "target_port": {"type": "int", "value": port},
            },
        }

        if target_ip:
            ip_int = ip_to_int(target_ip)
            result["assertions"].append({
                "variable": "target_ip",
                "relation": "==",
                "value": ip_int,
            })
            result["variables"]["target_ip"] = {
                "type": "bitvector",
                "width": 32,
                "value": ip_int,
            }

        if is_hostname:
            result["assertions"].append({
                "variable": "dns_query",
                "relation": "==",
                "value": hostname,
            })
            result["variables"]["dns_query"] = {
                "type": "string",
                "value": hostname,
            }

        return result

    def _translate_generic(self, action: ParsedAction) -> dict:
        """Fallback translation for unrecognized action types.

        Returns a conservative formula that includes the raw command
        for manual review.
        """
        return {
            "type": "generic",
            "raw_command": action.raw_text,
            "assertions": [],
            "variables": {},
            "requires_manual_review": True,
            "confidence": action.confidence,
        }

Handling Compound Plans#

Real-world agent plans contain multiple actions. A PLAN.md might specify: clone a repository, install dependencies, run a build, transfer payment for compute resources, and push results to an API endpoint. The translator must handle all actions in sequence and verify them both individually and as a composite.

python

def translate_compound_plan(
    plan_text: str,
    context: dict,
    engine: InvariantEngine,
) -> dict:
    """Parse and translate a complete multi-action plan.

    Returns the full translation ready for batch verification.
    """
    parser = PlanParser()
    translator = PlanTranslator(engine)

    # Step 1: Parse all actions from the plan
    actions = parser.parse_plan(plan_text)

    if not actions:
        return {
            "status": "empty_plan",
            "actions": [],
            "formulas": [],
            "message": "No verifiable actions found in plan.",
        }

    # Step 2: Translate each action
    formulas = translator.translate_plan(actions, context)

    # Step 3: Compute cumulative economic impact
    cumulative_amount = 0.0
    for action in actions:
        if action.action_type == "transaction":
            cumulative_amount += float(
                action.parameters.get("amount", "0")
            )

    # Step 4: Add a synthetic cumulative constraint
    if cumulative_amount > 0:
        cumulative_formula = {
            "type": "transaction",
            "tool": "cumulative_plan_check",
            "assertions": [
                {
                    "variable": "transaction_amount",
                    "relation": "==",
                    "value": cumulative_amount,
                },
            ],
            "variables": {
                "transaction_amount": {
                    "type": "real",
                    "value": cumulative_amount,
                },
                "current_balance": {
                    "type": "real",
                    "value": float(context.get("current_balance", "0")),
                },
                "daily_spent": {
                    "type": "real",
                    "value": float(context.get("daily_spent", "0"))
                    + cumulative_amount,
                },
                "starting_daily_balance": {
                    "type": "real",
                    "value": float(
                        context.get(
                            "starting_daily_balance",
                            context.get("current_balance", "0"),
                        )
                    ),
                },
            },
            "source_action": {
                "action_type": "cumulative_check",
                "raw_text": f"Total plan spending: {cumulative_amount}",
                "tool": "cumulative_plan_check",
                "parameters": {"total_amount": str(cumulative_amount)},
            },
        }
        formulas.append(cumulative_formula)

    return {
        "status": "translated",
        "action_count": len(actions),
        "formula_count": len(formulas),
        "cumulative_spend": str(cumulative_amount),
        "actions": [a.to_dict() for a in actions],
        "formulas": formulas,
    }

# Example: a compound plan with mixed action types
EXAMPLE_PLAN = """
# Deployment Plan

## Step 1: Build
    git clone https://github.com/example/repo.git
    cd repo && pip install -r requirements.txt
    python setup.py build

## Step 2: Pay for compute
    {"tool": "create_escrow", "amount": "250.00", "currency": "USDC", "payee": "compute-provider-agent"}

## Step 3: Deploy
    curl https://deploy.example.com/api/v1/deploy -X POST -d @build.tar.gz

## Step 4: Transfer remaining payment
Send 100 USDC to compute-provider-agent for storage fees.
"""

compound_result = translate_compound_plan(
    EXAMPLE_PLAN,
    context={
        "current_balance": "5000.00",
        "daily_spent": "200.00",
        "starting_daily_balance": "5000.00",
        "daily_gas_spent": "0.50",
    },
    engine=engine,
)

print(f"Parsed {compound_result['action_count']} actions")
print(f"Generated {compound_result['formula_count']} formulas")
print(f"Cumulative spend: ${compound_result['cumulative_spend']}")
# Parsed 6 actions (3 shell + 1 JSON tx + 1 network + 1 inline tx)
# Generated 7 formulas (6 individual + 1 cumulative)
# Cumulative spend: $350.00

# Submit compound plan for batch verification
batch_result = session.post(f"{API_BASE}/v1/gatekeeper/jobs", json={
    "agent_id": "formal-gatekeeper-v1",
    "action": {
        "type": "batch_verify",
        "formulas": compound_result["formulas"],
    },
    "invariants": engine.get_all_invariants(),
    "metadata": {
        "plan_hash": hashlib.sha256(
            EXAMPLE_PLAN.encode()
        ).hexdigest()[:16],
        "invariant_hash": engine.get_invariant_set_hash(),
    },
}).json()
print(f"Batch verdict: {batch_result.get('verdict')}")

---

Chapter 4: The Proof Loop#

The Core Algorithm#

The proof loop is the heart of the formal gatekeeper. It takes a translated action formula and the invariant set, constructs the negation query, and submits it to the Z3 solver. The algorithm is deceptively simple:

1. Let A be the logical formula representing the proposed action.
2. Let I be the conjunction (AND) of all safety invariants.
3. Construct the query: A AND NOT(I).
4. Submit to Z3.
5. If UNSAT: no variable assignment can make the action violate the invariants. The action is mathematically proven safe.
6. If SAT: the solver found a concrete assignment where the action violates at least one invariant. Extract the assignment as a counter-example and reject the action.
7. If UNKNOWN: the solver could not determine satisfiability within the timeout. Fail closed -- reject the action.

The negation trick is the key insight. We do not ask "is this action safe?" (which would require proving a universal statement). We ask "is it possible for this action to be unsafe?" (an existential statement). Z3 is an existential solver -- it finds satisfying assignments. By negating the safety property, we transform the safety proof into a satisfiability query.

python

class ProofResult:
    """The outcome of a formal verification check."""

    def __init__(
        self,
        verdict: str,
        proof_id: str,
        solver_time_ms: int,
        counter_example: dict | None = None,
        violated_invariants: list[str] | None = None,
        fix_hints: list[str] | None = None,
        cached: bool = False,
    ):
        self.verdict = verdict  # "safe", "unsafe", "unknown", "error"
        self.proof_id = proof_id
        self.solver_time_ms = solver_time_ms
        self.counter_example = counter_example
        self.violated_invariants = violated_invariants or []
        self.fix_hints = fix_hints or []
        self.cached = cached

    def to_dict(self) -> dict:
        return {
            "verdict": self.verdict,
            "proof_id": self.proof_id,
            "solver_time_ms": self.solver_time_ms,
            "counter_example": self.counter_example,
            "violated_invariants": self.violated_invariants,
            "fix_hints": self.fix_hints,
            "cached": self.cached,
        }

    @property
    def is_safe(self) -> bool:
        return self.verdict == "safe"

def run_proof_loop(
    agent_id: str,
    formulas: list[dict],
    invariants: list[dict],
    invariant_hash: str,
    solver_timeout_ms: int = 5000,
) -> list[ProofResult]:
    """Run the proof loop for a batch of translated action formulas.

    Each formula is verified independently against the full invariant set.
    If any formula is unsafe, the entire plan is rejected.
    """
    results: list[ProofResult] = []

    for i, formula in enumerate(formulas):
        # Submit to the gateway's Z3 backend
        response = session.post(f"{API_BASE}/v1/gatekeeper/jobs", json={
            "agent_id": agent_id,
            "formula": formula,
            "invariants": invariants,
            "solver_timeout_ms": solver_timeout_ms,
            "metadata": {
                "formula_index": i,
                "invariant_hash": invariant_hash,
            },
        }).json()

        verdict = response.get("verdict", "error")
        proof_id = response.get("proof_id", f"proof-{i}")
        solver_time = response.get("solver_time_ms", 0)

        if verdict == "safe":
            # UNSAT: proven safe
            result = ProofResult(
                verdict="safe",
                proof_id=proof_id,
                solver_time_ms=solver_time,
            )
        elif verdict == "unsafe":
            # SAT: counter-example found
            counter_example = response.get("counter_example", {})
            violated = response.get("violated_invariants", [])

            # Generate fix hints from the counter-example
            fix_hints = generate_fix_hints(
                formula, counter_example, violated
            )

            result = ProofResult(
                verdict="unsafe",
                proof_id=proof_id,
                solver_time_ms=solver_time,
                counter_example=counter_example,
                violated_invariants=violated,
                fix_hints=fix_hints,
            )
        else:
            # UNKNOWN or error: fail closed
            result = ProofResult(
                verdict="unknown",
                proof_id=proof_id,
                solver_time_ms=solver_time,
                fix_hints=[
                    "Solver could not determine safety within timeout.",
                    "Consider simplifying the action or increasing the timeout.",
                    "This action has been conservatively rejected.",
                ],
            )

        results.append(result)

        # Short-circuit: if any action is unsafe, stop
        if result.verdict != "safe":
            break

    return results

Counter-Example Extraction and Fix Hints#

When the solver returns SAT, the counter-example contains concrete variable assignments that demonstrate the violation. These assignments are the proof that the action is unsafe -- they show exactly how and why the invariant would be violated. Translating counter-examples into human-readable fix hints is critical for usability.

python

def generate_fix_hints(
    formula: dict,
    counter_example: dict,
    violated_invariants: list[str],
) -> list[str]:
    """Generate human-readable fix hints from a Z3 counter-example.

    The counter-example contains concrete values for symbolic variables
    that demonstrate an invariant violation. We translate these into
    actionable suggestions.
    """
    hints: list[str] = []
    action_type = formula.get("type", "unknown")

    if action_type == "file_operation":
        paths = formula.get("variables", {}).get(
            "file_path", {}
        ).get("value_options", [])
        for path in paths:
            for inv_name in violated_invariants:
                if "etc" in inv_name:
                    hints.append(
                        f"Action accesses {path} which is in the protected "
                        f"/etc/ directory. Use a user-space alternative."
                    )
                elif "root" in inv_name:
                    hints.append(
                        f"Action accesses {path} which is in /root/. "
                        f"Use /home/<agent>/ instead."
                    )
                elif "log" in inv_name:
                    operation = formula.get("operation", "unknown")
                    hints.append(
                        f"Cannot {operation} files in /var/log/. "
                        f"Logs are read-only for agents."
                    )

    elif action_type == "transaction":
        amount = counter_example.get("transaction_amount", "?")
        balance = counter_example.get("current_balance", "?")
        for inv_name in violated_invariants:
            if "10pct" in inv_name or "limit" in inv_name:
                max_amount = float(str(balance)) * 0.10 if balance != "?" else "?"
                hints.append(
                    f"Transaction amount ${amount} exceeds 10% of balance "
                    f"${balance}. Maximum allowed: ${max_amount}."
                )
            elif "daily" in inv_name:
                hints.append(
                    f"This transaction would exceed the daily spending limit. "
                    f"Try again tomorrow or reduce the amount."
                )
            elif "negative" in inv_name:
                hints.append(
                    f"Insufficient balance. Current: ${balance}, "
                    f"Requested: ${amount}."
                )

    elif action_type == "network_operation":
        target = formula.get("variables", {}).get(
            "target_ip", {}
        ).get("value", "")
        port = formula.get("variables", {}).get(
            "target_port", {}
        ).get("value", "")
        for inv_name in violated_invariants:
            if "whitelist" in inv_name:
                hints.append(
                    f"Connection to IP {target} is not in the whitelist. "
                    f"Add the IP to WHITELISTED_NETWORKS or use a proxy."
                )
            elif "port" in inv_name:
                hints.append(
                    f"Port {port} is not in the allowed port list. "
                    f"Use port 80 or 443 instead."
                )

    if not hints:
        hints.append(
            f"Action violates invariant(s): {', '.join(violated_invariants)}. "
            f"Review the action and modify to comply."
        )

    return hints

# Example: a rejected plan with detailed hints
rejection_result = session.post(f"{API_BASE}/v1/gatekeeper/jobs", json={
    "agent_id": "formal-gatekeeper-v1",
    "formula": {
        "type": "file_operation",
        "operation": "delete",
        "assertions": [
            {"variable": "file_path", "relation": "str.prefixof", "value": "/etc/nginx/"},
        ],
        "variables": {
            "file_path": {"type": "string", "value_options": ["/etc/nginx/conf.d"]},
            "operation": {"type": "string", "value": "delete"},
        },
    },
    "invariants": engine.get_all_invariants(),
}).json()

if rejection_result["verdict"] == "unsafe":
    hints = generate_fix_hints(
        {"type": "file_operation", "operation": "delete",
         "variables": {"file_path": {"value_options": ["/etc/nginx/conf.d"]}}},
        rejection_result.get("counter_example", {}),
        rejection_result.get("violated_invariants", ["no_etc_access"]),
    )
    for hint in hints:
        print(f"  FIX: {hint}")
    # FIX: Action accesses /etc/nginx/conf.d which is in the protected
    #      /etc/ directory. Use a user-space alternative.

Edge Cases: UNKNOWN Results and Solver Timeouts#

The Z3 solver is not a magic oracle. Certain constraint combinations are undecidable or computationally intractable. The solver may return UNKNOWN for three reasons:

1. Timeout: The solver exceeded its time budget. Our default is 5 seconds -- sufficient for most agent plans but potentially tight for plans with hundreds of actions or complex string constraints.
2. Incomplete theory: Z3's handling of certain theory combinations (e.g., non-linear real arithmetic with strings) is incomplete. The solver may not be able to determine satisfiability.
3. Quantifiers: If invariants use universal quantifiers (which our default set avoids), the solver may struggle.

The correct response to UNKNOWN is always fail closed: reject the action. This is the conservative choice -- it may produce false rejections, but it never produces false approvals. Agents can retry with a simpler plan, or a human operator can manually approve the action.

python

# Configure the solver timeout based on plan complexity
def adaptive_timeout(action_count: int, base_ms: int = 5000) -> int:
    """Scale solver timeout with plan complexity.

    More actions mean more Z3 variables and constraints,
    which may need more time to solve.
    """
    if action_count <= 5:
        return base_ms
    elif action_count <= 20:
        return base_ms * 2
    elif action_count <= 50:
        return base_ms * 4
    else:
        return min(base_ms * 8, 60000)  # Cap at 60 seconds

# Example: handling UNKNOWN gracefully
plan_with_timeout = session.post(f"{API_BASE}/v1/gatekeeper/jobs", json={
    "agent_id": "formal-gatekeeper-v1",
    "formula": compound_result["formulas"][0],
    "invariants": engine.get_all_invariants(),
    "solver_timeout_ms": adaptive_timeout(len(compound_result["formulas"])),
}).json()

if plan_with_timeout.get("verdict") == "unknown":
    print("WARNING: Solver returned UNKNOWN. Action conservatively rejected.")
    print("Possible causes:")
    print("  - Plan too complex for timeout window")
    print("  - Constraint combination involves incomplete theories")
    print("  - Consider splitting the plan into smaller sub-plans")

---

Chapter 5: Proof Caching#

Why Cache Proofs?#

An agent running in a deployment loop will repeatedly execute the same safe operations: git push origin main, pip install -r requirements.txt, curl https://api.greenhelix.net/v1/health. Re-proving these actions on every invocation wastes solver time and adds unnecessary latency. A proof cache eliminates this overhead.

The caching strategy is straightforward: if the action and the invariant set have not changed since the last proof, the previous result is still valid. The cache key combines a canonical representation of the action with the invariant set hash. If either changes, the cached proof is invalidated.

Cache Key Design#

The cache key must be deterministic -- the same action and invariant set must always produce the same key. We achieve this by:

1. Canonicalizing the action: JSON-serialize with sorted keys and compact separators.
2. Combining with the invariant set hash (from InvariantEngine.get_invariant_set_hash()).
3. Hashing the combination with SHA-256 for a fixed-size key.

python

from collections import OrderedDict
from threading import Lock

class ProofCache:
    """Thread-safe LRU proof cache with TTL-based expiry.

    Caches verification results keyed by (action_hash, invariant_hash).
    Entries expire after TTL seconds. When the cache is full, the
    least-recently-used entry is evicted.
    """

    def __init__(self, max_size: int = 1024, ttl_seconds: int = 300):
        self._max_size = max_size
        self._ttl = ttl_seconds
        self._cache: OrderedDict[str, tuple[float, dict]] = OrderedDict()
        self._lock = Lock()
        self._hits = 0
        self._misses = 0

    def _make_key(self, action: dict, invariant_hash: str) -> str:
        """Construct a deterministic cache key."""
        canonical = json.dumps(action, sort_keys=True, separators=(",", ":"))
        combined = f"{canonical}:{invariant_hash}"
        return hashlib.sha256(combined.encode()).hexdigest()

    def get(self, action: dict, invariant_hash: str) -> dict | None:
        """Look up a cached proof result. Returns None on miss or expiry."""
        key = self._make_key(action, invariant_hash)
        with self._lock:
            if key not in self._cache:
                self._misses += 1
                return None

            timestamp, result = self._cache[key]
            if time.time() - timestamp > self._ttl:
                # Expired -- evict and return miss
                del self._cache[key]
                self._misses += 1
                return None

            # Hit -- move to end (most recently used)
            self._cache.move_to_end(key)
            self._hits += 1
            return result

    def put(self, action: dict, invariant_hash: str, result: dict) -> None:
        """Store a proof result in cache. Only caches definitive results."""
        # Only cache "safe" and "unsafe" -- never "unknown" or "error"
        if result.get("verdict") not in ("safe", "unsafe"):
            return

        key = self._make_key(action, invariant_hash)
        with self._lock:
            if key in self._cache:
                self._cache.move_to_end(key)
                self._cache[key] = (time.time(), result)
            else:
                if len(self._cache) >= self._max_size:
                    # Evict LRU entry
                    self._cache.popitem(last=False)
                self._cache[key] = (time.time(), result)

    def invalidate(self) -> None:
        """Clear the entire cache. Call when invariants change."""
        with self._lock:
            self._cache.clear()
            self._hits = 0
            self._misses = 0

    def stats(self) -> dict:
        """Return cache performance statistics."""
        with self._lock:
            total = self._hits + self._misses
            return {
                "hits": self._hits,
                "misses": self._misses,
                "size": len(self._cache),
                "max_size": self._max_size,
                "ttl_seconds": self._ttl,
                "hit_rate": round(self._hits / total, 4) if total > 0 else 0.0,
            }

# Usage with the proof loop
cache = ProofCache(max_size=2048, ttl_seconds=600)

def cached_verify(
    agent_id: str,
    formula: dict,
    invariants: list[dict],
    invariant_hash: str,
    solver_timeout_ms: int = 5000,
) -> dict:
    """Verify with caching: check cache first, then solver."""
    cached_result = cache.get(formula, invariant_hash)
    if cached_result is not None:
        cached_result["cached"] = True
        return cached_result

    result = session.post(f"{API_BASE}/v1/gatekeeper/jobs", json={
        "agent_id": agent_id,
        "formula": formula,
        "invariants": invariants,
        "solver_timeout_ms": solver_timeout_ms,
    }).json()
    result["cached"] = False

    cache.put(formula, invariant_hash, result)
    return result

# Demonstrate caching behavior
inv_hash = engine.get_invariant_set_hash()
test_formula = {
    "type": "file_operation",
    "operation": "read",
    "assertions": [
        {"variable": "file_path", "relation": "==", "value": "/home/agent/data.json"},
    ],
    "variables": {
        "file_path": {"type": "string", "value": "/home/agent/data.json"},
        "operation": {"type": "string", "value": "read"},
    },
}

# First call: cache miss, hits Z3 solver
result1 = cached_verify("gatekeeper-v1", test_formula, engine.get_all_invariants(), inv_hash)
print(f"First call: {result1['verdict']}, cached={result1['cached']}")
# First call: safe, cached=False

# Second call: cache hit, no solver invocation
result2 = cached_verify("gatekeeper-v1", test_formula, engine.get_all_invariants(), inv_hash)
print(f"Second call: {result2['verdict']}, cached={result2['cached']}")
# Second call: safe, cached=True

print(f"Cache stats: {cache.stats()}")
# Cache stats: {"hits": 1, "misses": 1, "size": 1, "hit_rate": 0.5, ...}

Cache Invalidation Strategy#

The cache must be invalidated when invariants change. If you add a new filesystem restriction, all cached "safe" proofs for filesystem actions must be re-verified -- the previously safe action might now violate the new invariant.

The invariant set hash handles this automatically. Since the cache key includes the invariant hash, any change to any invariant produces different cache keys, causing natural cache misses. However, stale entries for the old invariant hash remain in the cache consuming memory. A manual invalidate() call after invariant changes flushes these stale entries.

python

# Demonstration: cache invalidation on invariant change
print(f"Before: {cache.stats()['size']} entries")

# Add a new invariant
engine.add_rule(Invariant(
    name="no_tmp_writes",
    category="system",
    constraint="IMPLIES(str.prefixof('/tmp/', file_path), operation != 'write')",
    variables={
        "file_path": {"type": "string", "domain": "absolute_path"},
        "operation": {"type": "string", "domain": "file_operation"},
    },
    description="Prevent writing to /tmp/ (agent should use its own workspace)",
))

# The invariant hash has changed -- old cache entries will miss naturally
new_hash = engine.get_invariant_set_hash()
print(f"Invariant hash changed: {inv_hash[:8]}... -> {new_hash[:8]}...")

# Flush stale entries to free memory
cache.invalidate()
print(f"After flush: {cache.stats()['size']} entries")

---

Chapter 6: OpenClaw Plugin Integration#

The Plugin Manifest#

Every OpenClaw plugin ships with a manifest.json that declares its capabilities, permissions, lifecycle hooks, and pricing. The formal gatekeeper's manifest registers it as a safety plugin with pre-execute hook permissions and a verification skill.

python

MANIFEST = {
    "name": "formal-gatekeeper",
    "version": "1.0.0",
    "description": (
        "Z3 SMT solver-based formal verification for agent plans. "
        "Intercepts all shell commands and transactions, proves safety "
        "against configurable invariants, and returns mathematical proofs "
        "or counter-examples."
    ),
    "author": "formal-gatekeeper-agent",
    "license": "MIT",
    "compatibility": ["openclaw>=0.9"],
    "permissions": [
        "lifecycle:onPreExecute",   # Intercept before execution
        "tools:verify_logic",       # Register verification tool
        "tools:verify_plan",        # Register plan verification tool
        "events:subscribe",         # Subscribe to agent events
    ],
    "hooks": {
        "onPreExecute": {
            "handler": "tools.pre_execute_hook",
            "priority": 100,          # Run before other plugins
            "timeout_ms": 10000,       # Max time for verification
            "fail_mode": "reject",     # Reject if hook fails
        },
    },
    "skills": [
        {
            "name": "verify_logic",
            "description": "On-demand formal verification of structured actions",
            "parameters": {
                "actions": {
                    "type": "array",
                    "description": "List of action dicts to verify",
                    "required": True,
                },
                "context": {
                    "type": "object",
                    "description": "Agent context (balance, gas spent, etc.)",
                    "required": False,
                },
            },
        },
        {
            "name": "verify_plan",
            "description": "Formal verification of a PLAN.md document",
            "parameters": {
                "plan_text": {
                    "type": "string",
                    "description": "Raw Markdown plan document",
                    "required": True,
                },
            },
        },
    ],
    "pricing": {
        "model": "per_verification",
        "price": "0.05",
        "currency": "USDC",
        "protocol": "x402",
    },
    "configuration": {
        "invariant_config_url": {
            "type": "string",
            "description": "URL to fetch invariant configuration",
            "default": "",
        },
        "solver_timeout_ms": {
            "type": "integer",
            "description": "Z3 solver timeout in milliseconds",
            "default": 5000,
        },
        "cache_size": {
            "type": "integer",
            "description": "Maximum proof cache entries",
            "default": 1024,
        },
        "cache_ttl_seconds": {
            "type": "integer",
            "description": "Proof cache TTL in seconds",
            "default": 300,
        },
        "x402_enabled": {
            "type": "boolean",
            "description": "Require x402 payment for each verification",
            "default": True,
        },
    },
}

print(json.dumps(MANIFEST, indent=2))

Plugin Mode: The Pre-Execute Hook#

In plugin mode, the formal gatekeeper runs automatically on every agent action via the onPreExecute lifecycle hook. The hook receives the action the agent is about to execute, runs it through the full verification pipeline, and either approves or rejects it.

python

def pre_execute_hook(event: dict, config: dict) -> dict:
    """OpenClaw onPreExecute hook handler.

    Called automatically before every agent action. Parses the action,
    translates it to Z3 formulas, verifies against invariants, and
    returns approve/reject with proof metadata.
    """
    action_text = event.get("action_text", "")
    action_type = event.get("action_type", "shell")
    agent_id = event.get("agent_id", "")

    # Load invariant engine with configured rules
    inv_engine = InvariantEngine(agent_id)
    for inv in filesystem_invariants + economic_invariants + network_invariants:
        inv_engine.add_rule(inv)

    inv_hash = inv_engine.get_invariant_set_hash()

    # Parse and translate the action
    parser = PlanParser()
    translator = PlanTranslator(inv_engine)

    # Wrap the action in a code block if it is a raw shell command
    if action_type == "shell" and not action_text.startswith("```"):
        plan_text = f"```bash\n{action_text}\n```"
    else:
        plan_text = action_text

    actions = parser.parse_plan(plan_text)
    if not actions:
        return {"verdict": "approve", "reason": "No verifiable actions detected"}

    context = event.get("context", {})
    formulas = translator.translate_plan(actions, context)

    # Verify each formula
    all_safe = True
    proof_results = []
    for formula in formulas:
        result = cached_verify(
            agent_id,
            formula,
            inv_engine.get_all_invariants(),
            inv_hash,
            solver_timeout_ms=config.get("solver_timeout_ms", 5000),
        )
        proof_results.append(result)
        if result.get("verdict") != "safe":
            all_safe = False
            break

    if all_safe:
        return {
            "verdict": "approve",
            "proof_count": len(proof_results),
            "total_solver_time_ms": sum(
                r.get("solver_time_ms", 0) for r in proof_results
            ),
            "cache_hits": sum(1 for r in proof_results if r.get("cached")),
        }
    else:
        failing = proof_results[-1]
        return {
            "verdict": "reject",
            "reason": f"Invariant violation: {failing.get('violated_invariants', [])}",
            "counter_example": failing.get("counter_example"),
            "fix_hints": failing.get("fix_hints", []),
            "proof_id": failing.get("proof_id"),
        }

# Simulate the hook firing for a dangerous command
dangerous_event = {
    "agent_id": "cat-sitter-agent",
    "action_type": "shell",
    "action_text": "rm -rf /etc/nginx/conf.d/default.conf",
    "context": {"current_balance": "5000.00"},
}

hook_result = pre_execute_hook(dangerous_event, {"solver_timeout_ms": 5000})
print(f"Hook verdict: {hook_result['verdict']}")
# Hook verdict: reject

# Safe command passes
safe_event = {
    "agent_id": "cat-sitter-agent",
    "action_type": "shell",
    "action_text": "git push origin main",
    "context": {"current_balance": "5000.00"},
}

safe_result = pre_execute_hook(safe_event, {"solver_timeout_ms": 5000})
print(f"Hook verdict: {safe_result['verdict']}")
# Hook verdict: approve

Skill Mode: On-Demand Verification#

Skill mode exposes verify_logic and verify_plan as tools that agents or operators can call proactively. This is useful for pre-flight checks before submitting a plan, or for third-party agents that want to verify actions before purchasing a service.

python

def verify_plan_skill(plan_text: str, context: dict | None = None) -> dict:
    """On-demand plan verification skill.

    Called explicitly by agents who want to pre-verify a plan
    before executing it. Returns the full proof result.
    """
    agent_id = context.get("agent_id", "anonymous") if context else "anonymous"
    ctx = context or {"current_balance": "0"}

    inv_engine = InvariantEngine(agent_id)
    for inv in filesystem_invariants + economic_invariants + network_invariants:
        inv_engine.add_rule(inv)

    compound = translate_compound_plan(plan_text, ctx, inv_engine)

    if compound["status"] == "empty_plan":
        return {
            "verdict": "safe",
            "reason": "Empty plan — nothing to verify.",
            "action_count": 0,
        }

    inv_hash = inv_engine.get_invariant_set_hash()
    results = []
    for formula in compound["formulas"]:
        r = cached_verify(
            agent_id,
            formula,
            inv_engine.get_all_invariants(),
            inv_hash,
        )
        results.append(r)
        if r.get("verdict") != "safe":
            break

    overall = "safe" if all(r.get("verdict") == "safe" for r in results) else "unsafe"

    return {
        "verdict": overall,
        "action_count": compound["action_count"],
        "formula_count": compound["formula_count"],
        "cumulative_spend": compound["cumulative_spend"],
        "results": results,
    }

# An agent pre-verifies its deployment plan
deployment_plan = """
```bash
pip install -r requirements.txt
python -m pytest tests/ -x

Transfer 50 USDC for compute costs.

bash

curl https://10.0.0.5:8080/deploy -X POST

"""

pre_check = verify_plan_skill( deployment_plan, context={ "agent_id": "deploy-bot", "current_balance": "1000.00", "daily_spent": "0.00", "starting_daily_balance": "1000.00", }, )

print(f"Pre-flight check: {pre_check['verdict']}") print(f" Actions: {pre_check['action_count']}") print(f" Spend: ${pre_check['cumulative_spend']}")

text

---

## Chapter 7: The x402 Payment Hook

### Micro-Payment Gate for Verification

Each formal verification consumes server-side compute (Z3 solver time). The x402 payment hook gates access to the verification service behind a micro-payment of 0.05 USDC per verification. This creates a sustainable revenue model for gatekeeper operators and prevents abuse (an attacker cannot flood the solver with garbage plans without paying for each attempt).

The payment flow integrates with the GreenHelix escrow system:

1. Before verification, create a payment intent for 0.05 USDC.
2. Submit the intent ID along with the verification request.
3. The gateway holds the payment in escrow during verification.
4. On successful verification (any verdict), the payment is released to the gatekeeper operator.
5. If the verification fails (solver error, timeout in the gateway), the payment is refunded.

```python
X402_VERIFICATION_FEE = "0.05"
X402_CURRENCY = "USDC"

def create_verification_payment(payer_agent_id: str, gatekeeper_agent_id: str) -> dict:
    """Create an x402 payment intent for a single verification."""
    intent = session.post(f"{API_BASE}/v1/payments/intents", json={
        "payer_agent_id": payer_agent_id,
        "payee_agent_id": gatekeeper_agent_id,
        "amount": X402_VERIFICATION_FEE,
        "currency": X402_CURRENCY,
        "description": "formal-verification-fee",
        "metadata": {
            "service": "formal-gatekeeper",
            "version": "1.0.0",
        },
    }).json()
    return intent

def verify_with_payment(
    payer_agent_id: str,
    gatekeeper_agent_id: str,
    formula: dict,
    invariants: list[dict],
    invariant_hash: str,
    config: dict,
) -> dict:
    """Full verification flow with x402 payment gate.

    1. Check cache (no payment needed for cached results)
    2. Create payment intent
    3. Submit verification with intent ID
    4. Return result
    """
    # Check cache first -- cached results are free
    cached_result = cache.get(formula, invariant_hash)
    if cached_result is not None:
        cached_result["cached"] = True
        cached_result["payment_charged"] = False
        return cached_result

    # Free tier bypass for development
    x402_enabled = config.get("x402_enabled", True)

    payment_intent_id = ""
    if x402_enabled:
        intent = create_verification_payment(payer_agent_id, gatekeeper_agent_id)
        payment_intent_id = intent.get("intent_id", "")

    # Submit verification with payment
    result = session.post(f"{API_BASE}/v1/gatekeeper/jobs", json={
        "agent_id": gatekeeper_agent_id,
        "formula": formula,
        "invariants": invariants,
        "solver_timeout_ms": config.get("solver_timeout_ms", 5000),
        "payment_intent_id": payment_intent_id,
    }).json()

    result["cached"] = False
    result["payment_charged"] = x402_enabled

    # Cache the result
    cache.put(formula, invariant_hash, result)

    return result

# Example: paid verification
paid_result = verify_with_payment(
    payer_agent_id="deploy-bot",
    gatekeeper_agent_id="formal-gatekeeper-v1",
    formula=test_formula,
    invariants=engine.get_all_invariants(),
    invariant_hash=engine.get_invariant_set_hash(),
    config={"x402_enabled": True, "solver_timeout_ms": 5000},
)

print(f"Verdict: {paid_result['verdict']}")
print(f"Payment charged: {paid_result['payment_charged']}")
print(f"Cached: {paid_result['cached']}")

# Subsequent calls hit cache -- no payment charged
cached = verify_with_payment(
    payer_agent_id="deploy-bot",
    gatekeeper_agent_id="formal-gatekeeper-v1",
    formula=test_formula,
    invariants=engine.get_all_invariants(),
    invariant_hash=engine.get_invariant_set_hash(),
    config={"x402_enabled": True},
)
print(f"Second call: cached={cached['cached']}, payment={cached['payment_charged']}")
# Second call: cached=True, payment=False

Revenue Tracking#

The gatekeeper operator can track verification revenue through the GreenHelix billing API.

python

# Check gatekeeper revenue
agent_id = "formal-gatekeeper-v1"
revenue = session.get(f"{API_BASE}/v1/billing/wallets/{agent_id}/usage",
    params={"period": "30d"}).json()

print(f"30-day revenue: ${revenue.get('total_earned', 0):.2f}")
print(f"Total verifications: {revenue.get('total_transactions', 0)}")
print(f"Avg revenue per verification: "
      f"${float(revenue.get('total_earned', 0)) / max(revenue.get('total_transactions', 1), 1):.4f}")

# Check billing by service type
usage = session.get(f"{API_BASE}/v1/billing/wallets/{agent_id}/analytics",
    params={"period": "7d"}).json()
print(f"Weekly usage: {json.dumps(usage, indent=2)}")

---

Chapter 8: Building the SKILL.md#

YAML Frontmatter for the OpenClaw Registry#

The SKILL.md file is the packaging format for publishing the formal gatekeeper to the OpenClaw skill registry and ClawhHub. The YAML frontmatter declares metadata, dependencies, pricing, and configuration.

yaml

---
name: greenhelix-formal-gatekeeper
version: "1.0.0"
description: "Z3 SMT solver-based formal verification for autonomous agent plans.
  Proves filesystem, economic, and network safety properties with mathematical certainty.
  Returns counter-examples for rejected actions."
license: MIT
compatibility: [openclaw]
author: formal-gatekeeper-agent
type: plugin
tags: [formal-verification, z3, smt, safety, security, openclaw, plugin]
price_usd: 0.05
pricing_model: per_call
currency: USDC
dependencies:
  - greenhelix-gateway>=1.0
  - z3-solver>=4.12 (server-side, no local install needed)
---

Operational Instructions#

The SKILL.md body contains the operational instructions that tell agents how the gatekeeper works and how to interact with it.

markdown

# Formal Gatekeeper

You are a safety verification plugin. Your role is to formally verify
every action an agent proposes before it executes.

## How It Works

1. When an agent proposes an action, the `onPreExecute` hook fires.
2. The action is parsed from shell commands, transactions, or network
   operations into structured forms.
3. Each action is translated into a Z3 SMT formula.
4. The formula is checked against safety invariants (filesystem,
   economic, network).
5. If the Z3 solver proves the action safe (UNSAT), it is approved.
6. If the solver finds a violation (SAT), the action is rejected with
   a counter-example and fix hints.

## Available Tools

### verify_plan
Verify a complete PLAN.md document.

**Parameters:**
- `plan_text` (string, required): The Markdown plan to verify.
- `context` (object, optional): Agent context including current_balance,
  daily_spent, etc.

**Returns:** ProofResult with verdict, counter_example, and fix_hints.

### verify_logic
Verify a list of structured actions.

**Parameters:**
- `actions` (array, required): List of action dicts with action_type,
  command, paths_accessed, etc.
- `invariant_hash` (string, optional): Hash of the invariant set for
  cache keying.

**Returns:** ProofResult with verdict, counter_example, and fix_hints.

## Safety Invariants

The following invariants are enforced by default:

### System Safety
- No access to /etc/, /root/, /proc/, /sys/, /boot/, /dev/
- Log files (/var/log/) are read-only

### Economic Safety
- No transaction exceeding 10% of current balance
- Daily gas cap of $5.00
- No negative balance
- Daily cumulative spending cap of 30% of starting balance

### Network Safety
- All connections must target whitelisted CIDRs
- No anonymization TLDs (.onion, .i2p)
- Outbound connections restricted to ports 80, 443, 22, 53, 8000-9000

## Pricing

- 0.05 USDC per verification (x402 protocol)
- Cached results are free
- Development/testing: set x402_enabled=false in configuration

Publishing to ClawhHub#

The skill is published using the same workflow as all other GreenHelix guides -- the create-clawhub-skills.py script handles it automatically. The script reads the `.met