PostgreSQL quickstart¶

Use PostgreSQL when proposals must survive process restarts or multiple workers need concurrency-correct lifecycle transitions.

The official postgresql/v1 store security profile requires host-protected private state and separate runtime and retention roles; it does not claim storage encryption or deletion from external copies.

python -m pip install "threvo-actions[postgres]==0.1.3"

The action schema can live beside the application's tables or in a dedicated PostgreSQL database. A dedicated database is the maximum-isolation option: use it when action credentials, backups, ownership, and incident blast radius must be separate from business data. The adapter never joins host tables, so only the migrator, runtime, and retention DSNs change. This separation does not make an action-store transaction atomic with the host application's transaction.

Apply migrations explicitly¶

The package never discovers credentials or migrates at import time. Put the DSN in an environment variable and name that variable on the command line:

export ACTIONS_MIGRATOR_DATABASE_URL='postgresql://migrator@localhost/actions'
threvo-actions postgres inspect \
  --dsn-env ACTIONS_MIGRATOR_DATABASE_URL --schema threvo_actions
threvo-actions postgres plan \
  --dsn-env ACTIONS_MIGRATOR_DATABASE_URL --schema threvo_actions

Both commands are read-only. inspect reports applied and pending versions; plan adds the exact rendered SQL and compatibility metadata for each pending migration. Confirm that the named environment variable points to the intended target and review the plan before allowing an operator or deployment workflow to mutate it. Then run:

threvo-actions postgres migrate \
  --dsn-env ACTIONS_MIGRATOR_DATABASE_URL --schema threvo_actions
threvo-actions postgres inspect \
  --dsn-env ACTIONS_MIGRATOR_DATABASE_URL --schema threvo_actions

Never put the DSN itself after --dsn-env; the argument is an environment variable name, so the DSN is not exposed in the threvo-actions process arguments. The example export still enters the literal DSN into shell history; load production credentials through your secret manager instead. migrate uses an advisory lock and applies packaged, forward-only migrations. The default lock wait is 30 seconds.

For a reviewed SQL artifact instead of a live package invocation, render a complete fresh-database script without credentials or a database driver:

threvo-actions postgres script --all --schema threvo_actions \
  > threvo-actions-bootstrap.sql

For an existing database, inspect first and pin the exact ledger version:

threvo-actions postgres script --from-version 3 \
  --schema threvo_actions --writers-quiesced \
  > threvo-actions-3-to-current.sql

Review and pin the generated file with the deployment release, then apply it once with the migrator credential while writers are actually stopped. The script validates that the database still has the declared migration prefix before applying DDL; it fails and rolls back if the target has drifted. Do not use the JSON emitted by postgres plan as an executable migration because it intentionally omits ledger operations and transaction guards. Configure the SQL client to propagate failures to the deployment runner; for psql, include --set ON_ERROR_STOP=1 when applying the file.

For an existing schema, the current lifecycle contract migrations require runtime and retention writers to be drained. When the plan reports that requirement, stop both writer lanes and rerun postgres migrate with --writers-quiesced. The flag is an explicit acknowledgement of that external deployment step; the command does not stop workers itself. A fresh database bootstrap does not require the flag.

Lifecycle migrations replace the database status constraint and transition trigger from the same closed Python contract used by the runtime. Before an upgrade, remove or explicitly remediate rows containing retired states; the migration transaction refuses them and rolls back without advancing migration history. Do not edit an applied migration checksum or map an unknown state to success. After remediation, run the same forward migration again and verify that inspect reports no pending versions.

The migrator DSN owns the tables by design. The second inspection is expected to warn about that ownership; never reuse this environment variable in an application process. Configure and inspect the runtime and retention DSNs separately as shown below.

Run the complete PostgreSQL example¶

The repository includes a full prepare → authority → execute → verify lifecycle using PostgresActionStore:

export DATABASE_URL='postgresql://localhost/actions'
uv run --extra postgres python -m examples.docs.postgres_runtime

Expected output:

verified
stored revision: 5
['proposal', 'authority', 'execution', 'execution', 'verification']

The program applies packaged migrations explicitly and uses one database role so it is easy to run locally:

Show the complete runnable example

"""A complete PostgreSQL-backed refund lifecycle.

Run against an empty local database with:

    DATABASE_URL=postgresql://localhost/actions \
      uv run --extra postgres python -m examples.docs.postgres_runtime

This example uses one database role for brevity. Production deployments should
separate migration, runtime, and retention credentials as documented.
"""

from __future__ import annotations

import asyncio
import os
import uuid
from datetime import UTC, datetime, timedelta

import asyncpg

from examples.docs.quickstart import (
    AGENT,
    CONSUMER,
    MANAGER,
    REQUESTER,
    TENANT,
    RefundCommand,
    build_demo,
)
from threvo_actions import (
    ActionRuntime,
    AuthorityDecision,
    AuthorityEvidence,
    OperationOutcome,
    ReadContext,
)
from threvo_actions.migrations import migrate_postgres
from threvo_actions.stores.postgres import PostgresActionStore, PostgresRetentionStore


class Identifiers:
    def new(self, prefix: str) -> str:
        return f"{prefix}:{uuid.uuid4()}"


class SystemClock:
    def now(self) -> datetime:
        return datetime.now(UTC)


