Skip to main content

Writing Arbiters (pt 2 - Off-chain Oracles)

Off-chain oracles are arbitration services that run outside the blockchain but submit their decisions on-chain via the TrustedOracleArbiter contract. Unlike on-chain arbiters (see pt 1) that validate within smart contract execution, oracles perform arbitration in an external environment where they can access APIs, run complex computations, maintain state databases, or integrate with external systems.

This guide covers implementing oracle services using the Alkahest SDKs. The patterns here apply to building production oracle infrastructure that can validate work submissions against arbitrary criteria.

Understanding Your Role as an Oracle​

Example scenario: Alice wants Bob to capitalize a string ("hello world" → "HELLO WORLD"). She creates an escrow offering payment, specifying you (Charlie) as the trusted oracle who will verify if Bob's work is correct. Bob submits his result, requests your arbitration, and if you approve, he can claim the payment.

Your oracle service needs to:

  1. Listen for arbitration requests that specify you as the oracle
  2. Extract the fulfillment data and original demand from attestations
  3. Validate the fulfillment according to the demand criteria
  4. Submit your approval or rejection decision on-chain

The TrustedOracleArbiter contract handles the on-chain logic - your job is to implement the validation logic and submit decisions.

For a complete example of how oracles fit into the escrow/fulfillment flow, see "Escrow Flow (pt 2 - Job Trading)".

Three Validation Patterns​

PatternReturnsStateEscrow AccessUse Case
ContextlessboolOracle maintains stateNoSignature verification, identity validation, format checking
Demand-BasedboolStatelessYes - reads demandCustom validation per escrow, test case validation
AsynchronousnullJob queueYes - reads demandTime-based monitoring, long-running computations

Decision flowchart:

  • Does validation require waiting over time?
    • Yes → Asynchronous
    • No → Does validation need the escrow's demand parameters?
      • Yes → Demand-Based
      • No → Contextless

Pattern 1: Contextless Validation​

Contextless oracles validate fulfillments based purely on the fulfillment's intrinsic properties and the oracle's own maintained state, without referencing the original escrow demand. This pattern is useful for building reusable validation services that work across any escrow.

When to use: Signature verification, format checking, identity validation, standard verification against a maintained registry.

Why contextless: The oracle provides a generic service (e.g., "I verify signatures from known identities") rather than validating against buyer-specific criteria. The validation logic doesn't depend on what Alice requested - only on what Bob submitted.

Composability: Because contextless oracles are generic and reusable, they can be easily composed with other arbiters using logical combinators like AllArbiter and AnyArbiter. For example, you could require that a fulfillment is both signed by a verified identity (contextless oracle) AND meets specific job criteria (demand-based oracle). See "Escrow Flow (pt 3 - Composing Demands)" for composition patterns.

Reference implementations:

Step 1: Define fulfillment format and registry state​

Define what fulfillments look like and what state you maintain:

import json
from dataclasses import dataclass
from typing import Dict
from eth_account import Account
from eth_account.messages import encode_defunct

@dataclass
class IdentityFulfillment:
    pubkey: str
    nonce: int
    data: str
    signature: str

# Oracle's internal registry (identity address -> current nonce)
# This represents the oracle's concept of which identities are valid
# and tracks nonces to prevent replay attacks
identity_registry: Dict[str, int] = {}

Step 2: Initialize your registry​

Before starting the listener, populate your oracle's state:

async def run_contextless_oracle(oracle_client):
    # Register known identities with starting nonces
    identity_registry.clear()
    identity_registry[identity_address_1.lower()] = 0
    identity_registry[identity_address_2.lower()] = 0
    # In production: load from database

    # ... rest of oracle setup ...

Step 3: Implement the validation callback​

The validation callback receives the fulfillment attestation and the demand data, then checks it against your registry:

