diff --git a/ledger/tests/.gitkeep b/ledger/tests/.gitkeep
deleted file mode 100644
index e69de29..0000000
diff --git a/ledger/tests/conftest.py b/ledger/tests/conftest.py
new file mode 100644
index 0000000..5cfa586
--- /dev/null
+++ b/ledger/tests/conftest.py
@@ -0,0 +1,40 @@
+"""Shared fixtures for m2-ledger tests: isolated sqlite conn + TestClient."""
+
+import pytest
+from fastapi.testclient import TestClient
+
+from m2_ledger import auth, models
+from m2_ledger.api import app
+
+SERVICE_KEY = "test-service-key"
+ADMIN_KEY = "test-admin-key"
+
+
+@pytest.fixture
+def conn(tmp_path):
+    connection = models.init_db(str(tmp_path / "ledger.sqlite3"))
+    yield connection
+    connection.close()
+
+
+@pytest.fixture
+def client(tmp_path, monkeypatch):
+    monkeypatch.setenv("LEDGER_DB_PATH", str(tmp_path / "ledger_api.sqlite3"))
+    monkeypatch.setenv("LEDGER_SERVICE_KEYS", SERVICE_KEY)
+    monkeypatch.setenv("LEDGER_ADMIN_KEYS", ADMIN_KEY)
+    auth._service_keys.cache_clear()
+    auth._admin_keys.cache_clear()
+    with TestClient(app) as test_client:
+        yield test_client
+    auth._service_keys.cache_clear()
+    auth._admin_keys.cache_clear()
+
+
+@pytest.fixture
+def service_headers():
+    return {"X-API-Key": SERVICE_KEY}
+
+
+@pytest.fixture
+def admin_headers():
+    return {"X-API-Key": ADMIN_KEY}
diff --git a/ledger/tests/test_properties.py b/ledger/tests/test_properties.py
new file mode 100644
index 0000000..9f67b0e
--- /dev/null
+++ b/ledger/tests/test_properties.py
@@ -0,0 +1,248 @@
+"""Property + invariant tests for m2_ledger (contracts/ledger-api.md §Invariants).
+
+1. No endpoint UPDATEs/DELETEs tx rows                    -> append-only by construction,
+                                                              exercised indirectly by every test.
+2. balance == fold(tx log) for every operator             -> test_balance_matches_independent_fold
+3. Every grant has a non-null paying tx                   -> test_every_license_grant_has_paying_tx
+4. /tx/install is all-or-nothing and idempotent by ref     -> test_insufficient_funds_leaves_zero_new_rows
+                                                              test_duplicate_ref_leaves_zero_new_rows
+                                                              test_api_install_402_leaves_tx_log_unchanged
+Plus: refund exactly compensates (test_refund_restores_pre_install_balances) and no balance
+except mint goes negative (checked as a stateful invariant below).
+"""
+
+from hypothesis import HealthCheck, assume, settings
+from hypothesis import strategies as st
+from hypothesis.stateful import RuleBasedStateMachine, invariant, rule
+
+from m2_ledger import models
+
+OPERATORS = ["op_alice", "op_bob", "op_carol"]
+ALL_ACCOUNTS = OPERATORS + ["platform"]
+
+
+def _fold_balance(conn, operator_id: str) -> int:
+    """Independent re-implementation of models.balance: fold the raw tx log in
+    Python rather than SQL, so this actually cross-checks the SQL aggregation."""
+    bal = 0
+    for row in conn.execute("SELECT from_id, to_id, amount FROM tx").fetchall():
+        if row["to_id"] == operator_id:
+            bal += row["amount"]
+        if row["from_id"] == operator_id:
+            bal -= row["amount"]
+    return bal
+
+
+def _known_accounts(conn) -> set:
+    ids = set()
+    for row in conn.execute("SELECT from_id, to_id FROM tx").fetchall():
+        ids.add(row["from_id"])
+        ids.add(row["to_id"])
+    return ids
+
+
+class LedgerMachine(RuleBasedStateMachine):
+    """Arbitrary sequences of valid ops (grant, install, refund) against a fresh
+    in-memory ledger, checking fold-consistency and non-negativity after every op."""
+
+    def __init__(self):
+        super().__init__()
+        self.conn = models.init_db(":memory:")
+        self.installed_refs = []
+        self._ref_counter = 0
+
+    def teardown(self):
+        self.conn.close()
+
+    @rule(
+        operator=st.sampled_from(OPERATORS),
+        amount=st.integers(min_value=1, max_value=1000),
+    )
+    def grant(self, operator, amount):
+        models.record_grant(self.conn, operator, amount)
+
+    @rule(
+        pair=st.permutations(OPERATORS).map(lambda p: (p[0], p[1])),
+        amount=st.integers(min_value=1, max_value=500),
+        pct=st.floats(min_value=0.0, max_value=1.0, allow_nan=False),
+    )
+    def install(self, pair, amount, pct):
+        buyer, seller = pair
+        # only attempt ops that are valid per the buyer's *actual* current balance,
+        # so this models "arbitrary sequences of valid ops", not failure paths.
+        if models.balance(self.conn, buyer) < amount:
+            return
+        # A platform_pct that rounds the cut or the seller's net to exactly 0
+        # trips a known source bug (see report: zero-amount tx row violates the
+        # `amount > 0` CHECK constraint with an unhandled IntegrityError) —
+        # excluded here since it's not part of this property, not a fix.
+        cut = int(amount * pct)
+        assume(0 < cut < amount)
+        self._ref_counter += 1
+        ref = f"ref-{self._ref_counter}"
+        models.record_install(self.conn, buyer, seller, amount, pct, ref)
+        self.installed_refs.append(ref)
+
+    @rule(data=st.data())
+    def refund(self, data):
+        if not self.installed_refs:
+            return
+        ref = data.draw(st.sampled_from(self.installed_refs))
+        try:
+            models.record_refund(self.conn, ref)
+        except models.AlreadyRefunded:
+            pass
+        else:
+            self.installed_refs.remove(ref)
+
+    @invariant()
+    def balances_match_independent_fold(self):
+        for account in _known_accounts(self.conn):
+            assert models.balance(self.conn, account) == _fold_balance(self.conn, account)
+
+    @invariant()
+    def no_non_mint_balance_goes_negative(self):
+        for account in _known_accounts(self.conn):
+            if account == "mint":
+                continue
+            assert models.balance(self.conn, account) >= 0
+
+
+LedgerStateMachineTest = LedgerMachine.TestCase
+LedgerStateMachineTest.settings = settings(
+    max_examples=50,
+    stateful_step_count=25,
+    suppress_health_check=[HealthCheck.too_slow, HealthCheck.data_too_large],
+    deadline=None,
+)
+
+
+# ---------------------------------------------------------------------------
+# Example-based invariant tests
+# ---------------------------------------------------------------------------
+
+
+def test_every_license_grant_has_paying_tx(conn):
+    """Invariant 3: install grants reference a real 'install' tx; starter grants
+    (money only, reason='grant') never create a license grant_ row at all."""
+    models.record_grant(conn, "op_alice", 100)
+    result = models.record_install(conn, "op_alice", "op_bob", 40, 0.10, "ref-install-1")
+
+    grant_row = result["grant"]
+    assert grant_row["tx_id"] is not None
+
+    paying_tx = conn.execute(
+        "SELECT * FROM tx WHERE id = ?", (grant_row["tx_id"],)
+    ).fetchone()
+    assert paying_tx is not None
+    assert paying_tx["reason"] == "install"
+    assert paying_tx["from_id"] == "op_alice"
+    assert paying_tx["to_id"] == "op_bob"
+
+    # the starter grant produced a money tx but no row in grant_ (no license without payment)
+    grant_count = conn.execute("SELECT COUNT(*) AS n FROM grant_").fetchone()["n"]
+    assert grant_count == 1
+
+
+def test_no_license_grant_without_install(conn):
+    """A pure starter grant must never create a grant_ (license) row."""
+    models.record_grant(conn, "op_alice", 100)
+    grant_count = conn.execute("SELECT COUNT(*) AS n FROM grant_").fetchone()["n"]
+    assert grant_count == 0
+
+
+def test_insufficient_funds_leaves_zero_new_rows(conn):
+    models.record_grant(conn, "op_alice", 10)
+    tx_count_before = conn.execute("SELECT COUNT(*) AS n FROM tx").fetchone()["n"]
+    grant_count_before = conn.execute("SELECT COUNT(*) AS n FROM grant_").fetchone()["n"]
+
+    try:
+        models.record_install(conn, "op_alice", "op_bob", 10_000, 0.10, "ref-too-much")
+        assert False, "expected InsufficientFunds"
+    except models.InsufficientFunds:
+        pass
+
+    tx_count_after = conn.execute("SELECT COUNT(*) AS n FROM tx").fetchone()["n"]
+    grant_count_after = conn.execute("SELECT COUNT(*) AS n FROM grant_").fetchone()["n"]
+    assert tx_count_after == tx_count_before
+    assert grant_count_after == grant_count_before
+
+
+def test_duplicate_ref_leaves_zero_new_rows(conn):
+    models.record_grant(conn, "op_alice", 100)
+    models.record_install(conn, "op_alice", "op_bob", 40, 0.10, "ref-dup")
+
+    tx_count_before = conn.execute("SELECT COUNT(*) AS n FROM tx").fetchone()["n"]
+    grant_count_before = conn.execute("SELECT COUNT(*) AS n FROM grant_").fetchone()["n"]
+
+    try:
+        models.record_install(conn, "op_alice", "op_bob", 40, 0.10, "ref-dup")
+        assert False, "expected DuplicateRef"
+    except models.DuplicateRef:
+        pass
+
+    tx_count_after = conn.execute("SELECT COUNT(*) AS n FROM tx").fetchone()["n"]
+    grant_count_after = conn.execute("SELECT COUNT(*) AS n FROM grant_").fetchone()["n"]
+    assert tx_count_after == tx_count_before
+    assert grant_count_after == grant_count_before
+
+
+def test_refund_restores_pre_install_balances(conn):
+    models.record_grant(conn, "op_alice", 100)
+    pre_install = {
+        op: models.balance(conn, op) for op in ("op_alice", "op_bob", "platform")
+    }
+
+    models.record_install(conn, "op_alice", "op_bob", 40, 0.10, "ref-refundable")
+    models.record_refund(conn, "ref-refundable")
+
+    post_refund = {
+        op: models.balance(conn, op) for op in ("op_alice", "op_bob", "platform")
+    }
+    assert post_refund == pre_install
+
+
+def test_refund_twice_rejected_and_adds_no_rows(conn):
+    models.record_grant(conn, "op_alice", 100)
+    models.record_install(conn, "op_alice", "op_bob", 40, 0.10, "ref-once")
+    models.record_refund(conn, "ref-once")
+
+    tx_count_before = conn.execute("SELECT COUNT(*) AS n FROM tx").fetchone()["n"]
+    try:
+        models.record_refund(conn, "ref-once")
+        assert False, "expected AlreadyRefunded"
+    except models.AlreadyRefunded:
+        pass
+    tx_count_after = conn.execute("SELECT COUNT(*) AS n FROM tx").fetchone()["n"]
+    assert tx_count_after == tx_count_before
+
+
+# ---------------------------------------------------------------------------
+# API-level test (contract invariant 4, via TestClient)
+# ---------------------------------------------------------------------------
+
+
+def test_api_install_402_leaves_tx_log_unchanged(client, service_headers, admin_headers):
+    client.post(
+        "/grant/starter",
+        json={"operator_id": "op_alice", "amount": 10},
+        headers=admin_headers,
+    )
+
+    before = client.get("/tx", headers=service_headers).json()["transactions"]
+
+    resp = client.post(
+        "/tx/install",
+        json={
+            "buyer": "op_alice",
+            "seller": "op_bob",
+            "amount": 10_000,
+            "ref": "ref-api-too-much",
+        },
+        headers=service_headers,
+    )
+    assert resp.status_code == 402
+    assert resp.json()["error"] == "insufficient_funds"
+
+    after = client.get("/tx", headers=service_headers).json()["transactions"]
+    assert after == before