async def main() -> None:
    dsn = os.environ.get("DATABASE_URL")
    if dsn is None:
        raise RuntimeError("set DATABASE_URL to a PostgreSQL DSN")

    pool = await asyncpg.create_pool(dsn, min_size=1, max_size=4)
    try:
        await migrate_postgres(pool, schema="threvo_actions")
        store = PostgresActionStore(pool, schema="threvo_actions")
        retention_store = PostgresRetentionStore(pool, schema="threvo_actions")
        clock = SystemClock()
        definition = build_demo().action
        runtime = ActionRuntime(
            store=store,
            retention_store=retention_store,
            clock=clock,
            identifiers=Identifiers(),
        )

        prepared = await runtime.prepare(
            definition,
            tenant_reference=TENANT,
            command=RefundCommand(order_reference="ORD-PG-42"),
            requesting_principal=REQUESTER,
            proposing_agent=AGENT,
        )
        record = await store.get(TENANT, prepared.proposal_reference)
        if record is None or record.commitment is None:
            raise RuntimeError("prepared proposal was not persisted")

        authorized = await runtime.record_authority(
            definition,
            evidence=AuthorityEvidence(
                tenant_reference=TENANT,
                action_type=definition.action_type,
                proposal_instance_reference=prepared.proposal_reference,
                semantic_effect_reference=record.semantic_effect_reference,
                authority=MANAGER,
                audience=(definition.authority_audience,),
                decision=AuthorityDecision.APPROVE,
                proposal_commitment=record.commitment.digest,
                channel_assurance=definition.authority_channel_assurance,
                issued_at=clock.now(),
                expires_at=clock.now() + timedelta(minutes=5),
            ),
            authenticated_authority=MANAGER,
        )
        if authorized.outcome is not OperationOutcome.AUTHORIZED:
            raise RuntimeError(f"authority failed: {authorized.outcome}")

        executed = await runtime.execute(
            definition,
            tenant_reference=TENANT,
            proposal_reference=prepared.proposal_reference,
        )
        if executed.outcome is not OperationOutcome.VERIFICATION_PENDING:
            raise RuntimeError(f"execution failed: {executed.outcome}")
        verified = await runtime.reconcile(
            definition,
            tenant_reference=TENANT,
            proposal_reference=prepared.proposal_reference,
        )
        if verified.outcome is not OperationOutcome.VERIFIED:
            raise RuntimeError(f"verification failed: {verified.outcome}")
        view = await runtime.read(
            definition,
            proposal_reference=prepared.proposal_reference,
            context=ReadContext(tenant_reference=TENANT, consumer=CONSUMER),
        )

        print(verified.outcome)
        print(f"stored revision: {view.revision}")
        print([receipt.receipt_type for receipt in view.receipts])
    finally:
        await pool.close()


if __name__ == "__main__":
    asyncio.run(main())

Create the stores in your application¶

import asyncpg

from threvo_actions import ActionRuntime
from threvo_actions.stores.postgres import (
    PostgresActionStore,
    PostgresRetentionStore,
)

runtime_pool = await asyncpg.create_pool(runtime_dsn)
retention_pool = await asyncpg.create_pool(retention_dsn)

runtime_store = PostgresActionStore(runtime_pool, schema="threvo_actions")
retention_store = PostgresRetentionStore(
    retention_pool,
    schema="threvo_actions",
)

runtime = ActionRuntime(
    store=runtime_store,
    retention_store=retention_store,
    clock=clock,
    identifiers=identifiers,
)

Here, clock implements Clock, identifiers implements IdentifierProvider, and the two DSNs come from your application's secret configuration. The complete program above supplies concrete versions.

The application creates and owns both pools. Use different database roles:

migrator owns schema changes but is never an application login;
runtime creates proposals and advances ordinary lifecycle state;
retention can run constrained erasure functions but cannot execute actions.

Generate the tested grant baseline without exposing a DSN:

threvo-actions postgres grants \
  --schema threvo_actions \
  --runtime-role actions_runtime \
  --retention-role actions_retention > actions-grants.sql

The command does not create roles or apply SQL. Review the file and apply it with the migrator after the schema is current.

After grants are applied, verify both application DSNs independently:

threvo-actions postgres inspect --dsn-env ACTIONS_RUNTIME_DATABASE_URL \
  --schema threvo_actions --require-separated-role
threvo-actions postgres inspect --dsn-env ACTIONS_RETENTION_DATABASE_URL \
  --schema threvo_actions --require-separated-role

Both commands must exit successfully before deployment. The migrator DSN owns the table by design and must not be used for either check or application process.

Gate application startup with the same pools or credentials:

threvo-actions postgres ready --dsn-env ACTIONS_RUNTIME_DATABASE_URL \
  --schema threvo_actions --lane runtime
threvo-actions postgres ready --dsn-env ACTIONS_RETENTION_DATABASE_URL \
  --schema threvo_actions --lane retention

The command is read-only and exits 3 for pending migrations, schema ownership, missing required privileges, or dangerous cross-lane privileges. Call check_postgres_readiness() directly when startup already owns an asyncpg pool.

What PostgreSQL guarantees¶

tenant-scoped proposal lookup;
compare-and-set revisions;
guarded lifecycle transitions;
atomic semantic-effect admission;
one verification lease at a time;
append-only active evidence;
constrained erasure through database-owned functions.

It does not provide distributed exactly-once effects. The executor still needs target-side idempotency and the verifier still needs an authoritative query.

Continue with the deployment and role guide for the complete grants, privacy boundary, and recovery assumptions.