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 symbolically unrolls the loop 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

The trusted effects expose one read effect (db_query, marked db_read) and one write effect (send_email). db_query's @deal.post caps the result length, which is what lets the prover bound the loop.

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

import deal
from tools.db.trusted.schemas import UserRow


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


@deal.has("db_read")
@deal.post(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 adds two domain helpers on top of the generic effect("send_email") and effect("db_query") relations: addresses_from(schema, field) 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 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 @deal.post 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

test: cases

cases: newsletter-pass count-pass no-emails-with-loop-pass literal-address-fail wrong-column-fail mixed-source-fail count-too-tight-fail

newsletter-pass:
    {{prove}} cases/newsletter_pass.py

count-pass:
    {{prove}} cases/count_pass.py

no-emails-with-loop-pass:
    {{prove}} cases/no_emails_with_loop_pass.py

literal-address-fail:
    if {{prove}} cases/literal_address_fail.py; then exit 1; fi

wrong-column-fail:
    if {{prove}} cases/wrong_column_fail.py; then exit 1; fi

mixed-source-fail:
    if {{prove}} cases/mixed_source_fail.py; then exit 1; fi

count-too-tight-fail:
    if {{prove}} cases/count_too_tight_fail.py; then exit 1; fi