Skip to main content

Proof-Oriented Event Sourcing — verify aggregates using property-based testing and state exploration

Project description

POES — Proof-Oriented Event Sourcing (Experimental)

This project is currently in an experimental stage, written by Claude Code. We do not recommending running this into production. This is also a coding agent first framework and skill. It is intended to be run in Claude Code, or any other coding agent. The proof of work is for the human.

We are looking for feedback on its usefulness and shortfalls. We also welcome pull requests, including those by coding agents.

Write your domain logic once. Verify it automatically.

POES is a Python library that verifies event-sourced aggregates using property-based testing, exhaustive state exploration, and temporal property checking — no external SMT solvers or model checkers required.

Installation

pip install poes

Or install from source:

git clone https://github.com/kurrent-io/poes.git
cd poes
pip install -e .

Optional dependencies

pip install poes[dev]        # pytest for running tests
pip install poes[kurrentdb]  # KurrentDB persistence support

Requirements

  • Python 3.10+
  • hypothesis (installed automatically)

Quick Start

from dataclasses import dataclass, replace
import hypothesis.strategies as st
from poes import Check

@dataclass(frozen=True)
class BankAccount:
    balance: int = 0
    is_open: bool = True

result = (
    Check.define("BankAccount", BankAccount)
    .with_initial(BankAccount(balance=0, is_open=True))
    .with_field("balance", st.integers(0, 1000))
    .with_field("is_open", st.booleans())
    .with_invariant("BalanceNonNegative", lambda s: s.balance >= 0)
    .with_invariant("BalanceBounded", lambda s: s.balance <= 1000)
    .with_parametric_transition("Deposit",
        params={"amount": st.integers(1, 500)},
        guard=lambda s, amount: s.is_open and s.balance + amount <= 1000,
        apply=lambda s, amount: replace(s, balance=s.balance + amount),
        ensures=lambda before, after, amount: after.balance == before.balance + amount)
    .with_transition("Withdraw",
        guard=lambda s: s.is_open and s.balance >= 50,
        apply=lambda s: replace(s, balance=s.balance - 50),
        ensures=lambda before, after: after.balance == before.balance - 50)
    .run()
)

assert result.all_passed

Output:

╔══════════════════════════════════════════════════════════════════════════════╗
║  VERIFYING: BankAccount
╚══════════════════════════════════════════════════════════════════════════════╝

  ┌─ Property Testing ──────────────────────────────────────────────────────────┐
  │  ✓ Deposit/BalanceNonNegative  (500 examples)
  │  ✓ Deposit/BalanceBounded  (500 examples)
  │  ✓ Withdraw/BalanceNonNegative  (500 examples)
  │  ✓ Withdraw/BalanceBounded  (500 examples)
  └─────────────────────────────────────────────────────────────────────────────┘

  ┌─ State Exploration ─────────────────────────────────────────────────────────┐
  │  ✓ All 22 reachable states safe
  └─────────────────────────────────────────────────────────────────────────────┘

  ✓ VERIFIED: All 4 proofs passed, 22 states explored (150ms)

What It Verifies

POES proves four categories of properties about your event-sourced aggregates:

  1. Property Testing — Hypothesis generates random states and checks that every transition preserves every invariant
  2. State Space Safety — BFS explores all reachable states from the initial state and verifies all invariants hold
  3. Temporal Properties — Liveness checks like "eventually", "leads-to", and "always-eventually" using SCC analysis
  4. Persistence Verification — Replays production events from KurrentDB and checks every intermediate state against all invariants

Built for Coding Agents

POES is designed so AI coding agents can verify the code they generate. An agent produces an event-sourced aggregate and runs POES verification in the same step — the result is deterministic and machine-readable. When a future change breaks a previously proven property, the agent gets a minimal counterexample showing exactly what broke.

Claude Code skill

This repo includes a Claude Code skill at .claude/skills/poes/SKILL.md. To use it:

  1. Clone this repo (or copy the .claude/skills/poes/ folder into your project)
  2. Register the skill in your project's .claude/settings.json:
{
  "skills": [
    ".claude/skills/poes/SKILL.md"
  ]
}

The skill teaches Claude Code the full POES API so it can write and verify aggregates for you.

API Overview

from poes import Check, FrozenMap

builder = (
    Check.define("Name", StateClass)
    .with_initial(initial_state)
    .with_field("field", strategy)
    .with_map_field("map_field", keys_strat, values_strat)
    .with_invariant("Name", predicate)
    .with_transition("Event", guard, apply, ensures)
    .with_parametric_transition("Event", params, guard, apply, ensures)
    .with_eventually("Name", predicate)
    .with_leads_to("Name", trigger, response)
    .with_always_eventually("Name", predicate)
    .expect_transitions(count)
    .with_max_examples(n)
)

result = builder.run()                          # Run verification
builder.generate_proof_of_work(path="proof.md") # Generate proof document

See the skill file for full API reference, templates, and common mistakes.

Samples

Sample Description
bank_account.py Deposits, withdrawals, balance bounds
gambling_wallet.py Casino betting with active bet tracking
inventory.py Warehouse stock with reservations
order_book.py Map-based state using FrozenMap
gift_card.py Full KurrentDB integration
hotel_reservation.py State diagram with temporal properties

Run a sample:

python samples/bank_account.py

Project Structure

src/poes/
├── __init__.py            # Package exports
├── check.py               # Check.define fluent API
├── frozen_map.py          # FrozenMap — hashable immutable dict
├── hypothesis_bridge.py   # Hypothesis property-based testing
├── explorer.py            # BFS state exploration
├── temporal.py            # SCC-based liveness checking
├── persistence_check.py   # Production data verification
├── repository.py          # KurrentDB persistence (Repository pattern)
└── docgen.py              # Markdown proof document generation

License

MIT

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

poes-0.2.0.tar.gz (34.0 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

poes-0.2.0-py3-none-any.whl (20.4 kB view details)

Uploaded Python 3

File details

Details for the file poes-0.2.0.tar.gz.

File metadata

  • Download URL: poes-0.2.0.tar.gz
  • Upload date:
  • Size: 34.0 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.12.9

File hashes

Hashes for poes-0.2.0.tar.gz
Algorithm Hash digest
SHA256 3ea6efdf5ed62fc139ab976ea3b24373e02d9dd2bfa5758f40892c4d9a18c173
MD5 07a3c4665223746fa17359ce1a8eef8c
BLAKE2b-256 8454798776a5372224032a4e1607c20c678e9ea0b2ee20ebc8cad601512e5293

See more details on using hashes here.

File details

Details for the file poes-0.2.0-py3-none-any.whl.

File metadata

  • Download URL: poes-0.2.0-py3-none-any.whl
  • Upload date:
  • Size: 20.4 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.12.9

File hashes

Hashes for poes-0.2.0-py3-none-any.whl
Algorithm Hash digest
SHA256 46ab1e04d31dc4c4a96aa91237b554de85e14aa898b75b2be7bc47d327df7ef2
MD5 1801145beaabb5d7f88ed388b5ba2225
BLAKE2b-256 4bb164829755e4d83e4af410ff235eee91cf16607a70839308ad2bc53e6a7a6f

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page