def verify_identity_decision(attestation, demand) -> bool:
    """
    Verify an identity fulfillment by checking:
    1. The signature is valid
    2. The nonce is greater than the last seen nonce
    3. The recovered address matches the claimed pubkey
    """
    try:
        # Step 3a: Extract fulfillment data
        obligation_str = oracle_client.oracle.extract_obligation_data(attestation)
        payload = json.loads(obligation_str)

        parsed = IdentityFulfillment(
            pubkey=payload['pubkey'],
            nonce=payload['nonce'],
            data=payload['data'],
            signature=payload['signature']
        )

        # Step 3b: Check against oracle's registry
        pubkey_lower = parsed.pubkey.lower()
        if pubkey_lower not in identity_registry:
            return False  # Unknown identity - not in our registry

        # Step 3c: Verify nonce progression (replay protection)
        current_nonce = identity_registry[pubkey_lower]
        if parsed.nonce <= current_nonce:
            return False

        # Step 3d: Verify signature format
        sig_len = len(parsed.signature) if parsed.signature else 0
        if not parsed.signature or sig_len not in [130, 132]:
            return False

        # Step 3e: Verify cryptographic signature
        message = f"{parsed.data}:{parsed.nonce}"
        encoded_message = encode_defunct(text=message)

        try:
            recovered = Account.recover_message(encoded_message, signature=parsed.signature)
        except Exception:
            return False

        if recovered.lower() != pubkey_lower:
            return False  # Signature mismatch

        # Step 3f: Update state and approve
        identity_registry[pubkey_lower] = parsed.nonce
        return True

    except Exception:
        return False

Step 4: Set up listener and cleanup​

Wire everything together with the listener:

from alkahest_py import ArbitrationMode

def callback(decision):
    # Optional: called after arbitration is submitted on-chain
    print(f"Arbitrated {decision.attestation.uid}: {decision.decision}")

# Listen and validate
decisions = await oracle_client.oracle.arbitrate_many(
    verify_identity_decision,
    callback,
    ArbitrationMode.AllUnarbitrated,
    timeout_seconds=10.0,
)

# Cleanup
identity_registry.clear()

Complete pattern:

  1. Define fulfillment format and registry state
  2. Initialize your oracle's registry/database
  3. Implement validation callback that:
    • Extracts fulfillment data
    • Checks against oracle's internal state
    • Performs validation (e.g., signature checks)
    • Updates state if needed
    • Returns true or false
  4. Set up listener with callback and cleanup

Pattern 2: Demand-Based Validation​

Demand-based oracles validate fulfillments against specific criteria provided by the buyer in the escrow demand. Each escrow can specify different requirements, and the oracle validates that Bob's fulfillment meets Alice's exact specifications.

When to use: Custom validation criteria per escrow, need to compare fulfillment against buyer's specifications, computational validation with test cases.

Why demand-based: Alice specifies exactly what she wants (e.g., "capitalize these specific test cases"), and the oracle verifies Bob's work matches those requirements. Different escrows have different demands, all validated by the same oracle.

Example flow:

  1. Alice creates escrow → demands oracle=charlie, data="capitalize hello world", offers 100 tokens
  2. Bob fulfills → submits "HELLO WORLD", references Alice's escrow via refUID
  3. Bob requests arbitration → asks Charlie to validate
  4. Charlie validates → extracts Bob's result and Alice's query, checks if "HELLO WORLD" matches uppercase("hello world"), submits decision on-chain
  5. Bob claims payment → uses approved attestation to collect escrow

Reference implementations:

Step 1: Define your demand format​

Decide what parameters buyers provide in their escrow demands:

import json
import subprocess
from dataclasses import dataclass
from typing import List

@dataclass
class ShellTestCase:
    input: str
    output: str

@dataclass
class ShellOracleDemand:
    description: str
    test_cases: List[ShellTestCase]

Buyers encode this as JSON in the TrustedOracleArbiter demand's data field.

Step 2: Implement validation with demand​

The callback receives both the fulfillment attestation and the demand data directly. The demand is passed from the ArbitrationRequested event, so there's no need to fetch the escrow from chain.

