Email-from-DB example

This example combines two trusted effects: db_query returns a bounded list of rows from a typed schema, and send_email sends to a given address. Agent code iterates over the query result and sends to each row's email field — the prover models the loop symbolically and checks that every recipient came from the right column.

It's the smallest example that exercises symbolic iteration over a trusted-call return value plus column-binding constraints.

Trusted module

Both functions declare their effect boundaries with @clauz3.trusted: db_query reads rows and send_email sends a message. db_query's @clauz3.ensures(...) caps the result length, which lets the prover bound the loop.

examples/email-from-db/tools/db/trusted/effects.py

from tools.db.trusted.schemas import UserRow

import clauz3


@clauz3.trusted
def send_email(addr: str, msg: str) -> None:
    """MOCK trusted email sender."""
    pass


@clauz3.trusted
@clauz3.ensures(lambda result: len(result) <= 100)
def db_query(table: str, where: dict[str, object] | None = None) -> list[UserRow]:
    """Trusted DB query returning at most 100 UserRows.

    ``table`` is the table name. ``where`` is an optional filter dict;
    only rows where all key=value constraints hold are returned.
    """
    return []

The schema is a clauz3.Row subclass — its fields become the row schema the prover binds against when iterating the query result.

examples/email-from-db/tools/db/trusted/schemas.py

import clauz3


class UserRow(clauz3.Row):
    name: str
    email: str
    consented: bool
    role: str

The contract module uses effect(send_email) and effect(db_query), passing the imported trusted functions. Its addresses_from(schema, field) helper binds every recipient to a column of a schema, and count_at_most(n) is a simple aggregate on the email relation.

examples/email-from-db/tools/db/trusted/contracts.py

"""Contract vocabulary for the email-from-db example.

Includes contracts for both the database query effects and the email effects.
"""

from tools.db.trusted.effects import db_query, send_email

from clauz3.spec import ContractSpec, contract, effect
from clauz3.spec import no_guarantees as core_no_guarantees

Email = effect(send_email)
DB = effect(db_query)


# ── Email contracts ───────────────────────────────────────────────────────────


@contract
def addresses_from(schema: type, field: str) -> ContractSpec:
    """Guarantee every email recipient came from `schema`.`field`."""
    column = getattr(schema, field)
    return Email.all(lambda e: e.addr == column)


@contract
def count_at_most(n: int) -> ContractSpec:
    """Guarantee at most n emails are sent."""
    return Email.count() <= n


@contract
def only(addresses: list[str]) -> ContractSpec:
    """Guarantee that every sent email targets one of these addresses."""
    return Email.all(lambda e: e.addr in addresses)


@contract
def none() -> ContractSpec:
    """Guarantee that no email is sent."""
    return Email.empty()


@contract
def no_guarantees() -> ContractSpec:
    """Make no guarantees about email effects."""
    return core_no_guarantees()


@contract
def unique_recipients() -> ContractSpec:
    """Guarantee that no reachable execution emails the same address twice."""
    return Email.distinct(lambda e: e.addr)


@contract
def at_most(count: int) -> ContractSpec:
    """Guarantee that at most count emails are sent."""
    return Email.count() <= count


# ── Database contracts ────────────────────────────────────────────────────────


@contract
def only_table(table: str) -> ContractSpec:
    """Guarantee every db_query reads from `table` only."""
    return DB.all(lambda e: e.table == table)


@contract
def only_where(filter_dict: dict[str, object]) -> ContractSpec:
    """Guarantee every db_query uses exactly this `where` filter."""
    return DB.all(lambda e: e.where == filter_dict)

A passing case

Looping over the consented users and emailing each one's email column discharges both addresses_from(UserRow, "email") and count_at_most(100) (the latter from db_query's @clauz3.ensures length cap).

examples/email-from-db/cases/newsletter_pass.py

# ruff: noqa
import clauz3
from tools.db.trusted import contracts
from tools.db.trusted.effects import db_query, send_email
from tools.db.trusted.schemas import UserRow


@clauz3.guarantee(contracts.addresses_from(UserRow, "email"))
@clauz3.guarantee(contracts.count_at_most(100))
def main() -> None:
    for row in db_query("users", where={"consented": True}):
        send_email(row.email, "Newsletter is out!")

A failing case

Asserting emails.none() and then sending inside a loop over db_query(...) is rejected: the loop can produce at least one reachable send_email fact.

examples/email-from-db/cases/email_loop_fail.py

# ruff: noqa
import clauz3
from tools.db.trusted import contracts
from tools.db.trusted.effects import db_query, send_email


@clauz3.guarantee(contracts.none())
def main() -> None:
    for row in db_query("users"):
        send_email(row.email, "hi")

All cases

Browse the full set on the all-cases page.

How they run

examples/email-from-db/Justfile

db_trusted := "tools/db/trusted"
prove := "uv run clauz3 prove --trusted-root " + db_trusted
# Proves each case in one process: *_pass.py must prove, *_fail.py must not.
prove-cases := "uv run python -m clauz3._prove_cases --trusted-root " + db_trusted

test: lint cases

lint:
    uv run clauz3 lint {{db_trusted}}/effects.py

cases:
    {{prove-cases}} cases/*_pass.py cases/*_fail.py