Build a Durable AI Workflow Engine
Build a PostgreSQL-backed AI workflow engine that recovers safely from crashes.
Introduction
30 Second Summary
A locked-account request can vanish when the service handling it stops halfway through. You are left starting over without knowing whether an earlier action already succeeded.
In this project, you will build a durable support-agent workflow with FastAPI. PostgreSQL will preserve every checkpoint through an interruption.
What You'll Build
You will watch an interrupted support request recover to completion while the final report proves that repeated execution produced one business effect.
By the end of this project, you'll have:
- A live AgentFlow API for submitting support requests. Its inspection endpoint shows each saved workflow state.
- A crash-recovery demonstration where a committed PostgreSQL checkpoint survives process loss.
- A reliability report proving full workflow completion with zero duplicate business effects.
- Secret Mission: Race two recovery workers through the acknowledgment gap while proving that repeated execution creates one business effect.
What do I need for this project?
Docker Desktop must be running before you begin. Every source file is supplied for this local project.
Before We Start
This opening checkpoint anchors the failure at the center of AgentFlow before the hands-on work begins. The project is a local teaching implementation of durable orchestration where an external effect can commit successfully before a worker crashes without recording its local acknowledgment.
Bootstrap the Transactional Stack
Durable workflow guarantees need a transactional database that survives worker restarts. This stage gives AgentFlow a real PostgreSQL foundation before any workflow logic depends on it.
You will use Docker Compose to run pinned PostgreSQL and Python images. A health-gated FastAPI service starts only after the database can accept connections.
In this step, get ready to:
- Create the supplied AgentFlow repository structure.
- Define the container runtime and transactional schema.
- Apply the initial migration and verify the API.
Create the repository and container stack
The repository pins every dependency that affects the experiment. The API uses Python 3.13.17. The database uses PostgreSQL 18.6.
- Start VS Code through your computer's app search.
- Create an `agentflow` folder on your Desktop.
- Open the `agentflow` folder as your VS Code workspace.
- Create an `app` folder inside `agentflow`.
- Create an `alembic` folder inside `agentflow`.
- Create a `versions` folder inside `alembic`.
- Create a `docs` folder inside `agentflow`.
- Create a `tests` folder inside `agentflow`.
- Create `requirements.txt` inside `agentflow` by pasting this dependency manifest:
fastapi==0.162.0
uvicorn==0.54.0
pydantic==2.14.1
SQLAlchemy==2.1.6
alembic==1.20.1
psycopg[binary]==3.3.6
pytest==9.1.1
httpx==0.28.1Why pin every dependency?
- Exact pins keep the framework and database-driver behavior reproducible.
- The test dependencies use the same package versions every time the API image is rebuilt.
- Save `requirements.txt`.
- Confirm that `requirements.txt` shows eight dependency lines in VS Code.
Missing a dependency line?
Compare the package names and version pins with the block above. Each dependency belongs on its own line.
Check my `requirements.txt` dependency pins.
- Create `Dockerfile` inside `agentflow` by pasting this container definition:
FROM python:3.13.17-slim-trixie
WORKDIR /app
COPY requirements.txt .
RUN python -m pip install -r requirements.txt
COPY . .
CMD ["uvicorn", "app.api:app", "--host", "0.0.0.0", "--port", "8000"]What does this container definition do?
- The base image supplies the pinned Python runtime inside the container.
- The separate dependency copy lets Docker reuse the installation layer when application files change.
- The final command serves `app.api:app` on port `8000`.
- Save `Dockerfile`.
- Confirm that the first line references `python:3.13.17-slim-trixie`.
Container definition looks different?
Check the capitalized filename first. Confirm that the image tag and `/app` working directory match the supplied definition.
Help me inspect my AgentFlow Dockerfile.
- Create `compose.yaml` inside `agentflow` by pasting this service configuration:
services:
db:
image: postgres:18.6
environment:
POSTGRES_USER: agentflow
POSTGRES_PASSWORD: agentflow
POSTGRES_DB: agentflow
healthcheck:
test: ["CMD-SHELL", "pg_isready -U $${POSTGRES_USER} -d $${POSTGRES_DB}"]
interval: 2s
timeout: 5s
retries: 10
volumes:
- db-data:/var/lib/postgresql
api:
build: .
environment:
DATABASE_URL: postgresql+psycopg://agentflow:agentflow@db:5432/agentflow
depends_on:
db:
condition: service_healthy
ports:
- "8000:8000"
volumes:
- .:/app
volumes:
db-data:How does health gating work?
- The `db` service stores PostgreSQL data in the named `db-data` volume.
- The health check asks PostgreSQL whether the `agentflow` database can accept connections.
- The `api` dependency waits for a healthy database before starting.
- The source bind mount keeps container code synchronized with the `agentflow` folder.
Why Compose for this stack?
A host-level Python installation could vary between computers. Compose keeps the Python runtime and PostgreSQL version inside reproducible containers.
A purpose-built workflow platform would hide the transaction boundaries you are here to inspect. This stack keeps the queue rows and checkpoints visible in PostgreSQL.
- Save `compose.yaml`.
- Confirm that the file defines `db`, `api`, and `db-data`.
Seeing YAML highlighting problems?
YAML depends on spaces. Check that `db` and `api` align beneath `services`.
Help me fix my Compose indentation.
- Create `README.md` inside `agentflow` by pasting this first section:
# AgentFlow
AgentFlow is a local durable workflow engine for a deterministic AI support agent. It demonstrates PostgreSQL checkpoints, leases, fencing tokens, action intents, acknowledgment-gap recovery, and idempotent business effects.
## Run
```bash
docker compose up -d db
docker compose run --build --rm api alembic upgrade head
docker compose up -d api
```
Open `http://localhost:8000/docs`.What does the run section capture?
The run section records the reproducible path from an empty local stack to a migrated API. It also gives the browser address used for the final check.
- Save `README.md`.
- Confirm that the file shows the project description and local run commands.
Markdown preview looks incomplete?
Check that both fenced code markers contain three backticks. The closing marker belongs directly after the three Compose commands.
Help me inspect the first half of my README.
- Append the remaining repository guidance to `README.md` by pasting this section after the documentation URL:
## Test and generate reliability evidence
```bash
docker compose run --rm api pytest -q
```
The crash-recovery test writes `artifacts/reliability-report.json`.
## Reliability claim
AgentFlow uses at-least-once tool execution. It demonstrates exactly one persisted business effect per stable idempotency key. A crash can cause the tool boundary to be invoked again, so it does not claim exactly-once execution.
## Documentation
- `docs/architecture.md`
- `docs/production-limitations.md`
- `artifacts/reliability-report.json`
## Cleanup
```bash
docker compose down
# Delete the PostgreSQL volume too:
docker compose down --volumes
```Why document the reliability claim now?
The README separates repeated execution from repeated business effects. Later tests turn that statement into measured evidence.
The cleanup section distinguishes stopping containers from deleting PostgreSQL data.
- Save `README.md`.
- Confirm that the file ends with the cleanup commands.
README sections out of order?
The test section belongs directly after the run section. Reliability, documentation, and cleanup follow it in that order.
Help me repair my complete README.
- Create an empty `app/__init__.py` file inside `app`.
- Create an empty `docs/architecture.md` file inside `docs`.
- Create an empty `docs/production-limitations.md` file inside `docs`.
- Create an empty `tests/conftest.py` file inside `tests`.
- Create an empty `tests/test_agentflow.py` file inside `tests`.
- Confirm that all five placeholder files appear in the VS Code file list.
Why create placeholders?
These paths establish the supplied repository layout without pulling later workflow logic into this stage. Later stages fill them with tests and design evidence.
Define the database schema
A durable checkpoint needs a schema before a worker can commit progress. SQLAlchemy Core defines that schema in Python. Alembic applies the matching revision to PostgreSQL.
- Create `alembic.ini` inside `agentflow` by pasting this first configuration section:
[alembic]
script_location = alembic
prepend_sys_path = .
sqlalchemy.url = postgresql+psycopg://agentflow:agentflow@db:5432/agentflow
[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic
[logger_root]
level = WARNING
handlers = console
qualname =
[logger_sqlalchemy]
level = WARNING
handlers =
qualname = sqlalchemy.engine
[logger_alembic]
level = INFO
handlers =
qualname = alembicWhat does this Alembic configuration control?
- The script location points Alembic at the `alembic` folder.
- The database URL reaches the `db` service through the Compose network.
- The logger sections keep migration output focused on schema activity.
- Save `alembic.ini`.
- Confirm that its first section points to `alembic` and the AgentFlow database URL.
Configuration sections not recognized?
Check that every section name keeps its surrounding brackets. Confirm that configuration properties use one equals sign.
Help me inspect my Alembic configuration.
- Append the console handler and formatter sections to `alembic.ini` by pasting this code:
[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic
[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%SHow is migration output formatted?
The console handler sends migration logs through a shared formatter. The format includes the severity and logger name before each message.
- Save `alembic.ini`.
- Confirm that the final line defines the time format.
Handler section missing?
Place the handler section after the Alembic logger section. Keep the formatter section at the end of the file.
Help me repair the final Alembic sections.
- Create `alembic/env.py` by pasting these imports and metadata settings:
from logging.config import fileConfig
from alembic import context
from sqlalchemy import engine_from_config, pool
from app.core import metadata
config = context.config
if config.config_file_name is not None:
fileConfig(config.config_file_name)
target_metadata = metadataWhat connects Alembic to the schema?
- The configuration object reads the settings from `alembic.ini`.
- The imported `metadata` object describes the tables that migrations manage.
- The file configuration activates the migration log format when a configuration file is present.
- Save `alembic/env.py`.
- Confirm that `target_metadata` references `metadata`.
Metadata import marked as unresolved?
The file path is correct before `app/core.py` exists. Finish the schema chunks below before using the container to import it.
Help me check the Alembic metadata wiring.
- Add offline migration support to `alembic/env.py` by pasting this function below `target_metadata = metadata`:
def run_migrations_offline() -> None:
context.configure(
url=config.get_main_option("sqlalchemy.url"),
target_metadata=target_metadata,
literal_binds=True,
)
with context.begin_transaction():
context.run_migrations()What does offline mode prepare?
Offline mode configures Alembic from the database URL without creating an engine connection. Literal binds let generated migration statements carry concrete values.
- Save `alembic/env.py`.
- Confirm that `run_migrations_offline()` contains one transaction scope.
Function indentation looks uneven?
Keep the configuration call and transaction scope indented inside `run_migrations_offline()`. Keep the migration call inside the transaction scope.
Help me fix the offline migration function.
- Add online migration execution to `alembic/env.py` by pasting this final section:
def run_migrations_online() -> None:
connectable = engine_from_config(
config.get_section(config.config_ini_section, {}),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
with connectable.connect() as connection:
context.configure(connection=connection, target_metadata=target_metadata)
with context.begin_transaction():
context.run_migrations()
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()How does online mode apply changes?
- The engine reads its connection settings from Alembic configuration.
- A live connection gives the migration transaction access to PostgreSQL.
- The final branch selects the matching execution mode.
- Save `alembic/env.py`.
- Confirm that the file ends by calling `run_migrations_online()` in the `else` branch.
Online migration block fails to align?
Keep the connection scope inside the function. Keep the final mode check outside both functions.
Help me inspect the online migration function.
- Create `alembic/versions/0001_agentflow.py` by pasting the revision metadata and workflow table:
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
revision = "0001_agentflow"
down_revision = None
branch_labels = None
depends_on = None
def upgrade() -> None:
op.create_table(
"workflows",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column("request", sa.Text(), nullable=False),
sa.Column("state", sa.String(32), nullable=False),
sa.Column("classification", postgresql.JSONB(), nullable=True),
sa.Column("plan", postgresql.JSONB(), nullable=True),
sa.Column("result", postgresql.JSONB(), nullable=True),
sa.Column("lease_owner", sa.String(100), nullable=True),
sa.Column("lease_expires_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("fence_token", sa.Integer(), nullable=False),
sa.Column("recovery_started_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True),
)What can a workflow row remember?
- State and JSON fields preserve deterministic decisions between worker runs.
- Lease fields identify the current owner and its expiration time.
- The fencing token increases when ownership changes.
- Timestamps make creation, recovery, updates, and completion measurable.
- Save `alembic/versions/0001_agentflow.py`.
- Confirm that the revision value is `0001_agentflow`.
Workflow migration has a syntax problem?
Check that every `sa.Column` call ends with a comma. Confirm that the final parenthesis closes `op.create_table`.
Help me inspect the workflow table migration.
- Add the intent and tool-ledger tables inside `upgrade()` after the workflows table:
op.create_table(
"action_intents",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column("workflow_id", postgresql.UUID(as_uuid=True), nullable=False, unique=True),
sa.Column("idempotency_key", sa.String(200), nullable=False, unique=True),
sa.Column("status", sa.String(32), nullable=False),
sa.Column("tool_result", postgresql.JSONB(), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("acknowledged_at", sa.DateTime(timezone=True), nullable=True),
)
op.create_table(
"tool_ledger",
sa.Column("idempotency_key", sa.String(200), primary_key=True),
sa.Column("result", postgresql.JSONB(), nullable=False),
sa.Column("attempts", sa.Integer(), nullable=False),
sa.Column("committed_at", sa.DateTime(timezone=True), nullable=False),
)Why separate intents from effects?
An action intent records what AgentFlow plans to invoke. The tool ledger records the independently committed business effect.
The stable idempotency key links repeated attempts to one effect record. The attempts counter preserves evidence that execution can repeat.
- Save `alembic/versions/0001_agentflow.py`.
- Confirm that `action_intents` and `tool_ledger` both contain `idempotency_key` columns.
Tables appear outside the upgrade function?
Keep both `op.create_table` calls indented by four spaces. They belong inside `upgrade()` after the workflows table call.
Help me fix the intent and ledger migration blocks.
- Add the downgrade path to `0001_agentflow.py` by pasting this function after `upgrade()`:
def downgrade() -> None:
op.drop_table("tool_ledger")
op.drop_table("action_intents")
op.drop_table("workflows")Why reverse the table order?
The downgrade removes the latest durability boundary first. This ordering also leaves room for future relationships between intents and workflows.
- Save `alembic/versions/0001_agentflow.py`.
- Confirm that `downgrade()` removes all three tables.
Downgrade function nested by mistake?
Align `def downgrade()` with `def upgrade()`. Keep each drop call indented inside the downgrade function.
Help me inspect the downgrade function.
- Create `app/core.py` by pasting these standard-library imports:
from __future__ import annotations
import json
import logging
import os
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any
from uuid import UUID, uuid4What do these imports prepare?
These modules support typed claims, UTC timestamps, structured logs, generated identifiers, future reports, and the later concurrency harness.
- Save `app/core.py`.
- Confirm that the first line enables postponed annotation evaluation.
Import names underlined?
Standard-library imports should resolve inside the pinned Python container. Check each name against the supplied block before continuing.
Help me check the standard-library imports.
- Add the SQLAlchemy imports to `app/core.py` by pasting this group below the standard-library imports:
from sqlalchemy import (
Column,
DateTime,
Integer,
MetaData,
String,
Table,
Text,
and_,
create_engine,
delete,
func,
insert,
or_,
select,
update,
)
from sqlalchemy.dialects.postgresql import JSONB, UUID as PG_UUID, insert as pg_insert
from sqlalchemy.orm import sessionmakerWhy use SQLAlchemy Core?
Core keeps the queue rows and guarded updates explicit. That visibility matters when later stages inspect row locks and ownership predicates.
The PostgreSQL-specific imports provide JSONB, UUID, and upsert support.
- Save `app/core.py`.
- Confirm that the import group includes `MetaData`, `Table`, and `sessionmaker`.
SQLAlchemy imports not resolved locally?
VS Code may be using your host interpreter while dependencies belong to the container. The build verification below confirms the imports in the actual runtime.
Help me inspect the SQLAlchemy import group.
- Add the database URL and workflows metadata to `app/core.py` by pasting this section below the imports:
DATABASE_URL = os.getenv(
"DATABASE_URL",
"postgresql+psycopg://agentflow:agentflow@db:5432/agentflow",
)
metadata = MetaData()
workflows = Table(
"workflows",
metadata,
Column("id", PG_UUID(as_uuid=True), primary_key=True),
Column("request", Text, nullable=False),
Column("state", String(32), nullable=False),
Column("classification", JSONB, nullable=True),
Column("plan", JSONB, nullable=True),
Column("result", JSONB, nullable=True),
Column("lease_owner", String(100), nullable=True),
Column("lease_expires_at", DateTime(timezone=True), nullable=True),
Column("fence_token", Integer, nullable=False),
Column("recovery_started_at", DateTime(timezone=True), nullable=True),
Column("created_at", DateTime(timezone=True), nullable=False),
Column("updated_at", DateTime(timezone=True), nullable=False),
Column("completed_at", DateTime(timezone=True), nullable=True),
)How does this metadata match the migration?
The table name and columns mirror the first migration. Alembic imports this shared `metadata` object through `alembic/env.py`.
The environment variable lets Compose provide the container database address.
- Save `app/core.py`.
- Confirm that `workflows` contains the state, lease, fence, recovery, and timestamp columns.
Metadata differs from the migration?
Compare the column names one at a time. A mismatch would make the Python schema describe a different database shape.
Help me compare the workflows schema.
- Add the action-intent and tool-ledger metadata to `app/core.py` by pasting this section:
action_intents = Table(
"action_intents",
metadata,
Column("id", PG_UUID(as_uuid=True), primary_key=True),
Column("workflow_id", PG_UUID(as_uuid=True), nullable=False, unique=True),
Column("idempotency_key", String(200), nullable=False, unique=True),
Column("status", String(32), nullable=False),
Column("tool_result", JSONB, nullable=True),
Column("created_at", DateTime(timezone=True), nullable=False),
Column("acknowledged_at", DateTime(timezone=True), nullable=True),
)
tool_ledger = Table(
"tool_ledger",
metadata,
Column("idempotency_key", String(200), primary_key=True),
Column("result", JSONB, nullable=False),
Column("attempts", Integer, nullable=False),
Column("committed_at", DateTime(timezone=True), nullable=False),
)What do these records represent?
An `action_intents` row captures planned work and its acknowledgment state. A `tool_ledger` row captures the effect result and total execution attempts for one stable key.
- Save `app/core.py`.
- Confirm that both table definitions match their migration counterparts.
Unique constraints missing?
Check `workflow_id` and `idempotency_key` in `action_intents`. Both columns use `unique=True`.
Help me inspect the durability tables.
- Finish the bootstrap layer in `app/core.py` by pasting these engine and utility definitions:
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(engine, expire_on_commit=False)
logger = logging.getLogger("agentflow")
logging.basicConfig(level=logging.INFO, format="%(message)s")
class CrashInjected(RuntimeError):
pass
class LostLease(RuntimeError):
pass
@dataclass(frozen=True)
class Claim:
workflow_id: UUID
worker_id: str
fence_token: int
def utcnow() -> datetime:
return datetime.now(timezone.utc)
def log_event(event: str, **fields: Any) -> None:
logger.info(json.dumps({"event": event, **fields}, default=str, sort_keys=True))What does the bootstrap layer expose?
- `engine` owns the database connection pool.
- `SessionLocal` creates transaction-capable SQLAlchemy sessions without expiring committed values.
- `Claim` carries the workflow owner and fencing token together.
- `log_event()` emits sortable JSON records for later reliability evidence.
- Save `app/core.py`.
- Confirm that the file ends with the `log_event()` definition.
Core module fails to parse?
Check the spacing between top-level classes and functions. Confirm that the `Claim` fields remain indented inside the dataclass.
Help me debug the bootstrap core module.
- Create `app/api.py` by pasting this health-check API:
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI(title="AgentFlow", version="1.0.0")
class WorkflowRequest(BaseModel):
request: str
@app.get("/health")
def health() -> dict[str, str]:
return {"status": "ok"}What does the first API prove?
The application metadata gives the generated documentation its AgentFlow title. The health route proves that the pinned FastAPI service can import and serve requests.
`WorkflowRequest` establishes the request shape that the next stage uses when durable workflow creation arrives.
- Save `app/api.py`.
- Confirm that the file defines `app`, `WorkflowRequest`, and `health()`.
API module shows syntax errors?
Check the route decorator and the dictionary returned by `health()`. Confirm that the class body contains one indented field.
Help me inspect the bootstrap API.
✔️ Awesome, I've got everything!
Your repository now contains the container configuration, schema metadata, initial migration, health API, README, and placeholder files. Save every open file before starting the stack.
ⓧ I'd like to double check the full code
Compare each populated file with these cumulative bootstrap versions. Keep `app/__init__.py`, `docs/architecture.md`, `docs/production-limitations.md`, `tests/conftest.py`, and `tests/test_agentflow.py` empty.
fastapi==0.162.0
uvicorn==0.54.0
pydantic==2.14.1
SQLAlchemy==2.1.6
alembic==1.20.1
psycopg[binary]==3.3.6
pytest==9.1.1
httpx==0.28.1FROM python:3.13.17-slim-trixie
WORKDIR /app
COPY requirements.txt .
RUN python -m pip install -r requirements.txt
COPY . .
CMD ["uvicorn", "app.api:app", "--host", "0.0.0.0", "--port", "8000"]services:
db:
image: postgres:18.6
environment:
POSTGRES_USER: agentflow
POSTGRES_PASSWORD: agentflow
POSTGRES_DB: agentflow
healthcheck:
test: ["CMD-SHELL", "pg_isready -U $${POSTGRES_USER} -d $${POSTGRES_DB}"]
interval: 2s
timeout: 5s
retries: 10
volumes:
- db-data:/var/lib/postgresql
api:
build: .
environment:
DATABASE_URL: postgresql+psycopg://agentflow:agentflow@db:5432/agentflow
depends_on:
db:
condition: service_healthy
ports:
- "8000:8000"
volumes:
- .:/app
volumes:
db-data:[alembic]
script_location = alembic
prepend_sys_path = .
sqlalchemy.url = postgresql+psycopg://agentflow:agentflow@db:5432/agentflow
[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic
[logger_root]
level = WARNING
handlers = console
qualname =
[logger_sqlalchemy]
level = WARNING
handlers =
qualname = sqlalchemy.engine
[logger_alembic]
level = INFO
handlers =
qualname = alembic
[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic
[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%Sfrom logging.config import fileConfig
from alembic import context
from sqlalchemy import engine_from_config, pool
from app.core import metadata
config = context.config
if config.config_file_name is not None:
fileConfig(config.config_file_name)
target_metadata = metadata
def run_migrations_offline() -> None:
context.configure(
url=config.get_main_option("sqlalchemy.url"),
target_metadata=target_metadata,
literal_binds=True,
)
with context.begin_transaction():
context.run_migrations()
def run_migrations_online() -> None:
connectable = engine_from_config(
config.get_section(config.config_ini_section, {}),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
with connectable.connect() as connection:
context.configure(connection=connection, target_metadata=target_metadata)
with context.begin_transaction():
context.run_migrations()
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
revision = "0001_agentflow"
down_revision = None
branch_labels = None
depends_on = None
def upgrade() -> None:
op.create_table(
"workflows",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column("request", sa.Text(), nullable=False),
sa.Column("state", sa.String(32), nullable=False),
sa.Column("classification", postgresql.JSONB(), nullable=True),
sa.Column("plan", postgresql.JSONB(), nullable=True),
sa.Column("result", postgresql.JSONB(), nullable=True),
sa.Column("lease_owner", sa.String(100), nullable=True),
sa.Column("lease_expires_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("fence_token", sa.Integer(), nullable=False),
sa.Column("recovery_started_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True),
)
op.create_table(
"action_intents",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column("workflow_id", postgresql.UUID(as_uuid=True), nullable=False, unique=True),
sa.Column("idempotency_key", sa.String(200), nullable=False, unique=True),
sa.Column("status", sa.String(32), nullable=False),
sa.Column("tool_result", postgresql.JSONB(), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("acknowledged_at", sa.DateTime(timezone=True), nullable=True),
)
op.create_table(
"tool_ledger",
sa.Column("idempotency_key", sa.String(200), primary_key=True),
sa.Column("result", postgresql.JSONB(), nullable=False),
sa.Column("attempts", sa.Integer(), nullable=False),
sa.Column("committed_at", sa.DateTime(timezone=True), nullable=False),
)
def downgrade() -> None:
op.drop_table("tool_ledger")
op.drop_table("action_intents")
op.drop_table("workflows")from __future__ import annotations
import json
import logging
import os
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any
from uuid import UUID, uuid4
from sqlalchemy import (
Column,
DateTime,
Integer,
MetaData,
String,
Table,
Text,
and_,
create_engine,
delete,
func,
insert,
or_,
select,
update,
)
from sqlalchemy.dialects.postgresql import JSONB, UUID as PG_UUID, insert as pg_insert
from sqlalchemy.orm import sessionmaker
DATABASE_URL = os.getenv(
"DATABASE_URL",
"postgresql+psycopg://agentflow:agentflow@db:5432/agentflow",
)
metadata = MetaData()
workflows = Table(
"workflows",
metadata,
Column("id", PG_UUID(as_uuid=True), primary_key=True),
Column("request", Text, nullable=False),
Column("state", String(32), nullable=False),
Column("classification", JSONB, nullable=True),
Column("plan", JSONB, nullable=True),
Column("result", JSONB, nullable=True),
Column("lease_owner", String(100), nullable=True),
Column("lease_expires_at", DateTime(timezone=True), nullable=True),
Column("fence_token", Integer, nullable=False),
Column("recovery_started_at", DateTime(timezone=True), nullable=True),
Column("created_at", DateTime(timezone=True), nullable=False),
Column("updated_at", DateTime(timezone=True), nullable=False),
Column("completed_at", DateTime(timezone=True), nullable=True),
)
action_intents = Table(
"action_intents",
metadata,
Column("id", PG_UUID(as_uuid=True), primary_key=True),
Column("workflow_id", PG_UUID(as_uuid=True), nullable=False, unique=True),
Column("idempotency_key", String(200), nullable=False, unique=True),
Column("status", String(32), nullable=False),
Column("tool_result", JSONB, nullable=True),
Column("created_at", DateTime(timezone=True), nullable=False),
Column("acknowledged_at", DateTime(timezone=True), nullable=True),
)
tool_ledger = Table(
"tool_ledger",
metadata,
Column("idempotency_key", String(200), primary_key=True),
Column("result", JSONB, nullable=False),
Column("attempts", Integer, nullable=False),
Column("committed_at", DateTime(timezone=True), nullable=False),
)
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(engine, expire_on_commit=False)
logger = logging.getLogger("agentflow")
logging.basicConfig(level=logging.INFO, format="%(message)s")
class CrashInjected(RuntimeError):
pass
class LostLease(RuntimeError):
pass
@dataclass(frozen=True)
class Claim:
workflow_id: UUID
worker_id: str
fence_token: int
def utcnow() -> datetime:
return datetime.now(timezone.utc)
def log_event(event: str, **fields: Any) -> None:
logger.info(json.dumps({"event": event, **fields}, default=str, sort_keys=True))from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI(title="AgentFlow", version="1.0.0")
class WorkflowRequest(BaseModel):
request: str
@app.get("/health")
def health() -> dict[str, str]:
return {"status": "ok"}# AgentFlow
AgentFlow is a local durable workflow engine for a deterministic AI support agent. It demonstrates PostgreSQL checkpoints, leases, fencing tokens, action intents, acknowledgment-gap recovery, and idempotent business effects.
## Run
```bash
docker compose up -d db
docker compose run --build --rm api alembic upgrade head
docker compose up -d api
```
Open `http://localhost:8000/docs`.
## Test and generate reliability evidence
```bash
docker compose run --rm api pytest -q
```
The crash-recovery test writes `artifacts/reliability-report.json`.
## Reliability claim
AgentFlow uses at-least-once tool execution. It demonstrates exactly one persisted business effect per stable idempotency key. A crash can cause the tool boundary to be invoked again, so it does not claim exactly-once execution.
## Documentation
- `docs/architecture.md`
- `docs/production-limitations.md`
- `artifacts/reliability-report.json`
## Cleanup
```bash
docker compose down
# Delete the PostgreSQL volume too:
docker compose down --volumes
```Start the stack and verify the API
The schema now exists as code. Applying revision `0001_agentflow` turns that definition into durable PostgreSQL tables.
- Open the integrated terminal in VS Code.
- Confirm that the terminal starts inside the `agentflow` folder.
The first image build can take a few minutes while Docker downloads the pinned images and installs dependencies. A quiet pause during that work is expected.
Before you run this, which service do you expect Compose to make healthy before Alembic connects?
- Build the API image and apply the initial migration by running:
docker compose run --build --rm api alembic upgrade headYou should see Alembic complete the upgrade to revision `0001_agentflow`. The one-off API container is removed after the command finishes.
What did this command establish?
Compose starts the database dependency and waits for its health check. Alembic then creates `workflows`, `action_intents`, and `tool_ledger` inside PostgreSQL.
Migration command did not complete?
- Confirm that Docker Desktop is still running.
- Check that `compose.yaml` uses spaces for indentation.
- Compare `app/core.py` with the full-code tab for a missing parenthesis or column.
Help me diagnose my AgentFlow migration failure.
- Start the API service in the background by running:
docker compose up -d apiYou should see Compose leave the `api` and `db` services running. The terminal returns to its prompt because detached mode keeps both containers in the background.
Why is the API safe to start now?
The database health check has passed. Revision `0001_agentflow` has also created the schema that future workflow operations require.
API service does not stay running?
Check the saved `Dockerfile` command and the `app/api.py` syntax. Confirm that port `8000` is available on your computer.
Help me troubleshoot the AgentFlow API container.
Before you check the browser, what project title do you expect the generated API documentation to show?
- Open `http://localhost:8000/docs` in your browser.
- Confirm that the page shows the AgentFlow API documentation.
- Confirm that the `GET /health` route appears in the route list.
That is the bootstrap complete. Your pinned API runtime now reaches a migrated PostgreSQL database through a health-gated local stack.
Documentation page does not load?
Confirm that the API startup command completed before refreshing the page. Check that the Compose port mapping remains `8000:8000`.
Help me restore the AgentFlow documentation page.
Your transactional foundation is live. Next, you will turn the empty workflow tables into committed state-machine checkpoints that survive interruption.
Commit Durable Workflow Checkpoints
Your transactional stack is running. However, the support workflow still needs durable progress that survives a worker interruption.
This step stores each decision in PostgreSQL as a separate checkpoint. You will deliberately stop a workflow after classification to expose the recovery problem.
In this step, get ready to:
- Create durable workflow records through the AgentFlow API.
- Commit deterministic decisions as guarded state transitions.
- Interrupt a workflow after classification to prove its checkpoint survives.
Persist workflow requests
A durable workflow stores its inputs before any worker starts processing them. This gives every later transition a stable workflow ID.
- In the Explorer sidebar in VS Code, select app/core.py.
- Find the end of log_event().
- Add the deterministic classification and planning functions below it:
def mock_llm_classify(request: str) -> dict[str, str]:
text = request.lower()
category = "account_access" if "password" in text or "login" in text else "general_support"
return {"category": category, "priority": "normal"}
def mock_llm_plan(classification: dict[str, str]) -> dict[str, list[str]]:
if classification["category"] == "account_access":
return {"steps": ["validate identity", "reset credentials", "confirm access"]}
return {"steps": ["inspect request", "apply documented fix", "confirm resolution"]}What Do These Functions Decide?
- The classifier maps password or login requests to account_access with normal priority.
- The planner converts that classification into a repeatable resolution sequence.
- Deterministic outputs let a recovery worker reproduce decisions without calling a live model.
- Save app/core.py.
- Restart the local services by running these commands in the VS Code terminal:
docker compose down
docker compose up -d apiWhat Does This Restart Do?
The first command removes the current containers and network while retaining the named volume. The second command starts the API after the database passes its health check.
- Refresh http://localhost:8000/docs.
You should see the AgentFlow API documentation again. This confirms that the updated module imports successfully inside the container.
Does the API Fail to Restart?
Check the new functions for matching indentation. Confirm that every dictionary closes with a brace.
Use Help me find the syntax or indentation problem in my new AgentFlow classification functions. if the container still stops.
Workflow creation needs one transaction to insert the initial row. Retrieval uses a fresh session so the returned state reflects committed database data.
- In app/core.py, add these persistence functions below mock_llm_plan():
def create_workflow(request: str) -> dict[str, Any]:
workflow_id = uuid4()
now = utcnow()
with SessionLocal.begin() as session:
session.execute(
insert(workflows).values(
id=workflow_id,
request=request,
state="PENDING",
fence_token=0,
created_at=now,
updated_at=now,
)
)
log_event("workflow_created", workflow_id=workflow_id)
return get_workflow(workflow_id) or {}
def get_workflow(workflow_id: UUID) -> dict[str, Any] | None:
with SessionLocal() as session:
row = session.execute(
select(workflows).where(workflows.c.id == workflow_id)
).mappings().one_or_none()
return dict(row) if row else NoneHow Does Creation Become Durable?
- The transaction inserts a unique workflow in the PENDING state.
- The initial fence token is 0 because no worker has claimed the workflow yet.
- The committed row is reloaded before the API returns it.
- Save app/core.py.
- Restart the local services by running these commands:
docker compose down
docker compose up -d apiWhat Does This Check Prove?
A clean API startup proves that Python can import the persistence functions. The PostgreSQL volume remains available across the restart.
- Refresh http://localhost:8000/docs.
You should see the AgentFlow API documentation load without an import failure.
The FastAPI layer now needs routes that call the persistence functions. The request route returns the committed row so you can immediately inspect its state.
- In the VS Code Explorer sidebar, select app/api.py.
- Replace its contents with the API code below:
from uuid import UUID
from fastapi import FastAPI
from pydantic import BaseModel
from app.core import create_workflow, get_workflow
app = FastAPI(title="AgentFlow", version="1.0.0")
class WorkflowRequest(BaseModel):
request: str
@app.get("/health")
def health() -> dict[str, str]:
return {"status": "ok"}
@app.post("/workflows")
def submit_workflow(payload: WorkflowRequest) -> dict:
return create_workflow(payload.request)
@app.get("/workflows/{workflow_id}")
def inspect_workflow(workflow_id: UUID) -> dict:
return get_workflow(workflow_id) or {"error": "not_found"}What Does the API Expose?
- The request model validates that each submitted payload contains a request string.
- The create route inserts a durable workflow through create_workflow().
- The inspection route reloads one workflow by its UUID.
- Save app/api.py.
- Restart the local services by running these commands:
docker compose down
docker compose up -d apiWhy Restart Again?
The API process imports its routes when the container starts. Restarting loads the new workflow endpoints into the running service.
- Refresh http://localhost:8000/docs.
- Confirm that the page lists POST /workflows.
- Confirm that the page lists GET /workflows/{workflow_id}.
Good progress. AgentFlow can now accept a support request and expose the committed workflow row through its API.
Are the Workflow Routes Missing?
Confirm that app/api.py imports both persistence functions. Check that both route decorators begin at the left edge of the file.
Use Help me diagnose why my FastAPI workflow routes do not appear in the API documentation. for a route-level check.
✔️ Awesome, I've got everything!
Your API exposes durable workflow creation and inspection. Make sure app/api.py is saved.
ⓧ I'd like to double check the full code
Compare your current app/api.py with this complete reference.
from uuid import UUID
from fastapi import FastAPI
from pydantic import BaseModel
from app.core import create_workflow, get_workflow
app = FastAPI(title="AgentFlow", version="1.0.0")
class WorkflowRequest(BaseModel):
request: str
@app.get("/health")
def health() -> dict[str, str]:
return {"status": "ok"}
@app.post("/workflows")
def submit_workflow(payload: WorkflowRequest) -> dict:
return create_workflow(payload.request)
@app.get("/workflows/{workflow_id}")
def inspect_workflow(workflow_id: UUID) -> dict:
return get_workflow(workflow_id) or {"error": "not_found"}This reference includes the existing health route plus the two durable workflow routes.
Guard each checkpoint
A checkpoint must belong to an active worker claim. The claim assigns an owner with a lease expiry and a monotonic fence token.
Why PostgreSQL Row Claims?
A row claim keeps the queue position and ownership metadata inside one transactional model. SQLAlchemy expresses the row lock through with_for_update(skip_locked=True).
The next stage pressure-tests this mechanism with expired leases and competing workers. This stage uses one interrupted owner to expose the stranded checkpoint.
- In app/core.py, add claim_workflow() below get_workflow():
def claim_workflow(worker_id: str, lease_seconds: int = 30) -> Claim | None:
now = utcnow()
with SessionLocal.begin() as session:
row = session.execute(
select(workflows)
.where(
workflows.c.state != "COMPLETED",
or_(
workflows.c.lease_owner.is_(None),
workflows.c.lease_expires_at < now,
),
)
.order_by(workflows.c.created_at)
.limit(1)
.with_for_update(skip_locked=True)
).mappings().one_or_none()
if row is None:
return None
token = row["fence_token"] + 1
recovered = row["lease_owner"] is not None
session.execute(
update(workflows)
.where(workflows.c.id == row["id"])
.values(
lease_owner=worker_id,
lease_expires_at=now + timedelta(seconds=lease_seconds),
fence_token=token,
recovery_started_at=(row["recovery_started_at"] or now) if recovered else None,
updated_at=now,
)
)
log_event(
"workflow_claimed",
workflow_id=row["id"],
worker_id=worker_id,
fence_token=token,
recovered=recovered,
)
return Claim(row["id"], worker_id, token)How Does the Claim Work?
- The query selects the oldest unfinished workflow whose lease is absent or expired.
- The row lock prevents another transaction from claiming the same row at that moment.
- The transaction records the worker ID with an expiry timestamp.
- The incremented fence token gives the claim a new authority number.
- Save app/core.py.
- Restart the local services by running these commands:
docker compose down
docker compose up -d apiWhat Does a Clean Startup Confirm?
The API can import the row-claim query and its lease calculations. The service remains available while its durable database volume survives the restart.
- Refresh http://localhost:8000/docs.
You should see the AgentFlow API documentation load with the workflow routes still listed.
Ownership alone does not protect later writes. Every checkpoint update also matches the worker ID with the current fence token and an unexpired lease.
- In app/core.py, add the lease guard and checkpoint function below claim_workflow():
def lease_guard(claim: Claim):
return and_(
workflows.c.id == claim.workflow_id,
workflows.c.lease_owner == claim.worker_id,
workflows.c.fence_token == claim.fence_token,
workflows.c.lease_expires_at > utcnow(),
)
def checkpoint(claim: Claim, expected_state: str, next_state: str, **values: Any) -> None:
with SessionLocal.begin() as session:
result = session.execute(
update(workflows)
.where(lease_guard(claim), workflows.c.state == expected_state)
.values(state=next_state, updated_at=utcnow(), **values)
)
if result.rowcount != 1:
raise LostLease(f"stale fence {claim.fence_token}")
log_event(
"checkpoint_committed",
workflow_id=claim.workflow_id,
worker_id=claim.worker_id,
state=next_state,
fence_token=claim.fence_token,
)What Makes the Update Guarded?
- The guard matches the workflow ID with the current worker.
- The guard requires the stored token to equal the claim token.
- The guard rejects a lease after its expiry time.
- The expected-state condition prevents a transition from being applied twice.
- Save app/core.py.
One transition per call creates the interruption boundary. A worker can classify a request and commit before planning begins.
- In app/core.py, add advance_one_step() below checkpoint().
- Add reset_database() at the bottom of the file for isolated integration tests:
def advance_one_step(claim: Claim, crash_after_tool_commit: bool = False) -> str:
row = get_workflow(claim.workflow_id)
if row is None:
raise KeyError(claim.workflow_id)
state = row["state"]
if state == "PENDING":
checkpoint(claim, state, "CLASSIFIED", classification=mock_llm_classify(row["request"]))
elif state == "CLASSIFIED":
checkpoint(claim, state, "PLANNED", plan=mock_llm_plan(row["classification"]))
return (get_workflow(claim.workflow_id) or {})["state"]
def reset_database() -> None:
with SessionLocal.begin() as session:
session.execute(delete(action_intents))
session.execute(delete(tool_ledger))
session.execute(delete(workflows))How Does One-Step Advancement Help?
- A PENDING workflow commits its classification as CLASSIFIED.
- A CLASSIFIED workflow commits its plan as PLANNED.
- Each call reloads the row after its transaction commits.
- The reset helper clears test data in dependency order.
- Save app/core.py.
- Restart the local services by running these commands:
docker compose down
docker compose up -d apiWhat Does This Restart Validate?
A successful restart validates the completed checkpoint module at import time. The next test exercises its database behavior.
- Refresh http://localhost:8000/docs.
You should see the AgentFlow API documentation with its health and workflow routes available.
Does the Checkpoint Module Fail to Load?
Confirm that lease_guard() appears before checkpoint(). Confirm that checkpoint() appears before advance_one_step().
Use Help me trace an import or syntax failure in my AgentFlow checkpoint functions. if the module still fails.
✔️ Awesome, I've got everything!
Your checkpoint engine now creates workflows and commits one guarded transition per call. Make sure app/core.py is saved.
ⓧ I'd like to double check the full code
Compare your current app/core.py with this cumulative reference.
from __future__ import annotations
import json
import logging
import os
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any
from uuid import UUID, uuid4
from sqlalchemy import (
Column,
DateTime,
Integer,
MetaData,
String,
Table,
Text,
and_,
create_engine,
delete,
func,
insert,
or_,
select,
update,
)
from sqlalchemy.dialects.postgresql import JSONB, UUID as PG_UUID, insert as pg_insert
from sqlalchemy.orm import sessionmaker
DATABASE_URL = os.getenv(
"DATABASE_URL",
"postgresql+psycopg://agentflow:agentflow@db:5432/agentflow",
)
metadata = MetaData()
workflows = Table(
"workflows",
metadata,
Column("id", PG_UUID(as_uuid=True), primary_key=True),
Column("request", Text, nullable=False),
Column("state", String(32), nullable=False),
Column("classification", JSONB, nullable=True),
Column("plan", JSONB, nullable=True),
Column("result", JSONB, nullable=True),
Column("lease_owner", String(100), nullable=True),
Column("lease_expires_at", DateTime(timezone=True), nullable=True),
Column("fence_token", Integer, nullable=False),
Column("recovery_started_at", DateTime(timezone=True), nullable=True),
Column("created_at", DateTime(timezone=True), nullable=False),
Column("updated_at", DateTime(timezone=True), nullable=False),
Column("completed_at", DateTime(timezone=True), nullable=True),
)
action_intents = Table(
"action_intents",
metadata,
Column("id", PG_UUID(as_uuid=True), primary_key=True),
Column("workflow_id", PG_UUID(as_uuid=True), nullable=False, unique=True),
Column("idempotency_key", String(200), nullable=False, unique=True),
Column("status", String(32), nullable=False),
Column("tool_result", JSONB, nullable=True),
Column("created_at", DateTime(timezone=True), nullable=False),
Column("acknowledged_at", DateTime(timezone=True), nullable=True),
)
tool_ledger = Table(
"tool_ledger",
metadata,
Column("idempotency_key", String(200), primary_key=True),
Column("result", JSONB, nullable=False),
Column("attempts", Integer, nullable=False),
Column("committed_at", DateTime(timezone=True), nullable=False),
)
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(engine, expire_on_commit=False)
logger = logging.getLogger("agentflow")
logging.basicConfig(level=logging.INFO, format="%(message)s")
class CrashInjected(RuntimeError):
pass
class LostLease(RuntimeError):
pass
@dataclass(frozen=True)
class Claim:
workflow_id: UUID
worker_id: str
fence_token: int
def utcnow() -> datetime:
return datetime.now(timezone.utc)
def log_event(event: str, **fields: Any) -> None:
logger.info(json.dumps({"event": event, **fields}, default=str, sort_keys=True))
def mock_llm_classify(request: str) -> dict[str, str]:
text = request.lower()
category = "account_access" if "password" in text or "login" in text else "general_support"
return {"category": category, "priority": "normal"}
def mock_llm_plan(classification: dict[str, str]) -> dict[str, list[str]]:
if classification["category"] == "account_access":
return {"steps": ["validate identity", "reset credentials", "confirm access"]}
return {"steps": ["inspect request", "apply documented fix", "confirm resolution"]}
def create_workflow(request: str) -> dict[str, Any]:
workflow_id = uuid4()
now = utcnow()
with SessionLocal.begin() as session:
session.execute(
insert(workflows).values(
id=workflow_id,
request=request,
state="PENDING",
fence_token=0,
created_at=now,
updated_at=now,
)
)
log_event("workflow_created", workflow_id=workflow_id)
return get_workflow(workflow_id) or {}
def get_workflow(workflow_id: UUID) -> dict[str, Any] | None:
with SessionLocal() as session:
row = session.execute(
select(workflows).where(workflows.c.id == workflow_id)
).mappings().one_or_none()
return dict(row) if row else None
def claim_workflow(worker_id: str, lease_seconds: int = 30) -> Claim | None:
now = utcnow()
with SessionLocal.begin() as session:
row = session.execute(
select(workflows)
.where(
workflows.c.state != "COMPLETED",
or_(
workflows.c.lease_owner.is_(None),
workflows.c.lease_expires_at < now,
),
)
.order_by(workflows.c.created_at)
.limit(1)
.with_for_update(skip_locked=True)
).mappings().one_or_none()
if row is None:
return None
token = row["fence_token"] + 1
recovered = row["lease_owner"] is not None
session.execute(
update(workflows)
.where(workflows.c.id == row["id"])
.values(
lease_owner=worker_id,
lease_expires_at=now + timedelta(seconds=lease_seconds),
fence_token=token,
recovery_started_at=(row["recovery_started_at"] or now) if recovered else None,
updated_at=now,
)
)
log_event(
"workflow_claimed",
workflow_id=row["id"],
worker_id=worker_id,
fence_token=token,
recovered=recovered,
)
return Claim(row["id"], worker_id, token)
def lease_guard(claim: Claim):
return and_(
workflows.c.id == claim.workflow_id,
workflows.c.lease_owner == claim.worker_id,
workflows.c.fence_token == claim.fence_token,
workflows.c.lease_expires_at > utcnow(),
)
def checkpoint(claim: Claim, expected_state: str, next_state: str, **values: Any) -> None:
with SessionLocal.begin() as session:
result = session.execute(
update(workflows)
.where(lease_guard(claim), workflows.c.state == expected_state)
.values(state=next_state, updated_at=utcnow(), **values)
)
if result.rowcount != 1:
raise LostLease(f"stale fence {claim.fence_token}")
log_event(
"checkpoint_committed",
workflow_id=claim.workflow_id,
worker_id=claim.worker_id,
state=next_state,
fence_token=claim.fence_token,
)
def advance_one_step(claim: Claim, crash_after_tool_commit: bool = False) -> str:
row = get_workflow(claim.workflow_id)
if row is None:
raise KeyError(claim.workflow_id)
state = row["state"]
if state == "PENDING":
checkpoint(claim, state, "CLASSIFIED", classification=mock_llm_classify(row["request"]))
elif state == "CLASSIFIED":
checkpoint(claim, state, "PLANNED", plan=mock_llm_plan(row["classification"]))
return (get_workflow(claim.workflow_id) or {})["state"]
def reset_database() -> None:
with SessionLocal.begin() as session:
session.execute(delete(action_intents))
session.execute(delete(tool_ledger))
session.execute(delete(workflows))This reference combines the existing metadata with workflow creation, claiming, guarded checkpoints, one-step advancement, and test cleanup.
Interrupt after one checkpoint
An in-memory workflow would lose its classification when its worker disappears. This test stops immediately after the first committed transition and reloads the row through a separate database session.
The test intentionally leaves the workflow stranded in CLASSIFIED. Recovery comes after you observe that durable shortfall.
- In the VS Code Explorer sidebar, select tests/conftest.py.
- Replace its contents with the database cleanup fixture below:
import pytest
from app.core import reset_database
@pytest.fixture(autouse=True)
def clean_database():
reset_database()
yield
reset_database()Why Clean the Database Around Every Test?
The autouse fixture removes rows before each test. Its teardown removes rows again after the test finishes.
Each integration test therefore starts with an empty workflow queue and produces independent evidence.
- Save tests/conftest.py.
✔️ Awesome, I've got everything!
Your automatic database cleanup fixture is saved in tests/conftest.py.
ⓧ I'd like to double check the full code
Compare your fixture with this complete reference.
import pytest
from app.core import reset_database
@pytest.fixture(autouse=True)
def clean_database():
reset_database()
yield
reset_database()This fixture resets all three AgentFlow tables around every test.
The Pytest integration test creates a password request and gives it to interrupted-worker. One call to advance_one_step() commits the classification before the test reloads the row.
- In the VS Code Explorer sidebar, select tests/test_agentflow.py.
- Replace its contents with the integration test scaffold below:
import pytest
from fastapi.testclient import TestClient
from app.api import app
from app.core import (
advance_one_step,
claim_workflow,
create_workflow,
get_workflow,
)
client = TestClient(app)
def test_normal_api_workflow_completes():
created = client.post("/workflows", json={"request": "I cannot login after a password reset"}).json()
response = client.post("/workers/api-worker/run-once").json()
finished = client.get(f"/workflows/{created['id']}").json()
assert response == {"worked": True}
assert finished["state"] == "COMPLETED"
assert finished["result"]["account_unlocked"] is True
def test_checkpoint_survives_interruption():
workflow = create_workflow("My password no longer works")
claim = claim_workflow("interrupted-worker")
assert claim is not None
assert advance_one_step(claim) == "CLASSIFIED"
persisted = get_workflow(workflow["id"])
assert persisted is not None
assert persisted["state"] == "CLASSIFIED"
assert persisted["classification"]["category"] == "account_access"What Does the Interruption Test Prove?
- The workflow begins as a committed PENDING row.
- The worker claim supplies the ownership information required by the checkpoint guard.
- One advancement commits the CLASSIFIED state with its classification payload.
- The final lookup proves another session can recover the stored decision.
- Save tests/test_agentflow.py.
✔️ Awesome, I've got everything!
Your interruption test is saved. The targeted run below executes only the checkpoint test.
ⓧ I'd like to double check the full code
Compare your current tests/test_agentflow.py with this reference.
import pytest
from fastapi.testclient import TestClient
from app.api import app
from app.core import (
advance_one_step,
claim_workflow,
create_workflow,
get_workflow,
)
client = TestClient(app)
def test_normal_api_workflow_completes():
created = client.post("/workflows", json={"request": "I cannot login after a password reset"}).json()
response = client.post("/workers/api-worker/run-once").json()
finished = client.get(f"/workflows/{created['id']}").json()
assert response == {"worked": True}
assert finished["state"] == "COMPLETED"
assert finished["result"]["account_unlocked"] is True
def test_checkpoint_survives_interruption():
workflow = create_workflow("My password no longer works")
claim = claim_workflow("interrupted-worker")
assert claim is not None
assert advance_one_step(claim) == "CLASSIFIED"
persisted = get_workflow(workflow["id"])
assert persisted is not None
assert persisted["state"] == "CLASSIFIED"
assert persisted["classification"]["category"] == "account_access"This cumulative test file includes the later normal-completion scaffold plus the runnable checkpoint test for this stage.
Before you run the targeted test, do you think the interrupted workflow will reach completion or remain at its last committed checkpoint?
- Run the interrupted-checkpoint test against PostgreSQL with this command:
docker compose run --rm api pytest -q tests/test_agentflow.py::test_checkpoint_survives_interruptionWhat Should You Observe?
You should see 1 passed. The passing assertion confirms that the reloaded workflow remains CLASSIFIED with an account_access classification.
The workflow is intentionally unfinished. Its durable checkpoint survives while no worker currently advances it.
That is the durability gap made visible. The decision survived process loss even though the workflow did not complete.
Does the Checkpoint Test Fail?
Confirm that the migration is still applied. Check that the database service is healthy before rerunning the targeted test.
Compare the PENDING and CLASSIFIED strings across the test with advance_one_step().
Use Help me debug why the AgentFlow checkpoint integration test is failing. for a focused diagnosis.
Your workflow now remembers its last committed decision after interruption. Next, you will let a new worker recover that stranded checkpoint while preventing the expired owner from writing again.
Coordinate Workers with Leases
Your durable checkpoint preserves a workflow after its worker disappears. That stranded row now needs a safe new owner before processing can continue.
In this step, competing workers use PostgreSQL row locks to claim different work. Each lease carries a fencing token that removes an expired worker's authority.
In this step, get ready to:
- Claim eligible workflows without blocking competing workers.
- Guard every checkpoint with a current lease and fencing token.
- Prove that two worker identities can complete one shared queue.
Claim work with leases and fencing tokens
A row lock makes the claim atomic inside its transaction. The persisted lease keeps ownership visible after that transaction releases its lock.
Why use queue row locks?
PostgreSQL documents SKIP LOCKED for queue-like tables with multiple consumers. A worker skips rows another transaction already holds, so competitors can claim different workflows.
This keeps the queue row, lease, token, and checkpoint inside one inspectable transactional model. Session advisory locks would separate ownership from the workflow record.
- Switch back to app/core.py in the VS Code editor from the previous step.
- Extend the import section with the concurrency, data class, and lease-time helpers below:
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from datetime import datetime, timedelta, timezoneWhat do these imports provide?
- ThreadPoolExecutor gives the integration test independent thread-backed worker identities.
- dataclass creates an immutable value object for each successful claim.
- timedelta converts a lease duration into an expiration timestamp.
- Add and_ and or_ to the existing SQLAlchemy import group.
- Find logging.basicConfig in app/core.py.
- Add the lease error and immutable claim model below that logging configuration:
class LostLease(RuntimeError):
pass
@dataclass(frozen=True)
class Claim:
workflow_id: UUID
worker_id: str
fence_token: intWhat does a claim represent?
- Claim carries the workflow identifier assigned to a worker.
- worker_id identifies the current lease owner.
- fence_token records that owner's generation number.
- LostLease stops an expired or replaced owner from committing another checkpoint.
Claiming needs one uninterrupted transaction. The first half locks the oldest eligible row while skipping work already locked by another worker.
- Find get_workflow() in app/core.py.
- Paste the first half of claim_workflow() below get_workflow().
- Keep your cursor inside the function for the second half:
def claim_workflow(worker_id: str, lease_seconds: int = 30) -> Claim | None:
now = utcnow()
with SessionLocal.begin() as session:
row = session.execute(
select(workflows)
.where(
workflows.c.state != "COMPLETED",
or_(
workflows.c.lease_owner.is_(None),
workflows.c.lease_expires_at < now,
),
)
.order_by(workflows.c.created_at)
.limit(1)
.with_for_update(skip_locked=True)
).mappings().one_or_none()
if row is None:
return None
token = row["fence_token"] + 1
recovered = row["lease_owner"] is not NoneHow does the claim query work?
- state != "COMPLETED" removes finished workflows from the queue.
- or_(...) accepts unowned rows or rows whose lease has expired.
- with_for_update(skip_locked=True) locks one eligible row without waiting for a competitor's row.
- token increments the prior generation so an older claim becomes stale.
- Complete claim_workflow() by pasting this second half directly below recovered = row["lease_owner"] is not None:
session.execute(
update(workflows)
.where(workflows.c.id == row["id"])
.values(
lease_owner=worker_id,
lease_expires_at=now + timedelta(seconds=lease_seconds),
fence_token=token,
recovery_started_at=(row["recovery_started_at"] or now) if recovered else None,
updated_at=now,
)
)
log_event(
"workflow_claimed",
workflow_id=row["id"],
worker_id=worker_id,
fence_token=token,
recovered=recovered,
)
return Claim(row["id"], worker_id, token)What gets committed with the claim?
- lease_owner stores the active worker identity.
- lease_expires_at defines when another worker may recover the row.
- fence_token persists the new ownership generation.
- recovery_started_at records when an abandoned workflow first entered recovery.
- Add lease_guard() directly below claim_workflow():
def lease_guard(claim: Claim):
return and_(
workflows.c.id == claim.workflow_id,
workflows.c.lease_owner == claim.worker_id,
workflows.c.fence_token == claim.fence_token,
workflows.c.lease_expires_at > utcnow(),
)Why does the guard check four values?
The workflow ID targets one row. The owner, token, and expiration checks prove that the caller still holds its current authority.
A worker fails this guard after its lease expires. It also fails after a recovery worker claims the row with a higher token.
- Save app/core.py.
- Open the VS Code Problems panel to confirm that the completed function has no Python syntax errors.
Seeing syntax or indentation problems?
- Check that the second claim block begins inside the existing with SessionLocal.begin() as session: scope.
- Check that return Claim(...) aligns with the function's with statement.
- Confirm that and_ and or_ appear in the SQLAlchemy import group.
Help me diagnose my worker-claim syntax or indentation issue.
Fence checkpoints and run competing workers
A successful claim only establishes ownership. Every later state update must prove that the same owner still holds the same unexpired generation.
- Find the existing checkpoint() function in app/core.py.
- Replace that function with the guarded version below:
def checkpoint(claim: Claim, expected_state: str, next_state: str, **values: Any) -> None:
with SessionLocal.begin() as session:
result = session.execute(
update(workflows)
.where(lease_guard(claim), workflows.c.state == expected_state)
.values(state=next_state, updated_at=utcnow(), **values)
)
if result.rowcount != 1:
raise LostLease(f"stale fence {claim.fence_token}")
log_event(
"checkpoint_committed",
workflow_id=claim.workflow_id,
worker_id=claim.worker_id,
state=next_state,
fence_token=claim.fence_token,
)How does checkpoint fencing work?
- lease_guard(claim) embeds ownership checks in the state update itself.
- expected_state prevents two transitions from consuming the same checkpoint.
- rowcount must equal one for the worker to retain authority.
- LostLease turns a stale update into an explicit failure.
Each call to run_once() creates its own session through the lower-level claim and checkpoint helpers. This keeps worker transactions independent even when threads run concurrently.
- Find advance_one_step() in app/core.py.
- Add run_once() below the state-transition functions:
def run_once(worker_id: str, crash_after_tool_commit: bool = False) -> bool:
claim = claim_workflow(worker_id)
if claim is None:
return False
for _ in range(8):
state = advance_one_step(claim, crash_after_tool_commit=crash_after_tool_commit)
if state == "COMPLETED":
return True
raise RuntimeError("workflow exceeded transition limit")What does one worker run do?
The worker claims one workflow before advancing its checkpoints. It returns False when no eligible work remains.
The transition limit catches a state machine that loops without reaching COMPLETED.
- Add run_until_idle() directly below run_once():
def run_until_idle(worker_id: str) -> int:
completed = 0
while run_once(worker_id):
completed += 1
return completedWhat does the completion count prove?
Each worker counts only workflows that its own calls complete. Adding both workers' counts later proves that the shared queue did not lose work.
- Add expire_lease() directly below run_until_idle():
def expire_lease(workflow_id: UUID) -> None:
with SessionLocal.begin() as session:
session.execute(
update(workflows)
.where(workflows.c.id == workflow_id)
.values(lease_expires_at=utcnow() - timedelta(seconds=1))
)
log_event("lease_expired_for_test", workflow_id=workflow_id)Why expire a lease deterministically?
Moving the expiration one second into the past creates a repeatable recovery condition. Tests can trigger reclamation immediately without waiting for a real timeout.
- Add run_competing_workers() directly below expire_lease():
def run_competing_workers(*worker_ids: str) -> list[int]:
with ThreadPoolExecutor(max_workers=len(worker_ids)) as executor:
futures = [executor.submit(run_until_idle, worker_id) for worker_id in worker_ids]
return [future.result() for future in futures]How do the competitors stay independent?
Each submitted future calls run_until_idle() with a distinct worker identity. Their claim operations create separate SQLAlchemy transaction scopes.
ThreadPoolExecutor is the deterministic test harness. Production workers would run as separately supervised processes.
- Save app/core.py.
- Switch back to tests/test_agentflow.py.
- Add run_competing_workers to the existing import list from app.core.
- Add workflows to the same import list.
- Add the concurrent queue test below the checkpoint test:
def test_two_workers_complete_concurrent_queue():
for number in range(8):
create_workflow(f"Support request {number}")
completed_per_worker = run_competing_workers("worker-a", "worker-b")
with SessionLocal() as session:
states = session.execute(select(workflows.c.state)).scalars().all()
assert sum(completed_per_worker) == 8
assert states == ["COMPLETED"] * 8What does this test prove?
- The loop creates eight independent queue rows.
- worker-a and worker-b compete for those rows concurrently.
- The summed completion count proves that the workers processed all eight workflows.
- The state query proves that every durable row reached COMPLETED.
- Save tests/test_agentflow.py.
Before you run the test, do you expect both workers to block on the same first workflow or claim different eligible rows?
- Run the concurrent queue test from the VS Code terminal:
docker compose run --rm api pytest -q tests/test_agentflow.py::test_two_workers_complete_concurrent_queueWhat does this check run?
Docker Compose starts a one-off API container against the running PostgreSQL service. Pytest runs only the concurrent queue test.
You should see 1 passed. The two workers claimed different rows until all eight durable workflows reached COMPLETED.
Concurrent queue test not passing?
- Check that with_for_update(skip_locked=True) remains attached to the workflow-selection query.
- Check that lease_guard() compares the claim's worker ID and fencing token.
- Check that each helper uses SessionLocal instead of sharing one session between threads.
Help me diagnose why the two-worker queue test is failing.
Expose one worker run through the API
The coordination logic now works below the HTTP boundary. A worker endpoint lets you trigger one queue-processing run through the existing FastAPI service.
- Switch back to app/api.py.
- Replace the existing import from app.core with this updated line:
from app.core import create_workflow, get_workflow, run_onceWhat does this import add?
run_once connects the API layer to the worker coordination path. The request's worker ID becomes the lease owner stored in PostgreSQL.
- Add the worker route below inspect_workflow():
@app.post("/workers/{worker_id}/run-once")
def run_worker_once(worker_id: str) -> dict[str, bool]:
return {"worked": run_once(worker_id)}What does the endpoint return?
The endpoint returns worked as True when the worker completes a claimed workflow. It returns False when the queue has no eligible work.
- Save app/api.py.
- Restart the running API service so Uvicorn loads the new route:
docker compose restart apiWhat does the restart change?
Docker Compose restarts only the api service. PostgreSQL and its named data volume remain available.
Before you refresh the API documentation, which new route do you expect to find under the AgentFlow endpoints?
- Return to http://localhost:8000/docs in your browser.
- Refresh the page.
- Confirm that POST /workers/{worker_id}/run-once appears in the API documentation.
You should see the new worker endpoint beside the existing workflow routes. This proves that the HTTP layer loaded the coordination path.
Worker endpoint missing from the docs?
- Confirm that the route decorator begins with @app.post.
- Confirm that run_once appears in the import from app.core.
- Refresh the documentation after the API container finishes restarting.
Help me find why my worker route is missing from FastAPI's documentation.
✔️ Awesome, I've got everything!
Your API now exposes one worker run. Double check that you saved app/api.py before the final test.
ⓧ I'd like to double check the full code
Compare your complete app/api.py file with this version:
from uuid import UUID
from fastapi import FastAPI
from pydantic import BaseModel
from app.core import create_workflow, get_workflow, run_once
app = FastAPI(title="AgentFlow", version="1.0.0")
class WorkflowRequest(BaseModel):
request: str
@app.get("/health")
def health() -> dict[str, str]:
return {"status": "ok"}
@app.post("/workflows")
def submit_workflow(payload: WorkflowRequest) -> dict:
return create_workflow(payload.request)
@app.get("/workflows/{workflow_id}")
def inspect_workflow(workflow_id: UUID) -> dict:
return get_workflow(workflow_id) or {"error": "not_found"}
@app.post("/workers/{worker_id}/run-once")
def run_worker_once(worker_id: str) -> dict[str, bool]:
return {"worked": run_once(worker_id)}For the final check, do you expect worker competition to change after exposing the same run_once() function through FastAPI?
- Run the required concurrent queue verification again:
docker compose run --rm api pytest -q tests/test_agentflow.py::test_two_workers_complete_concurrent_queueWhat confirms success?
The targeted test exercises two competing worker identities against the PostgreSQL queue. A passing result confirms that the API edit preserved claim, lease, and fencing behavior.
You should see 1 passed. That is the hard part complete: expired work can gain a new owner while stale owners lose permission to advance it.
Your workers now coordinate safely across durable checkpoints. Next, you will protect the external business effect that a database lease cannot make atomic.
Reconcile Idempotent Tool Effects
Your leases now decide which worker may update a workflow. A remaining gap appears when an external effect commits before AgentFlow records its acknowledgment.
In this step, you will separate the action intent, the simulated tool effect, and the acknowledgment across deliberate PostgreSQL transaction boundaries. A stable idempotency key turns an uncertain retry into one durable business effect.
In this step, get ready to:
- Persist an action intent before calling the simulated tool.
- Deduplicate repeated tool attempts through a stable ledger key.
- Crash after the effect commits before recovering the workflow safely.
Persist the action intent
An action intent records what AgentFlow plans to do before it crosses the tool boundary. The intent shares a transaction with the transition to ACTION_PENDING.
Why use an action intent?
A tool call cannot join the same transaction as a remote business effect. The durable intent gives recovery workers a stable record of the unfinished action.
The key <workflow-id>:resolve-account remains unchanged across every retry. That identity is the foundation of safe reconciliation.
- In app/core.py, locate the PostgreSQL dialect import.
- Replace that import with the following line:
from sqlalchemy.dialects.postgresql import JSONB, UUID as PG_UUID, insert as pg_insertWhat does this import add?
The pg_insert alias exposes the PostgreSQL-specific insert builder from SQLAlchemy. You will use its conflict-handling methods for the tool ledger.
- Save app/core.py.
- Confirm your editor shows no Python syntax issue on the updated import.
- In app/core.py, locate the existing LostLease exception.
- Insert the following exception directly above LostLease:
class CrashInjected(RuntimeError):
passWhat does this exception represent?
CrashInjected marks the exact failure point after the tool ledger commits. Raising it prevents the later acknowledgment transaction from running.
- Save app/core.py.
- Confirm the new exception sits immediately above LostLease.
- In app/core.py, locate the end of checkpoint().
- Add the following function directly below checkpoint():
def create_action_intent(claim: Claim) -> None:
key = f"{claim.workflow_id}:resolve-account"
now = utcnow()
with SessionLocal.begin() as session:
session.execute(
pg_insert(action_intents)
.values(
id=uuid4(),
workflow_id=claim.workflow_id,
idempotency_key=key,
status="PENDING",
created_at=now,
)
.on_conflict_do_nothing(index_elements=["idempotency_key"])
)
result = session.execute(
update(workflows)
.where(lease_guard(claim), workflows.c.state == "PLANNED")
.values(state="ACTION_PENDING", updated_at=now)
)
if result.rowcount != 1:
raise LostLease(f"stale fence {claim.fence_token}")
log_event("action_intent_committed", workflow_id=claim.workflow_id, idempotency_key=key)What does this code do?
- The stable key combines the workflow ID with resolve-account so every retry identifies the same logical action.
- The conflict-safe insert preserves the existing intent if the same key already exists.
- The guarded update moves the current owner from PLANNED to ACTION_PENDING inside the same transaction.
- The row-count check rejects a worker whose lease or fencing token has become stale.
- Save app/core.py.
- Confirm your editor recognizes create_action_intent without a syntax warning.
Seeing an issue in the intent function?
Check that pg_insert appears in the PostgreSQL dialect import. Confirm that create_action_intent() begins at the left edge of the file.
Ask for help with the transaction structure:
Commit the idempotent tool effect
The simulated tool uses a separate transaction because an external provider commits independently from AgentFlow. Its tool_ledger row models the durable effect that exists outside the workflow acknowledgment.
- In app/core.py, locate the end of create_action_intent().
- Add the simulated tool function directly below it:
def execute_simulated_tool(idempotency_key: str) -> dict[str, Any]:
result_value = {"account_unlocked": True, "resolution_code": "SIM-200"}
statement = (
pg_insert(tool_ledger)
.values(
idempotency_key=idempotency_key,
result=result_value,
attempts=1,
committed_at=utcnow(),
)
.on_conflict_do_update(
index_elements=["idempotency_key"],
set_={"attempts": tool_ledger.c.attempts + 1},
)
.returning(tool_ledger.c.result, tool_ledger.c.attempts)
)
with SessionLocal.begin() as session:
row = session.execute(statement).mappings().one()
log_event(
"tool_attempt_committed",
idempotency_key=idempotency_key,
attempts=row["attempts"],
)
return dict(row["result"])How does the ledger deduplicate retries?
- The first attempt inserts one result with an attempts value of 1.
- PostgreSQL ON CONFLICT DO UPDATE increments the existing row when the stable key returns.
- The returning clause gives every caller the original committed result plus the latest attempt count.
- One primary-key row represents one business effect even when execution occurs repeatedly.
- Save app/core.py.
- Confirm the new function contains one transaction around the ledger statement.
- In app/core.py, locate the end of execute_simulated_tool().
- Add the acknowledgment function directly below it:
def acknowledge_action(claim: Claim, tool_result: dict[str, Any]) -> None:
now = utcnow()
with SessionLocal.begin() as session:
intent = session.execute(
select(action_intents).where(action_intents.c.workflow_id == claim.workflow_id)
).mappings().one()
session.execute(
update(action_intents)
.where(action_intents.c.id == intent["id"])
.values(status="ACKNOWLEDGED", tool_result=tool_result, acknowledged_at=now)
)
result = session.execute(
update(workflows)
.where(lease_guard(claim), workflows.c.state == "ACTION_PENDING")
.values(
state="COMPLETED",
result=tool_result,
completed_at=now,
updated_at=now,
lease_owner=None,
lease_expires_at=None,
)
)
if result.rowcount != 1:
raise LostLease(f"stale fence {claim.fence_token}")
log_event("workflow_completed", workflow_id=claim.workflow_id, fence_token=claim.fence_token)How does acknowledgment stay fenced?
- The intent changes to ACKNOWLEDGED with the committed tool result.
- The workflow reaches COMPLETED only when lease_guard() still matches the current owner.
- A stale worker triggers LostLease before it can commit the guarded transaction.
- Successful completion clears the lease because the workflow no longer needs an owner.
- Save app/core.py.
- Confirm the workflow update still calls lease_guard(claim).
Seeing an issue in the ledger code?
Check that execute_simulated_tool() uses tool_ledger inside its own transaction. Confirm that acknowledge_action() updates the intent before the guarded workflow update commits.
Ask for help tracing the transaction boundaries:
Crash between effect and acknowledgment
The acknowledgment gap is the interval after execute_simulated_tool() commits and before acknowledge_action() begins. A process loss here leaves durable evidence of the effect while the workflow remains ACTION_PENDING.
- In app/core.py, locate the existing advance_one_step() function.
- Replace the complete function with this version:
def advance_one_step(claim: Claim, crash_after_tool_commit: bool = False) -> str:
row = get_workflow(claim.workflow_id)
if row is None:
raise KeyError(claim.workflow_id)
state = row["state"]
if state == "PENDING":
checkpoint(claim, state, "CLASSIFIED", classification=mock_llm_classify(row["request"]))
elif state == "CLASSIFIED":
checkpoint(claim, state, "PLANNED", plan=mock_llm_plan(row["classification"]))
elif state == "PLANNED":
create_action_intent(claim)
elif state == "ACTION_PENDING":
with SessionLocal() as session:
intent = session.execute(
select(action_intents).where(action_intents.c.workflow_id == claim.workflow_id)
).mappings().one()
tool_result = execute_simulated_tool(intent["idempotency_key"])
if crash_after_tool_commit:
log_event("crash_injected", workflow_id=claim.workflow_id, point="after_tool_commit")
raise CrashInjected("tool committed before workflow acknowledgment")
acknowledge_action(claim, tool_result)
return (get_workflow(claim.workflow_id) or {})["state"]How does recovery replay the action?
- A PLANNED workflow creates its intent before entering ACTION_PENDING.
- An ACTION_PENDING workflow reloads the stable key from action_intents.
- The crash flag raises CrashInjected only after the independent ledger transaction returns.
- A recovery worker follows the same branch with the same key before attempting a fenced acknowledgment.
- Save app/core.py.
- Confirm advance_one_step() now handles PLANNED and ACTION_PENDING as separate states.
✔️ Awesome, I've got everything!
Your app/core.py file now separates the intent, effect, and acknowledgment boundaries.
ⓧ I'd like to double check the full code
Compare your cumulative app/core.py file with this version:
from __future__ import annotations
import json
import logging
import os
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Any
from uuid import UUID, uuid4
from sqlalchemy import (
Column,
DateTime,
Integer,
MetaData,
String,
Table,
Text,
and_,
create_engine,
delete,
insert,
or_,
select,
update,
)
from sqlalchemy.dialects.postgresql import JSONB, UUID as PG_UUID, insert as pg_insert
from sqlalchemy.orm import sessionmaker
DATABASE_URL = os.getenv(
"DATABASE_URL",
"postgresql+psycopg://agentflow:agentflow@db:5432/agentflow",
)
metadata = MetaData()
workflows = Table(
"workflows",
metadata,
Column("id", PG_UUID(as_uuid=True), primary_key=True),
Column("request", Text, nullable=False),
Column("state", String(32), nullable=False),
Column("classification", JSONB, nullable=True),
Column("plan", JSONB, nullable=True),
Column("result", JSONB, nullable=True),
Column("lease_owner", String(100), nullable=True),
Column("lease_expires_at", DateTime(timezone=True), nullable=True),
Column("fence_token", Integer, nullable=False),
Column("recovery_started_at", DateTime(timezone=True), nullable=True),
Column("created_at", DateTime(timezone=True), nullable=False),
Column("updated_at", DateTime(timezone=True), nullable=False),
Column("completed_at", DateTime(timezone=True), nullable=True),
)
action_intents = Table(
"action_intents",
metadata,
Column("id", PG_UUID(as_uuid=True), primary_key=True),
Column("workflow_id", PG_UUID(as_uuid=True), nullable=False, unique=True),
Column("idempotency_key", String(200), nullable=False, unique=True),
Column("status", String(32), nullable=False),
Column("tool_result", JSONB, nullable=True),
Column("created_at", DateTime(timezone=True), nullable=False),
Column("acknowledged_at", DateTime(timezone=True), nullable=True),
)
tool_ledger = Table(
"tool_ledger",
metadata,
Column("idempotency_key", String(200), primary_key=True),
Column("result", JSONB, nullable=False),
Column("attempts", Integer, nullable=False),
Column("committed_at", DateTime(timezone=True), nullable=False),
)
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(engine, expire_on_commit=False)
logger = logging.getLogger("agentflow")
logging.basicConfig(level=logging.INFO, format="%(message)s")
class CrashInjected(RuntimeError):
pass
class LostLease(RuntimeError):
pass
@dataclass(frozen=True)
class Claim:
workflow_id: UUID
worker_id: str
fence_token: int
def utcnow() -> datetime:
return datetime.now(timezone.utc)
def log_event(event: str, **fields: Any) -> None:
logger.info(json.dumps({"event": event, **fields}, default=str, sort_keys=True))
def mock_llm_classify(request: str) -> dict[str, str]:
text = request.lower()
category = "account_access" if "password" in text or "login" in text else "general_support"
return {"category": category, "priority": "normal"}
def mock_llm_plan(classification: dict[str, str]) -> dict[str, list[str]]:
if classification["category"] == "account_access":
return {"steps": ["validate identity", "reset credentials", "confirm access"]}
return {"steps": ["inspect request", "apply documented fix", "confirm resolution"]}
def create_workflow(request: str) -> dict[str, Any]:
workflow_id = uuid4()
now = utcnow()
with SessionLocal.begin() as session:
session.execute(
insert(workflows).values(
id=workflow_id,
request=request,
state="PENDING",
fence_token=0,
created_at=now,
updated_at=now,
)
)
log_event("workflow_created", workflow_id=workflow_id)
return get_workflow(workflow_id) or {}
def get_workflow(workflow_id: UUID) -> dict[str, Any] | None:
with SessionLocal() as session:
row = session.execute(
select(workflows).where(workflows.c.id == workflow_id)
).mappings().one_or_none()
return dict(row) if row else None
def claim_workflow(worker_id: str, lease_seconds: int = 30) -> Claim | None:
now = utcnow()
with SessionLocal.begin() as session:
row = session.execute(
select(workflows)
.where(
workflows.c.state != "COMPLETED",
or_(
workflows.c.lease_owner.is_(None),
workflows.c.lease_expires_at < now,
),
)
.order_by(workflows.c.created_at)
.limit(1)
.with_for_update(skip_locked=True)
).mappings().one_or_none()
if row is None:
return None
token = row["fence_token"] + 1
recovered = row["lease_owner"] is not None
session.execute(
update(workflows)
.where(workflows.c.id == row["id"])
.values(
lease_owner=worker_id,
lease_expires_at=now + timedelta(seconds=lease_seconds),
fence_token=token,
recovery_started_at=(row["recovery_started_at"] or now) if recovered else None,
updated_at=now,
)
)
log_event(
"workflow_claimed",
workflow_id=row["id"],
worker_id=worker_id,
fence_token=token,
recovered=recovered,
)
return Claim(row["id"], worker_id, token)
def lease_guard(claim: Claim):
return and_(
workflows.c.id == claim.workflow_id,
workflows.c.lease_owner == claim.worker_id,
workflows.c.fence_token == claim.fence_token,
workflows.c.lease_expires_at > utcnow(),
)
def checkpoint(claim: Claim, expected_state: str, next_state: str, **values: Any) -> None:
with SessionLocal.begin() as session:
result = session.execute(
update(workflows)
.where(lease_guard(claim), workflows.c.state == expected_state)
.values(state=next_state, updated_at=utcnow(), **values)
)
if result.rowcount != 1:
raise LostLease(f"stale fence {claim.fence_token}")
log_event(
"checkpoint_committed",
workflow_id=claim.workflow_id,
worker_id=claim.worker_id,
state=next_state,
fence_token=claim.fence_token,
)
def create_action_intent(claim: Claim) -> None:
key = f"{claim.workflow_id}:resolve-account"
now = utcnow()
with SessionLocal.begin() as session:
session.execute(
pg_insert(action_intents)
.values(
id=uuid4(),
workflow_id=claim.workflow_id,
idempotency_key=key,
status="PENDING",
created_at=now,
)
.on_conflict_do_nothing(index_elements=["idempotency_key"])
)
result = session.execute(
update(workflows)
.where(lease_guard(claim), workflows.c.state == "PLANNED")
.values(state="ACTION_PENDING", updated_at=now)
)
if result.rowcount != 1:
raise LostLease(f"stale fence {claim.fence_token}")
log_event("action_intent_committed", workflow_id=claim.workflow_id, idempotency_key=key)
def execute_simulated_tool(idempotency_key: str) -> dict[str, Any]:
result_value = {"account_unlocked": True, "resolution_code": "SIM-200"}
statement = (
pg_insert(tool_ledger)
.values(
idempotency_key=idempotency_key,
result=result_value,
attempts=1,
committed_at=utcnow(),
)
.on_conflict_do_update(
index_elements=["idempotency_key"],
set_={"attempts": tool_ledger.c.attempts + 1},
)
.returning(tool_ledger.c.result, tool_ledger.c.attempts)
)
with SessionLocal.begin() as session:
row = session.execute(statement).mappings().one()
log_event(
"tool_attempt_committed",
idempotency_key=idempotency_key,
attempts=row["attempts"],
)
return dict(row["result"])
def acknowledge_action(claim: Claim, tool_result: dict[str, Any]) -> None:
now = utcnow()
with SessionLocal.begin() as session:
intent = session.execute(
select(action_intents).where(action_intents.c.workflow_id == claim.workflow_id)
).mappings().one()
session.execute(
update(action_intents)
.where(action_intents.c.id == intent["id"])
.values(status="ACKNOWLEDGED", tool_result=tool_result, acknowledged_at=now)
)
result = session.execute(
update(workflows)
.where(lease_guard(claim), workflows.c.state == "ACTION_PENDING")
.values(
state="COMPLETED",
result=tool_result,
completed_at=now,
updated_at=now,
lease_owner=None,
lease_expires_at=None,
)
)
if result.rowcount != 1:
raise LostLease(f"stale fence {claim.fence_token}")
log_event("workflow_completed", workflow_id=claim.workflow_id, fence_token=claim.fence_token)
def advance_one_step(claim: Claim, crash_after_tool_commit: bool = False) -> str:
row = get_workflow(claim.workflow_id)
if row is None:
raise KeyError(claim.workflow_id)
state = row["state"]
if state == "PENDING":
checkpoint(claim, state, "CLASSIFIED", classification=mock_llm_classify(row["request"]))
elif state == "CLASSIFIED":
checkpoint(claim, state, "PLANNED", plan=mock_llm_plan(row["classification"]))
elif state == "PLANNED":
create_action_intent(claim)
elif state == "ACTION_PENDING":
with SessionLocal() as session:
intent = session.execute(
select(action_intents).where(action_intents.c.workflow_id == claim.workflow_id)
).mappings().one()
tool_result = execute_simulated_tool(intent["idempotency_key"])
if crash_after_tool_commit:
log_event("crash_injected", workflow_id=claim.workflow_id, point="after_tool_commit")
raise CrashInjected("tool committed before workflow acknowledgment")
acknowledge_action(claim, tool_result)
return (get_workflow(claim.workflow_id) or {})["state"]
def run_once(worker_id: str, crash_after_tool_commit: bool = False) -> bool:
claim = claim_workflow(worker_id)
if claim is None:
return False
for _ in range(8):
state = advance_one_step(claim, crash_after_tool_commit=crash_after_tool_commit)
if state == "COMPLETED":
return True
raise RuntimeError("workflow exceeded transition limit")
def run_until_idle(worker_id: str) -> int:
completed = 0
while run_once(worker_id):
completed += 1
return completed
def expire_lease(workflow_id: UUID) -> None:
with SessionLocal.begin() as session:
session.execute(
update(workflows)
.where(workflows.c.id == workflow_id)
.values(lease_expires_at=utcnow() - timedelta(seconds=1))
)
log_event("lease_expired_for_test", workflow_id=workflow_id)
def run_competing_workers(*worker_ids: str) -> list[int]:
with ThreadPoolExecutor(max_workers=len(worker_ids)) as executor:
futures = [executor.submit(run_until_idle, worker_id) for worker_id in worker_ids]
return [future.result() for future in futures]
def reset_database() -> None:
with SessionLocal.begin() as session:
session.execute(delete(action_intents))
session.execute(delete(tool_ledger))
session.execute(delete(workflows))How to compare this file
Check the imports first. Compare the intent, tool, acknowledgment, and state-transition functions next. Finish by confirming the existing lease and worker functions remain unchanged.
The implementation now contains the failure point. The first test slice proves that the tool can commit while the workflow remains unfinished.
- In tests/test_agentflow.py, locate the existing import group from app.core.
- Replace that import group with this cumulative version:
from app.core import (
CrashInjected,
SessionLocal,
action_intents,
advance_one_step,
claim_workflow,
create_workflow,
expire_lease,
get_workflow,
run_competing_workers,
run_once,
tool_ledger,
workflows,
)What do these test imports add?
The test catches CrashInjected as an expected failure point. It inspects action_intents and tool_ledger after recovery.
- Save tests/test_agentflow.py.
- Confirm each imported identifier exists in app/core.py.
- In tests/test_agentflow.py, locate the end of test_two_workers_complete_concurrent_queue().
- Add this first test slice directly below it:
def test_ack_gap_recovery_has_one_business_effect():
workflow = create_workflow("Please unlock my account after a password reset")
with pytest.raises(CrashInjected):
run_once("crasher", crash_after_tool_commit=True)
after_crash = get_workflow(workflow["id"])
assert after_crash is not None
assert after_crash["state"] == "ACTION_PENDING"What does this test slice prove?
Pytest expects the injected crash because the failure is deliberate. The final assertion proves AgentFlow never committed its acknowledgment.
The separate tool transaction has already returned before the exception occurs. This creates the uncertainty that a recovery worker must reconcile.
- Save tests/test_agentflow.py.
Before you run the test, do you think the workflow reaches COMPLETED after the tool commits and the worker crashes?
- Run the acknowledgment-gap test through Docker Compose with this command:
docker compose run --rm api pytest -q tests/test_agentflow.py::test_ack_gap_recovery_has_one_business_effectWhat should you see?
You should see 1 passed. The passing assertion confirms that the workflow is stranded in ACTION_PENDING after the tool transaction commits.
That stranded checkpoint is the intended shortfall. A lease controls database ownership while the acknowledgment gap still survives a process crash.
- Return to tests/test_agentflow.py.
- Continue the same test directly below its ACTION_PENDING assertion with this recovery logic:
expire_lease(workflow["id"])
with ThreadPoolExecutor(max_workers=2) as executor:
futures = [
executor.submit(run_once, "recovery-a"),
executor.submit(run_once, "recovery-b"),
]
[future.result() for future in futures]
finished = get_workflow(workflow["id"])
with SessionLocal() as session:
ledger_rows = session.execute(select(tool_ledger)).mappings().all()
intent = session.execute(select(action_intents)).mappings().one()
assert finished is not None
assert finished["state"] == "COMPLETED"
assert intent["status"] == "ACKNOWLEDGED"
assert len(ledger_rows) == 1
assert ledger_rows[0]["attempts"] >= 2How does the recovery proof work?
- expire_lease() makes the abandoned workflow eligible for a new owner.
- recovery-a and recovery-b compete through separate worker sessions.
- The winning worker replays the stable key before acknowledging the action.
- The assertions require multiple attempts plus exactly one ledger row.
- Save tests/test_agentflow.py.
Before you run the completed test, do you expect two recovery workers to create two business-effect rows?
- Run the completed recovery test with the same command:
docker compose run --rm api pytest -q tests/test_agentflow.py::test_ack_gap_recovery_has_one_business_effectWhat should you see now?
You should see 1 passed again. The completed assertions prove the workflow reaches COMPLETED with one ledger row.
The row records an attempts value of at least 2. Execution repeated while the business effect remained singular.
You closed the hardest durability gap in AgentFlow. A committed tool result can now survive a lost acknowledgment and converge through safe replay.
Test not passing?
Confirm the API container can reach the running database. Check that revision 0001_agentflow remains applied.
Verify that the crash occurs after execute_simulated_tool() returns. Confirm that both recovery workers use the same intent key.
Ask for help tracing the failed assertion:
✔️ Awesome, I've got everything!
Your recovery test now proves at-least-once execution with exactly one business effect per stable key.
ⓧ I'd like to double check the full code
Compare your cumulative tests/test_agentflow.py file with this version:
from concurrent.futures import ThreadPoolExecutor
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select
from app.api import app
from app.core import (
CrashInjected,
SessionLocal,
action_intents,
advance_one_step,
claim_workflow,
create_workflow,
expire_lease,
get_workflow,
run_competing_workers,
run_once,
tool_ledger,
workflows,
)
client = TestClient(app)
def test_normal_api_workflow_completes():
created = client.post("/workflows", json={"request": "I cannot login after a password reset"}).json()
response = client.post("/workers/api-worker/run-once").json()
finished = client.get(f"/workflows/{created['id']}").json()
assert response == {"worked": True}
assert finished["state"] == "COMPLETED"
assert finished["result"]["account_unlocked"] is True
def test_checkpoint_survives_interruption():
workflow = create_workflow("My password no longer works")
claim = claim_workflow("interrupted-worker")
assert claim is not None
assert advance_one_step(claim) == "CLASSIFIED"
persisted = get_workflow(workflow["id"])
assert persisted is not None
assert persisted["state"] == "CLASSIFIED"
assert persisted["classification"]["category"] == "account_access"
def test_two_workers_complete_concurrent_queue():
for number in range(8):
create_workflow(f"Support request {number}")
completed_per_worker = run_competing_workers("worker-a", "worker-b")
with SessionLocal() as session:
states = session.execute(select(workflows.c.state)).scalars().all()
assert sum(completed_per_worker) == 8
assert states == ["COMPLETED"] * 8
def test_ack_gap_recovery_has_one_business_effect():
workflow = create_workflow("Please unlock my account after a password reset")
with pytest.raises(CrashInjected):
run_once("crasher", crash_after_tool_commit=True)
after_crash = get_workflow(workflow["id"])
assert after_crash is not None
assert after_crash["state"] == "ACTION_PENDING"
expire_lease(workflow["id"])
with ThreadPoolExecutor(max_workers=2) as executor:
futures = [
executor.submit(run_once, "recovery-a"),
executor.submit(run_once, "recovery-b"),
]
[future.result() for future in futures]
finished = get_workflow(workflow["id"])
with SessionLocal() as session:
ledger_rows = session.execute(select(tool_ledger)).mappings().all()
intent = session.execute(select(action_intents)).mappings().one()
assert finished is not None
assert finished["state"] == "COMPLETED"
assert intent["status"] == "ACKNOWLEDGED"
assert len(ledger_rows) == 1
assert ledger_rows[0]["attempts"] >= 2How to compare this test file
Check the updated core imports first. Compare the acknowledgment-gap test next. Confirm the earlier checkpoint and concurrency tests remain unchanged.
Your intent, ledger, and acknowledgment boundaries now recover the failure that leases alone cannot solve. Next up, you will turn these guarantees into measured reliability evidence.
Measure Reliability Evidence
Your acknowledgment-gap recovery now preserves one business effect even when a worker executes the tool more than once. The durable records in PostgreSQL prove the behavior for one run.
A reliability claim also needs repeatable tests. This step turns the workflow state into measured evidence through structured JSON events, integration tests, a generated report, and documented limitations.
In this step, get ready to:
- Calculate completion, execution, effect, duplication, and recovery measurements.
- Generate reliability evidence from the acknowledgment-gap integration test.
- Document the architecture, reliability claim, and production limitations.
Calculate reliability measurements
Your existing state transitions record lifecycle events through log_event(). The next layer aggregates the persisted rows into measurements that describe completion, retries, effects, and recovery time.
- Switch back to app/core.py in the VS Code editor from earlier.
- Update the import section so it includes the path helper used to create the report file:
from __future__ import annotations
import json
import logging
import os
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any
from uuid import UUID, uuid4Why add the path helper?
The Path class creates the artifacts directory when needed. It also writes the reliability report using an explicit text encoding.
- Add the database aggregation portion of reliability_metrics() near the bottom of app/core.py:
def reliability_metrics() -> dict[str, Any]:
with SessionLocal() as session:
total = session.execute(select(func.count()).select_from(workflows)).scalar_one()
completed = session.execute(
select(func.count()).select_from(workflows).where(workflows.c.state == "COMPLETED")
).scalar_one()
effects = session.execute(select(func.count()).select_from(tool_ledger)).scalar_one()
attempts = session.execute(
select(func.coalesce(func.sum(tool_ledger.c.attempts), 0))
).scalar_one()
intents = session.execute(select(func.count()).select_from(action_intents)).scalar_one()
recovered = session.execute(
select(workflows.c.recovery_started_at, workflows.c.completed_at).where(
workflows.c.recovery_started_at.is_not(None),
workflows.c.completed_at.is_not(None),
)
).all()What do these queries measure?
- The workflow counts measure how much submitted work reaches COMPLETED.
- The ledger counts separate tool execution attempts from persisted business effects.
- The recovery timestamps capture how long an expired workflow takes to complete after reclamation begins.
- Complete reliability_metrics() by adding the latency calculation and returned measurement fields directly below the database queries:
latencies = [
(completed_at - started_at).total_seconds() * 1000
for started_at, completed_at in recovered
]
return {
"workflow_completion_rate": completed / total if total else 0.0,
"workflows_total": total,
"workflows_completed": completed,
"tool_execution_attempts": attempts,
"business_effects": effects,
"duplicate_business_effects": max(0, effects - intents),
"recovery_latency_ms": round(sum(latencies) / len(latencies), 3) if latencies else None,
"semantics": "at-least-once execution, exactly-once business effect per idempotency key",
}How do the measurements support the claim?
The attempts counter can exceed the effect count because retries may cross the tool boundary more than once. The duplicate-effect measurement stays at zero when each stable idempotency key owns one ledger row.
The semantics string states the exact boundary being demonstrated. It avoids claiming that execution itself happens once.
- Add the report writer directly below reliability_metrics():
def write_reliability_report() -> dict[str, Any]:
report = reliability_metrics()
path = Path("artifacts/reliability-report.json")
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
return reportWhat does the report writer do?
The function takes a fresh measurement snapshot after the recovery scenario. It writes sorted, indented JSON to artifacts/reliability-report.json so the evidence can be reviewed outside the test process.
- Replace the contents of app/api.py with the completed API implementation below:
from uuid import UUID
from fastapi import FastAPI
from pydantic import BaseModel
from app.core import create_workflow, get_workflow, reliability_metrics, run_once
app = FastAPI(title="AgentFlow", version="1.0.0")
class WorkflowRequest(BaseModel):
request: str
@app.get("/health")
def health() -> dict[str, str]:
return {"status": "ok"}
@app.post("/workflows")
def submit_workflow(payload: WorkflowRequest) -> dict:
return create_workflow(payload.request)
@app.get("/workflows/{workflow_id}")
def inspect_workflow(workflow_id: UUID) -> dict:
return get_workflow(workflow_id) or {"error": "not_found"}
@app.post("/workers/{worker_id}/run-once")
def run_worker_once(worker_id: str) -> dict[str, bool]:
return {"worked": run_once(worker_id)}
@app.get("/metrics")
def metrics() -> dict:
return reliability_metrics()What changed in the API?
The FastAPI service now exposes the persisted measurements through /metrics. The endpoint calculates its response from current database state each time it receives a request.
- Save app/core.py.
- Save app/api.py.
- Reload the running service with the updated source by running these commands:
docker compose down
docker compose up -d apiWhat do these commands change?
The first command replaces the existing containers while retaining the named database volume. The second command starts the API after the database passes its health check.
- Return to http://localhost:8000/docs in your browser.
- Confirm that the API documentation now lists GET /metrics.
The measurement endpoint is live. Your service can now expose reliability state without reading test output manually.
Cannot see the metrics endpoint?
Confirm that app/api.py imports reliability_metrics. Check that both edited files were saved before the containers restarted.
If the API does not start, help me trace the metrics endpoint startup failure.
Generate evidence from the recovery test
A report has value when the same scenario that proves recovery also generates it. The Pytest integration test therefore asserts the measurements after two workers recover the acknowledgment gap.
- Switch back to tests/test_agentflow.py in VS Code.
- Update the import group so the test can call write_reliability_report():
from concurrent.futures import ThreadPoolExecutor
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select
from app.api import app
from app.core import (
CrashInjected,
SessionLocal,
action_intents,
advance_one_step,
claim_workflow,
create_workflow,
expire_lease,
get_workflow,
run_competing_workers,
run_once,
tool_ledger,
workflows,
write_reliability_report,
)Why import the report writer in the test?
The recovery test already controls the exact crash and replay sequence. Calling the writer there ensures the report describes the scenario whose final database state the assertions just inspected.
- Add the report assertions to the end of test_ack_gap_recovery_has_one_business_effect():
report = write_reliability_report()
assert report["workflow_completion_rate"] == 1.0
assert report["business_effects"] == 1
assert report["duplicate_business_effects"] == 0
assert report["tool_execution_attempts"] >= 2
assert report["recovery_latency_ms"] is not NoneWhat does this prove?
The completed workflow contributes a completion rate of 1.0. One ledger row records the business effect even though the tool attempt count reaches at least two.
A non-null recovery latency confirms that the workflow was reclaimed after its abandoned lease expired. The report now ties the reliability claim to measurable database state.
- Save tests/test_agentflow.py.
Before you run the recovery test, which value do you expect to be larger: tool execution attempts or persisted business effects?
- Run the acknowledgment-gap integration test and generate the report with this command:
docker compose run --rm api pytest -q tests/test_agentflow.py::test_ack_gap_recovery_has_one_business_effectWhat should the test produce?
The test passes after the crash leaves the workflow in ACTION_PENDING and recovery completes it. The same run creates artifacts/reliability-report.json through the source bind mount.
- Expand the artifacts folder in the VS Code Explorer.
- Open artifacts/reliability-report.json.
- Confirm that workflow_completion_rate is 1.0.
- Confirm that duplicate_business_effects is 0.
- Confirm that tool_execution_attempts is at least 2.
That is the central result: the recovery path executed the tool boundary more than once while retaining one business effect. Your reliability claim now has generated evidence behind it.
Report missing or assertions failing?
Make sure the test calls write_reliability_report() after the workflow and ledger assertions. Confirm that the repository bind mount in compose.yaml still maps the current folder to /app.
For a failing count or missing artifact, help me debug the acknowledgment-gap reliability report.
Document and prove the reliability claim
Measured evidence explains what happened. The architecture and limitations explain why the result is credible, where each transaction ends, and which guarantees remain outside this teaching implementation.
- Replace docs/architecture.md with the completed architecture documentation below:
# AgentFlow Architecture
```mermaid
flowchart LR
C[API client] --> A[FastAPI]
A --> W[(workflows)]
A --> R[Worker identities]
R -->|claim with SKIP LOCKED| W
R -->|checkpoint with fence| W
R --> I[(action_intents)]
R -->|stable idempotency key| T[Simulated external tool]
T -->|independent transaction| L[(tool_ledger)]
R -->|reconcile then acknowledge| I
R -->|complete| W
```
Durability boundaries:
1. Each workflow transition commits independently.
2. Entering `ACTION_PENDING` and creating its intent share one transaction.
3. The simulated external tool commits `tool_ledger` in a separate transaction.
4. Intent acknowledgment and workflow completion share a later transaction guarded by the current fencing token.
The ledger shares the PostgreSQL server only to keep the lab local. It is deliberately accessed through a separate transaction to model a remote system whose commit cannot be atomic with AgentFlow acknowledgment.How does the diagram expose the gap?
The Mermaid flowchart separates the workflow checkpoint, action intent, tool ledger, and acknowledgment. The independent ledger transaction makes the crash window visible instead of hiding it behind one database box.
- Save docs/architecture.md.
- Preview docs/architecture.md in VS Code.
- Confirm that the rendered flow shows the worker path through the intent, simulated tool, ledger, acknowledgment, and workflow completion.
- Replace docs/production-limitations.md with the completed limitations below:
# Production Limitations
- The deterministic mock LLM teaches orchestration, not model quality, prompt security, or nondeterministic replay.
- Worker threads are test harness identities. Production workers should be separate supervised processes with heartbeats, graceful shutdown, bounded retries, and backoff.
- Leases rely on application timestamps. A production design should use database time consistently and renew long-running leases.
- Fencing protects AgentFlow database writes. A real external service must enforce its idempotency key or fencing token independently.
- The simulated tool ledger is in the same PostgreSQL cluster. A remote provider introduces network timeouts, retention policies, authentication, rate limits, and provider-specific reconciliation APIs.
- `SKIP LOCKED` is queue-oriented and intentionally provides an inconsistent view. Fairness, starvation, priorities, dead-letter handling, and queue partitioning are not implemented.
- The action ledger needs retention and archival rules so stable keys are not removed while retries remain possible.
- The API omits authentication, authorization, tenant isolation, request quotas, tracing, secret management, and schema-evolution deployment strategy.
- Exactly-once execution is not claimed. The worker can execute the tool call multiple times. The demonstrated guarantee is exactly one persisted business effect for one stable idempotency key.Why document the non-guarantees?
The limitations separate the local simulation from a production workflow platform. They also identify the controls a remote provider must enforce for stable keys and fencing to remain effective.
- Save docs/production-limitations.md.
- Update the first half of README.md so the run and evidence instructions match the completed repository:
# AgentFlow
AgentFlow is a local durable workflow engine for a deterministic AI support agent. It demonstrates PostgreSQL checkpoints, leases, fencing tokens, action intents, acknowledgment-gap recovery, and idempotent business effects.
## Run
```bash
docker compose up -d db
docker compose run --build --rm api alembic upgrade head
docker compose up -d api
```
Open `http://localhost:8000/docs`.
## Test and generate reliability evidence
```bash
docker compose run --rm api pytest -q
```
The crash-recovery test writes `artifacts/reliability-report.json`.What does this README section provide?
The runbook gives another engineer a reproducible path from an empty runtime to a migrated API. The test command connects the suite directly to its generated evidence artifact.
- Complete README.md with the reliability claim, documentation links, and cleanup commands:
## Reliability claim
AgentFlow uses at-least-once tool execution. It demonstrates exactly one persisted business effect per stable idempotency key. A crash can cause the tool boundary to be invoked again, so it does not claim exactly-once execution.
## Documentation
- `docs/architecture.md`
- `docs/production-limitations.md`
- `artifacts/reliability-report.json`
## Cleanup
```bash
docker compose down
# Delete the PostgreSQL volume too:
docker compose down --volumes
```Why state the claim this precisely?
The README distinguishes invocation count from persisted effect count. That distinction keeps the guarantee aligned with the ledger evidence produced by the recovery test.
- Save README.md.
✔️ Awesome, I've got everything!
Your metrics, tests, report writer, architecture notes, limitations, and README now describe the same reliability boundary.
ⓧ I'd like to double check the full code
Compare your completed API with this final app/api.py file.
from uuid import UUID
from fastapi import FastAPI
from pydantic import BaseModel
from app.core import create_workflow, get_workflow, reliability_metrics, run_once
app = FastAPI(title="AgentFlow", version="1.0.0")
class WorkflowRequest(BaseModel):
request: str
@app.get("/health")
def health() -> dict[str, str]:
return {"status": "ok"}
@app.post("/workflows")
def submit_workflow(payload: WorkflowRequest) -> dict:
return create_workflow(payload.request)
@app.get("/workflows/{workflow_id}")
def inspect_workflow(workflow_id: UUID) -> dict:
return get_workflow(workflow_id) or {"error": "not_found"}
@app.post("/workers/{worker_id}/run-once")
def run_worker_once(worker_id: str) -> dict[str, bool]:
return {"worked": run_once(worker_id)}
@app.get("/metrics")
def metrics() -> dict:
return reliability_metrics()Compare your completed workflow engine with this final app/core.py file.
from __future__ import annotations
import json
import logging
import os
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any
from uuid import UUID, uuid4
from sqlalchemy import (
Column,
DateTime,
Integer,
MetaData,
String,
Table,
Text,
and_,
create_engine,
delete,
func,
insert,
or_,
select,
update,
)
from sqlalchemy.dialects.postgresql import JSONB, UUID as PG_UUID, insert as pg_insert
from sqlalchemy.orm import sessionmaker
DATABASE_URL = os.getenv(
"DATABASE_URL",
"postgresql+psycopg://agentflow:agentflow@db:5432/agentflow",
)
metadata = MetaData()
workflows = Table(
"workflows",
metadata,
Column("id", PG_UUID(as_uuid=True), primary_key=True),
Column("request", Text, nullable=False),
Column("state", String(32), nullable=False),
Column("classification", JSONB, nullable=True),
Column("plan", JSONB, nullable=True),
Column("result", JSONB, nullable=True),
Column("lease_owner", String(100), nullable=True),
Column("lease_expires_at", DateTime(timezone=True), nullable=True),
Column("fence_token", Integer, nullable=False),
Column("recovery_started_at", DateTime(timezone=True), nullable=True),
Column("created_at", DateTime(timezone=True), nullable=False),
Column("updated_at", DateTime(timezone=True), nullable=False),
Column("completed_at", DateTime(timezone=True), nullable=True),
)
action_intents = Table(
"action_intents",
metadata,
Column("id", PG_UUID(as_uuid=True), primary_key=True),
Column("workflow_id", PG_UUID(as_uuid=True), nullable=False, unique=True),
Column("idempotency_key", String(200), nullable=False, unique=True),
Column("status", String(32), nullable=False),
Column("tool_result", JSONB, nullable=True),
Column("created_at", DateTime(timezone=True), nullable=False),
Column("acknowledged_at", DateTime(timezone=True), nullable=True),
)
tool_ledger = Table(
"tool_ledger",
metadata,
Column("idempotency_key", String(200), primary_key=True),
Column("result", JSONB, nullable=False),
Column("attempts", Integer, nullable=False),
Column("committed_at", DateTime(timezone=True), nullable=False),
)
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(engine, expire_on_commit=False)
logger = logging.getLogger("agentflow")
logging.basicConfig(level=logging.INFO, format="%(message)s")
class CrashInjected(RuntimeError):
pass
class LostLease(RuntimeError):
pass
@dataclass(frozen=True)
class Claim:
workflow_id: UUID
worker_id: str
fence_token: int
def utcnow() -> datetime:
return datetime.now(timezone.utc)
def log_event(event: str, **fields: Any) -> None:
logger.info(json.dumps({"event": event, **fields}, default=str, sort_keys=True))
def mock_llm_classify(request: str) -> dict[str, str]:
text = request.lower()
category = "account_access" if "password" in text or "login" in text else "general_support"
return {"category": category, "priority": "normal"}
def mock_llm_plan(classification: dict[str, str]) -> dict[str, list[str]]:
if classification["category"] == "account_access":
return {"steps": ["validate identity", "reset credentials", "confirm access"]}
return {"steps": ["inspect request", "apply documented fix", "confirm resolution"]}
def create_workflow(request: str) -> dict[str, Any]:
workflow_id = uuid4()
now = utcnow()
with SessionLocal.begin() as session:
session.execute(
insert(workflows).values(
id=workflow_id,
request=request,
state="PENDING",
fence_token=0,
created_at=now,
updated_at=now,
)
)
log_event("workflow_created", workflow_id=workflow_id)
return get_workflow(workflow_id) or {}
def get_workflow(workflow_id: UUID) -> dict[str, Any] | None:
with SessionLocal() as session:
row = session.execute(
select(workflows).where(workflows.c.id == workflow_id)
).mappings().one_or_none()
return dict(row) if row else None
def claim_workflow(worker_id: str, lease_seconds: int = 30) -> Claim | None:
now = utcnow()
with SessionLocal.begin() as session:
row = session.execute(
select(workflows)
.where(
workflows.c.state != "COMPLETED",
or_(
workflows.c.lease_owner.is_(None),
workflows.c.lease_expires_at < now,
),
)
.order_by(workflows.c.created_at)
.limit(1)
.with_for_update(skip_locked=True)
).mappings().one_or_none()
if row is None:
return None
token = row["fence_token"] + 1
recovered = row["lease_owner"] is not None
session.execute(
update(workflows)
.where(workflows.c.id == row["id"])
.values(
lease_owner=worker_id,
lease_expires_at=now + timedelta(seconds=lease_seconds),
fence_token=token,
recovery_started_at=(row["recovery_started_at"] or now) if recovered else None,
updated_at=now,
)
)
log_event(
"workflow_claimed",
workflow_id=row["id"],
worker_id=worker_id,
fence_token=token,
recovered=recovered,
)
return Claim(row["id"], worker_id, token)
def lease_guard(claim: Claim):
return and_(
workflows.c.id == claim.workflow_id,
workflows.c.lease_owner == claim.worker_id,
workflows.c.fence_token == claim.fence_token,
workflows.c.lease_expires_at > utcnow(),
)
def checkpoint(claim: Claim, expected_state: str, next_state: str, **values: Any) -> None:
with SessionLocal.begin() as session:
result = session.execute(
update(workflows)
.where(lease_guard(claim), workflows.c.state == expected_state)
.values(state=next_state, updated_at=utcnow(), **values)
)
if result.rowcount != 1:
raise LostLease(f"stale fence {claim.fence_token}")
log_event(
"checkpoint_committed",
workflow_id=claim.workflow_id,
worker_id=claim.worker_id,
state=next_state,
fence_token=claim.fence_token,
)
def create_action_intent(claim: Claim) -> None:
key = f"{claim.workflow_id}:resolve-account"
now = utcnow()
with SessionLocal.begin() as session:
session.execute(
pg_insert(action_intents)
.values(
id=uuid4(),
workflow_id=claim.workflow_id,
idempotency_key=key,
status="PENDING",
created_at=now,
)
.on_conflict_do_nothing(index_elements=["idempotency_key"])
)
result = session.execute(
update(workflows)
.where(lease_guard(claim), workflows.c.state == "PLANNED")
.values(state="ACTION_PENDING", updated_at=now)
)
if result.rowcount != 1:
raise LostLease(f"stale fence {claim.fence_token}")
log_event("action_intent_committed", workflow_id=claim.workflow_id, idempotency_key=key)
def execute_simulated_tool(idempotency_key: str) -> dict[str, Any]:
result_value = {"account_unlocked": True, "resolution_code": "SIM-200"}
statement = (
pg_insert(tool_ledger)
.values(
idempotency_key=idempotency_key,
result=result_value,
attempts=1,
committed_at=utcnow(),
)
.on_conflict_do_update(
index_elements=["idempotency_key"],
set_={"attempts": tool_ledger.c.attempts + 1},
)
.returning(tool_ledger.c.result, tool_ledger.c.attempts)
)
with SessionLocal.begin() as session:
row = session.execute(statement).mappings().one()
log_event(
"tool_attempt_committed",
idempotency_key=idempotency_key,
attempts=row["attempts"],
)
return dict(row["result"])
def acknowledge_action(claim: Claim, tool_result: dict[str, Any]) -> None:
now = utcnow()
with SessionLocal.begin() as session:
intent = session.execute(
select(action_intents).where(action_intents.c.workflow_id == claim.workflow_id)
).mappings().one()
session.execute(
update(action_intents)
.where(action_intents.c.id == intent["id"])
.values(status="ACKNOWLEDGED", tool_result=tool_result, acknowledged_at=now)
)
result = session.execute(
update(workflows)
.where(lease_guard(claim), workflows.c.state == "ACTION_PENDING")
.values(
state="COMPLETED",
result=tool_result,
completed_at=now,
updated_at=now,
lease_owner=None,
lease_expires_at=None,
)
)
if result.rowcount != 1:
raise LostLease(f"stale fence {claim.fence_token}")
log_event("workflow_completed", workflow_id=claim.workflow_id, fence_token=claim.fence_token)
def advance_one_step(claim: Claim, crash_after_tool_commit: bool = False) -> str:
row = get_workflow(claim.workflow_id)
if row is None:
raise KeyError(claim.workflow_id)
state = row["state"]
if state == "PENDING":
checkpoint(claim, state, "CLASSIFIED", classification=mock_llm_classify(row["request"]))
elif state == "CLASSIFIED":
checkpoint(claim, state, "PLANNED", plan=mock_llm_plan(row["classification"]))
elif state == "PLANNED":
create_action_intent(claim)
elif state == "ACTION_PENDING":
with SessionLocal() as session:
intent = session.execute(
select(action_intents).where(action_intents.c.workflow_id == claim.workflow_id)
).mappings().one()
tool_result = execute_simulated_tool(intent["idempotency_key"])
if crash_after_tool_commit:
log_event("crash_injected", workflow_id=claim.workflow_id, point="after_tool_commit")
raise CrashInjected("tool committed before workflow acknowledgment")
acknowledge_action(claim, tool_result)
return (get_workflow(claim.workflow_id) or {})["state"]
def run_once(worker_id: str, crash_after_tool_commit: bool = False) -> bool:
claim = claim_workflow(worker_id)
if claim is None:
return False
for _ in range(8):
state = advance_one_step(claim, crash_after_tool_commit=crash_after_tool_commit)
if state == "COMPLETED":
return True
raise RuntimeError("workflow exceeded transition limit")
def run_until_idle(worker_id: str) -> int:
completed = 0
while run_once(worker_id):
completed += 1
return completed
def expire_lease(workflow_id: UUID) -> None:
with SessionLocal.begin() as session:
session.execute(
update(workflows)
.where(workflows.c.id == workflow_id)
.values(lease_expires_at=utcnow() - timedelta(seconds=1))
)
log_event("lease_expired_for_test", workflow_id=workflow_id)
def run_competing_workers(*worker_ids: str) -> list[int]:
with ThreadPoolExecutor(max_workers=len(worker_ids)) as executor:
futures = [executor.submit(run_until_idle, worker_id) for worker_id in worker_ids]
return [future.result() for future in futures]
def reliability_metrics() -> dict[str, Any]:
with SessionLocal() as session:
total = session.execute(select(func.count()).select_from(workflows)).scalar_one()
completed = session.execute(
select(func.count()).select_from(workflows).where(workflows.c.state == "COMPLETED")
).scalar_one()
effects = session.execute(select(func.count()).select_from(tool_ledger)).scalar_one()
attempts = session.execute(
select(func.coalesce(func.sum(tool_ledger.c.attempts), 0))
).scalar_one()
intents = session.execute(select(func.count()).select_from(action_intents)).scalar_one()
recovered = session.execute(
select(workflows.c.recovery_started_at, workflows.c.completed_at).where(
workflows.c.recovery_started_at.is_not(None),
workflows.c.completed_at.is_not(None),
)
).all()
latencies = [
(completed_at - started_at).total_seconds() * 1000
for started_at, completed_at in recovered
]
return {
"workflow_completion_rate": completed / total if total else 0.0,
"workflows_total": total,
"workflows_completed": completed,
"tool_execution_attempts": attempts,
"business_effects": effects,
"duplicate_business_effects": max(0, effects - intents),
"recovery_latency_ms": round(sum(latencies) / len(latencies), 3) if latencies else None,
"semantics": "at-least-once execution, exactly-once business effect per idempotency key",
}
def write_reliability_report() -> dict[str, Any]:
report = reliability_metrics()
path = Path("artifacts/reliability-report.json")
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
return report
def reset_database() -> None:
with SessionLocal.begin() as session:
session.execute(delete(action_intents))
session.execute(delete(tool_ledger))
session.execute(delete(workflows))Compare your integration suite with this final tests/test_agentflow.py file.
from concurrent.futures import ThreadPoolExecutor
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select
from app.api import app
from app.core import (
CrashInjected,
SessionLocal,
action_intents,
advance_one_step,
claim_workflow,
create_workflow,
expire_lease,
get_workflow,
run_competing_workers,
run_once,
tool_ledger,
workflows,
write_reliability_report,
)
client = TestClient(app)
def test_normal_api_workflow_completes():
created = client.post("/workflows", json={"request": "I cannot login after a password reset"}).json()
response = client.post("/workers/api-worker/run-once").json()
finished = client.get(f"/workflows/{created['id']}").json()
assert response == {"worked": True}
assert finished["state"] == "COMPLETED"
assert finished["result"]["account_unlocked"] is True
def test_checkpoint_survives_interruption():
workflow = create_workflow("My password no longer works")
claim = claim_workflow("interrupted-worker")
assert claim is not None
assert advance_one_step(claim) == "CLASSIFIED"
persisted = get_workflow(workflow["id"])
assert persisted is not None
assert persisted["state"] == "CLASSIFIED"
assert persisted["classification"]["category"] == "account_access"
def test_two_workers_complete_concurrent_queue():
for number in range(8):
create_workflow(f"Support request {number}")
completed_per_worker = run_competing_workers("worker-a", "worker-b")
with SessionLocal() as session:
states = session.execute(select(workflows.c.state)).scalars().all()
assert sum(completed_per_worker) == 8
assert states == ["COMPLETED"] * 8
def test_ack_gap_recovery_has_one_business_effect():
workflow = create_workflow("Please unlock my account after a password reset")
with pytest.raises(CrashInjected):
run_once("crasher", crash_after_tool_commit=True)
after_crash = get_workflow(workflow["id"])
assert after_crash is not None
assert after_crash["state"] == "ACTION_PENDING"
expire_lease(workflow["id"])
with ThreadPoolExecutor(max_workers=2) as executor:
futures = [
executor.submit(run_once, "recovery-a"),
executor.submit(run_once, "recovery-b"),
]
[future.result() for future in futures]
finished = get_workflow(workflow["id"])
with SessionLocal() as session:
ledger_rows = session.execute(select(tool_ledger)).mappings().all()
intent = session.execute(select(action_intents)).mappings().one()
assert finished is not None
assert finished["state"] == "COMPLETED"
assert intent["status"] == "ACKNOWLEDGED"
assert len(ledger_rows) == 1
assert ledger_rows[0]["attempts"] >= 2
report = write_reliability_report()
assert report["workflow_completion_rate"] == 1.0
assert report["business_effects"] == 1
assert report["duplicate_business_effects"] == 0
assert report["tool_execution_attempts"] >= 2
assert report["recovery_latency_ms"] is not NoneCompare your architecture notes with this final docs/architecture.md file.
# AgentFlow Architecture
```mermaid
flowchart LR
C[API client] --> A[FastAPI]
A --> W[(workflows)]
A --> R[Worker identities]
R -->|claim with SKIP LOCKED| W
R -->|checkpoint with fence| W
R --> I[(action_intents)]
R -->|stable idempotency key| T[Simulated external tool]
T -->|independent transaction| L[(tool_ledger)]
R -->|reconcile then acknowledge| I
R -->|complete| W
```
Durability boundaries:
1. Each workflow transition commits independently.
2. Entering `ACTION_PENDING` and creating its intent share one transaction.
3. The simulated external tool commits `tool_ledger` in a separate transaction.
4. Intent acknowledgment and workflow completion share a later transaction guarded by the current fencing token.
The ledger shares the PostgreSQL server only to keep the lab local. It is deliberately accessed through a separate transaction to model a remote system whose commit cannot be atomic with AgentFlow acknowledgment.Compare your limitations with this final docs/production-limitations.md file.
# Production Limitations
- The deterministic mock LLM teaches orchestration, not model quality, prompt security, or nondeterministic replay.
- Worker threads are test harness identities. Production workers should be separate supervised processes with heartbeats, graceful shutdown, bounded retries, and backoff.
- Leases rely on application timestamps. A production design should use database time consistently and renew long-running leases.
- Fencing protects AgentFlow database writes. A real external service must enforce its idempotency key or fencing token independently.
- The simulated tool ledger is in the same PostgreSQL cluster. A remote provider introduces network timeouts, retention policies, authentication, rate limits, and provider-specific reconciliation APIs.
- `SKIP LOCKED` is queue-oriented and intentionally provides an inconsistent view. Fairness, starvation, priorities, dead-letter handling, and queue partitioning are not implemented.
- The action ledger needs retention and archival rules so stable keys are not removed while retries remain possible.
- The API omits authentication, authorization, tenant isolation, request quotas, tracing, secret management, and schema-evolution deployment strategy.
- Exactly-once execution is not claimed. The worker can execute the tool call multiple times. The demonstrated guarantee is exactly one persisted business effect for one stable idempotency key.Compare your project runbook with this final README.md file.
# AgentFlow
AgentFlow is a local durable workflow engine for a deterministic AI support agent. It demonstrates PostgreSQL checkpoints, leases, fencing tokens, action intents, acknowledgment-gap recovery, and idempotent business effects.
## Run
```bash
docker compose up -d db
docker compose run --build --rm api alembic upgrade head
docker compose up -d api
```
Open `http://localhost:8000/docs`.
## Test and generate reliability evidence
```bash
docker compose run --rm api pytest -q
```
The crash-recovery test writes `artifacts/reliability-report.json`.
## Reliability claim
AgentFlow uses at-least-once tool execution. It demonstrates exactly one persisted business effect per stable idempotency key. A crash can cause the tool boundary to be invoked again, so it does not claim exactly-once execution.
## Documentation
- `docs/architecture.md`
- `docs/production-limitations.md`
- `artifacts/reliability-report.json`
## Cleanup
```bash
docker compose down
# Delete the PostgreSQL volume too:
docker compose down --volumes
```Before you run the complete suite, do you expect the interruption, concurrent-worker, and acknowledgment-gap tests to preserve the same guarantees when executed together?
- Run every PostgreSQL integration test and regenerate the evidence by running:
docker compose run --rm api pytest -qWhat should the full suite prove?
- The API test proves that a submitted support request reaches COMPLETED.
- The interruption test proves that CLASSIFIED survives outside worker memory.
- The concurrent queue test proves that competing workers complete all eligible workflows.
- The recovery test proves that multiple tool attempts retain one business effect and regenerate the report.
You should see the complete test suite pass. The refreshed report should contain a completion rate of 1.0 and zero duplicate business effects.
You have moved from a durability claim to reproducible evidence. AgentFlow now measures recovery, exposes its semantics, and documents the boundary between repeated execution and one persisted effect.
Secret mission
Secret Mission: Win the Acknowledgment-Gap Race
Force an acknowledgment gap after the simulated tool commits. Then make two recovery workers race. Their competing outcomes will prove that multiple attempts still produce one business effect.
Clean Up Your Resources
Clean Up Your Resources
AgentFlow runs entirely on your computer through Docker Compose with no ongoing cost. Choose whether to keep its resources running, pause them for later, or delete them entirely.
Resources you used:
- The api container serving the FastAPI application.
- The db container running PostgreSQL 18.6.
- The db-data named PostgreSQL volume containing the durable workflow data.
- The artifacts/reliability-report.json evidence file containing the recovery measurements.
Keep everything running
No action is needed. Choose this option if you want to continue testing recovery behavior or extending AgentFlow.
- Your api container stays available at http://localhost:8000/docs.
- Your db container continues serving the AgentFlow database.
- Your db-data volume retains the durable workflow checkpoints.
- Your artifacts/reliability-report.json file remains available as reliability evidence.
Pause - I'll come back to this later
Shut down the AgentFlow containers to free local memory. The named volume preserves your PostgreSQL checkpoints.
- Pause the local stack by running this command from the AgentFlow repository in the terminal from earlier:
docker compose downWhat Does This Command Do?
This removes the api container and the db container. It also removes their Compose network.
The db-data volume remains available for your next AgentFlow run.
You've freed the runtime memory while preserving the durable checkpoint data in db-data.
Delete - I don't want to use this again
This option removes AgentFlow's runtime resources and generated report. Your source repository remains available.
Deleting db-data permanently erases the saved PostgreSQL data.
- Delete the containers and named volume by running this command from the AgentFlow repository in the terminal from earlier:
docker compose down --volumesWhat Gets Deleted?
This removes the api container and the db container. It also removes their Compose network.
The command removes the db-data volume with every saved workflow checkpoint and ledger row.
Your AgentFlow runtime now has a clean slate. The source files remain untouched.
- Delete the generated reliability evidence by running this command from the AgentFlow repository:
rm artifacts/reliability-report.jsonWhat Gets Removed?
This removes only artifacts/reliability-report.json. The rest of the repository stays available as source code and documentation.
That's the cleanup complete. AgentFlow's runtime resources and generated evidence are removed.
Nice Work!
Nice Work!
You made it! AgentFlow now preserves support-workflow progress across interruptions in PostgreSQL. Your recovery tests also prove that a committed tool effect survives a missing acknowledgment.
You've learned how to:
- Build a FastAPI workflow service whose durable checkpoints preserve every deterministic decision across interruptions.
- Coordinate competing workers through SKIP LOCKED claims protected by leases plus fencing tokens.
- Generate Pytest integration evidence for at-least-once execution with exactly one business effect per stable idempotency key.
- Secret Mission: Recovered the acknowledgment-gap race with two competing workers. Verified that only one recovery future reports completed work. Confirmed that repeated attempts preserve one business effect.
Ready to quiz yourself?