async def decision_function(attestation, demand):
    """Evaluate whether the fulfillment meets the demand requirements"""
    try:
        # Step 2a: Extract fulfillment
        statement = oracle_client.oracle.extract_obligation_data(attestation)
    except Exception as e:
        print(f"Failed to extract obligation: {e}")
        return False

    # Step 2b: Parse demand from callback argument
    try:
        demand_json = json.loads(bytes(demand).decode('utf-8'))
    except Exception as e:
        print(f"Failed to parse demand: {e}")
        return False

    # Step 2c: Apply validation logic using demand parameters
    # Run each test case to verify Bob's submission works correctly
    for case in demand_json['test_cases']:
        command = f'echo "$INPUT" | {statement}'
        try:
            result = subprocess.run(
                ["bash", "-c", command],
                env={"INPUT": case['input']},
                capture_output=True,
                text=True,
                timeout=5
            )
            if result.returncode != 0:
                return False
            output = result.stdout.rstrip('\n')
            if output != case['output']:
                return False
        except Exception:
            return False

    return True

Step 3: Understanding the data flow​

ArbitrationRequested Event
  └─ obligation: fulfillment UID
  └─ oracle: charlie_address
  └─ demand: bytes (your custom demand format)
        │
        ▼
  Callback receives (attestation, demand)
        │
        ├─ attestation: Fulfillment Attestation
        │    └─ data: StringObligation { item: "tr '[:lower:]' '[:upper:]'" }
        │    └─ refUID: points to escrow
        │
        └─ demand: bytes (passed directly from event)
             └─ Your custom format (e.g., JSON with test_cases)

The demand data is passed directly to your callback via the ArbitrationRequested event, so you don't need to fetch the escrow attestation from chain.

Step 4: Set up the listener​

from alkahest_py import ArbitrationMode

def callback(decision):
    """Called when arbitration completes"""
    print(f"Arbitrated {decision.attestation.uid}: {decision.decision}")

# Listen and arbitrate
decisions = await oracle_client.oracle.arbitrate_many(
    decision_function,
    callback,
    ArbitrationMode.AllUnarbitrated,
    timeout_seconds=10.0,
)

# Verify decisions
assert all(d.decision for d in decisions), "Oracle rejected fulfillment"

Complete pattern:

  1. Define demand format (your oracle's API)
  2. Implement validation callback:
    • Extract fulfillment data from the attestation
    • Parse demand data from the callback argument
    • Apply validation logic comparing fulfillment to demand
    • Return true or false
  3. Set up listener with callback

Pattern 3: Asynchronous Validation​

Asynchronous oracles handle validation that cannot complete immediately. They require monitoring over time, accumulating data, or waiting for external conditions. The oracle schedules work for later, and a background worker submits the decision when ready.

When to use: Time-based monitoring (uptime checks, deadline validation), accumulating evidence over multiple observations, long-running computations, waiting for consensus from multiple sources.

Why asynchronous: Some validation is inherently time-based. For example, "verify this server stays up for 24 hours" cannot be validated instantly - you must schedule checks over time and make a final decision later.

Architecture:

Listener (receives requests) → Job Queue (stores pending work)
                                     ↓
                                Worker (processes jobs over time)
                                     ↓
                                On-chain Submission (final decision)

Reference implementations:

Step 1: Define demand format and job state​

Define both the demand buyers will provide and the internal state for tracking scheduled work:

import asyncio
import json
from dataclasses import dataclass, field
from typing import Dict, List, Optional

@dataclass
class UptimeDemand:
    service_url: str
    min_uptime: float        # Required uptime percentage (0.0-1.0)
    start: int               # Start time for monitoring window
    end: int                 # End time for monitoring window
    check_interval_secs: int

@dataclass
class PingEvent:
    delay_secs: float
    success: bool

@dataclass
class UptimeJob:
    min_uptime: float
    schedule: List[PingEvent]
    demand: bytes

# Shared state between listener and worker
@dataclass
class SchedulerContext:
    job_db: Dict[str, UptimeJob] = field(default_factory=dict)
    url_index: Dict[str, str] = field(default_factory=dict)
    notify: Optional[asyncio.Event] = field(default_factory=asyncio.Event)

Step 2: Implement the scheduling callback​

The listener callback schedules work but does not make a decision — it returns None (null) to defer the decision to the worker:

async def schedule_decision(attestation, demand):
    """Schedule uptime monitoring but DON'T make a decision yet."""
    try:
        # Step 2a: Extract service URL from fulfillment
        statement = oracle_client.oracle.extract_obligation_data(attestation)

        # Look up the fulfillment UID from our URL index
        uid = ctx.url_index.get(statement)
        if uid is None or uid in ctx.job_db:
            return None

        # Step 2b: Parse demand from callback argument
        demand_json = json.loads(bytes(demand).decode("utf-8"))

        # Verify URL matches
        if statement != demand_json["service_url"]:
            return None

        # Step 2c: Create monitoring schedule
        total_span = max(demand_json["end"] - demand_json["start"], 1)
        interval = max(demand_json["check_interval_secs"], 1)
        checks = max(total_span // interval, 1)

        schedule = []
        for i in range(checks):
            schedule.append(PingEvent(
                delay_secs=0.1 + i * 0.025,
                success=True,  # Will be determined by actual ping
            ))

        # Step 2d: Store job for background processing
        ctx.job_db[uid] = UptimeJob(
            min_uptime=demand_json["min_uptime"],
            schedule=schedule,
            demand=bytes(demand),
        )
        ctx.notify.set()  # Wake up worker

        # Step 2e: Return None to defer decision to the worker
        return None

    except Exception:
        return None

Step 3: Implement the background worker​

Create a worker that processes scheduled jobs and submits decisions:

async def run_worker(ctx: SchedulerContext, oracle_client) -> None:
    """Background worker that processes uptime jobs and submits decisions."""
    while True:
        # Step 3a: Get next job from queue
        uid = None
        job = None
        for k, v in list(ctx.job_db.items()):
            uid = k
            job = v
            del ctx.job_db[k]
            break

        if uid is not None and job is not None:
            # Step 3b: Execute the monitoring schedule
            successes = 0
            total_checks = max(len(job.schedule), 1)

            for ping in job.schedule:
                await asyncio.sleep(ping.delay_secs)
                # In production: actually ping the service
                if ping.success:
                    successes += 1

            # Step 3c: Calculate result and make decision
            uptime = successes / total_checks
            decision = uptime >= job.min_uptime

            # Step 3d: Submit decision on-chain
            # CRITICAL: Unlike synchronous oracles where arbitrate_many
            # handles submission automatically, async oracles must MANUALLY
            # call oracle.arbitrate() to submit their decision
            await oracle_client.oracle.arbitrate(uid, list(job.demand), decision)
        else:
            # Step 3e: Wait for new work
            ctx.notify.clear()
            try:
                await asyncio.wait_for(ctx.notify.wait(), timeout=2.0)
            except asyncio.TimeoutError:
                if not ctx.job_db:
                    break  # No work and no notifications - exit

Step 4: Wire everything together​

Set up the listener, worker, and shared state:

from alkahest_py import ArbitrationMode

# Step 4a: Initialize shared state
ctx = SchedulerContext()
ctx.url_index[service_url] = fulfillment_uid

# Step 4b: Start background worker
worker_task = asyncio.create_task(run_worker(ctx, oracle_client))

# Step 4c: Start listener (scheduling callback from Step 2)
decisions = await oracle_client.oracle.arbitrate_many(
    schedule_decision,
    callback,
    ArbitrationMode.AllUnarbitrated,
    timeout_seconds=10.0,
)

# Step 4d: Wait for worker and cleanup
await asyncio.wait_for(worker_task, timeout=10.0)

Complete asynchronous oracle pattern:

  1. Define demand format and job state structures
  2. Implement scheduling callback that:
    • Extracts fulfillment data
    • Parses demand from callback argument
    • Creates job schedule
    • Stores job in shared queue
    • Returns None/null (defers decision)
  3. Implement background worker that:
    • Polls job queue
    • Executes scheduled work
    • Makes decision based on results
    • Calls arbitrate() to submit decision on-chain
  4. Wire together with shared state and cleanup

Key differences from synchronous patterns:

AspectSynchronous (Patterns 1 & 2)Asynchronous (Pattern 3)
Callback return valuetrue or falseNone/null (defers decision)
Decision submissionAutomatic via SDKManual via arbitrate()
ArchitectureSingle callback functionCallback + background worker
State managementOptional (Pattern 1 only)Required (job queue)
TimingInstant validationDelayed validation over time

Choosing the Right Pattern​

Quick decision tree:

  1. Does validation require waiting/monitoring over time?
    • YES → Pattern 3: Asynchronous
  2. Does validation need the escrow's demand parameters?
    • YES → Pattern 2: Demand-Based
    • NO → Pattern 1: Contextless

Detailed comparison:

Validation TypePatternComplexityExample
Signature verificationContextlessLowVerify identity attestations
Format/standard checkingContextlessLowValidate JSON schemas
Test case validationDemand-BasedMediumRun buyer-specified tests
Computational verificationDemand-BasedMediumCheck algorithmic solutions
Uptime monitoringAsynchronousHighVerify 99% uptime over 24h
Consensus votingAsynchronousHighWait for multiple approvals

Production Considerations​

Error Handling and Logging​

Always handle errors gracefully and log for debugging:

import logging

logger = logging.getLogger(__name__)

# In validation callback
try:
    data = parse_data(attestation)
except Exception as e:
    logger.error(f"Failed to parse attestation {attestation.uid}: {e}")
    return False  # Reject invalid data

# Log decisions
logger.info(
    f"Oracle decision for {attestation.uid}: "
    f"{'approved' if decision else 'rejected'}"
)

State Persistence​

For async oracles, persist state to survive restarts:

import sqlite3
import json

class PersistentJobDb:
    def __init__(self, db_path: str):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS jobs (
                uid TEXT PRIMARY KEY,
                job TEXT NOT NULL
            )
        """)
        self.conn.commit()

    def insert_job(self, uid: str, job: UptimeJob):
        job_json = json.dumps({
            'fulfillment_uid': job.fulfillment_uid,
            'service_url': job.service_url,
            'min_uptime': job.min_uptime,
            'checks_remaining': job.checks_remaining,
            'successes': job.successes,
            'total_checks': job.total_checks,
        })
        self.conn.execute(
            "INSERT OR REPLACE INTO jobs (uid, job) VALUES (?, ?)",
            (uid, job_json)
        )
        self.conn.commit()

    def get_next_job(self):
        cursor = self.conn.execute("SELECT uid, job FROM jobs LIMIT 1")
        row = cursor.fetchone()
        if not row:
            return None

        uid, job_json = row
        job_data = json.loads(job_json)
        job = UptimeJob(**job_data)

        self.conn.execute("DELETE FROM jobs WHERE uid = ?", (uid,))
        self.conn.commit()

        return (uid, job)

Retry Logic​

Handle transaction failures gracefully:

import asyncio

async def submit_with_retry(
    oracle_client,
    uid: str,
    demand: bytes,
    decision: bool,
    max_attempts: int = 3
):
    for attempt in range(max_attempts):
        try:
            await oracle_client.oracle.arbitrate(uid, list(demand), decision)
            return
        except Exception as e:
            logger.warning(f"Arbitration attempt {attempt + 1} failed: {e}")
            if attempt < max_attempts - 1:
                await asyncio.sleep(2 ** attempt)

    raise Exception(f"Failed to submit arbitration after {max_attempts} attempts")

Reference Implementations​

See the full working examples in the test suites:

These tests demonstrate the complete flow including escrow creation (Alice), fulfillment submission (Bob), oracle validation (Charlie), and payment collection.