Checkpointing & Durable Execution

Checkpointing enables agents to persist their state and resume execution after interruptions, process restarts, or failures.

Overview

Durable execution provides:

State persistence - Save execution state at checkpoints
Resume capability - Continue from last checkpoint
Failure recovery - Survive process crashes and restarts
Long-running tasks - Handle tasks that span hours or days

Checkpointer Backends

Backend	Use Case	Persistence	Performance
`session_service`	Production (recommended)	Durable	High
`postgres`	Production (traditional)	Durable	High
`memory`	Development/testing	None	Fastest

Basic Usage

Enable Durable Execution


from isa_agent_sdk import query, ISAAgentOptions, ExecutionMode
 
options = ISAAgentOptions(
    execution_mode=ExecutionMode.COLLABORATIVE,
    checkpoint_frequency=5  # Checkpoint every 5 steps
)
 
async for msg in query("Long running task...", options=options):
    print(msg.content, end="" if msg.is_text else "\n")

Session Management


from isa_agent_sdk import query, ISAAgentOptions
 
# Start a new session with explicit ID
options = ISAAgentOptions(
    session_id="my-task-123",
    user_id="user-456"
)
 
async for msg in query("Start processing files...", options=options):
    if msg.is_checkpoint:
        print(f"Checkpoint saved: {msg.session_id}")
    elif msg.is_text:
        print(msg.content, end="")

Resume Execution


from isa_agent_sdk import resume
 
# Resume from a previous session
async for msg in resume(
    session_id="my-task-123",
    resume_value={"continue": True}  # Optional: provide input
):
    print(msg.content, end="" if msg.is_text else "\n")

Synchronous Resume


from isa_agent_sdk import resume_sync
 
result = resume_sync(
    session_id="my-task-123",
    resume_value={"authorized": True}
)
print(result.content)

Checkpointer Configuration

Using Session Service (Recommended)


# Set via environment variable
# CHECKPOINTER_BACKEND=session_service
 
# Or configure in options
options = ISAAgentOptions(
    execution_mode=ExecutionMode.COLLABORATIVE
)

Using PostgreSQL Directly


# Set via environment variable
# CHECKPOINTER_BACKEND=postgres
# DATABASE_URL=postgresql://user:pass@host:5432/db

Using Memory (Development)


# Set via environment variable
# CHECKPOINTER_BACKEND=memory

Programmatic Access

Get the Checkpointer


from isa_agent_sdk.services.persistence import (
    get_checkpointer,
    get_durable_service
)
 
# Get the durable service
durable = get_durable_service()
 
# Get the checkpointer instance
checkpointer = durable.get_checkpointer()

Initialize Async Pool (FastAPI)


from isa_agent_sdk.services.persistence import initialize_async_pool
from contextlib import asynccontextmanager
 
@asynccontextmanager
async def lifespan(app):
    # Initialize the connection pool at startup
    await initialize_async_pool()
    yield
    # Cleanup handled automatically

Session Service Checkpointer

The SessionServiceCheckpointer integrates with the isA platform’s session microservice:


from isa_agent_sdk.services.persistence import SessionServiceCheckpointer
 
# Automatically used when backend is "session_service"
# Requires session_service_url in config

Features:

Thread-safe async operations
Automatic serialization
Distributed session management
Built-in retry logic

Execution Manager

For advanced control over execution state:


from isa_agent_sdk.services.persistence import ExecutionManager
 
manager = ExecutionManager()
 
# Check execution status
status = await manager.get_status("my-task-123")
print(f"Status: {status}")
 
# List pending executions
pending = await manager.list_pending(user_id="user-456")
for execution in pending:
    print(f"- {execution.session_id}: {execution.status}")

Checkpoint Events in Streaming


async for msg in query("Process all files", options=options):
    if msg.is_checkpoint:
        # Execution paused at checkpoint
        checkpoint_data = msg.metadata
        print(f"Checkpoint: {checkpoint_data}")
 
        # Respond to continue
        await msg.respond({"continue": True})
 
    elif msg.type == "progress":
        print(f"Progress: {msg.progress_percent}%")
 
    elif msg.is_text:
        print(msg.content, end="")

Failure Recovery

Automatic Retry

The SDK automatically handles transient failures:


options = ISAAgentOptions(
    execution_mode=ExecutionMode.COLLABORATIVE,
    max_iterations=30  # Prevents infinite loops
)

Manual Recovery


from isa_agent_sdk import get_session_state, resume
 
# Check session state
state = await get_session_state("my-task-123")
 
if state and state.get("status") == "interrupted":
    # Resume from last checkpoint
    async for msg in resume("my-task-123"):
        print(msg.content)

Integration with HIL

Checkpointing works seamlessly with Human-in-the-Loop:


from isa_agent_sdk import query, request_tool_permission, ExecutionMode
 
options = ISAAgentOptions(
    execution_mode=ExecutionMode.COLLABORATIVE
)
 
async for msg in query("Delete old files", options=options):
    if msg.is_hil_request:
        # Execution paused, waiting for human input
        # State is saved - can resume even after restart
 
        tool = msg.metadata.get("tool_name")
        print(f"Permission requested for: {tool}")
 
        # User approves (could be hours later)
        await msg.respond({"authorized": True})
 
    elif msg.is_text:
        print(msg.content, end="")

Checkpoint Compaction

For sessions that run for hours or days, checkpoint history can grow large. CheckpointCompactor prunes old checkpoints while preserving any checkpoints you’ve explicitly tagged.


from isa_agent_sdk.services.persistence import CheckpointCompactor
 
compactor = CheckpointCompactor(
    session_id="long-job-xyz",
    max_checkpoints=50,             # Keep at most 50 untagged checkpoints
    compaction_interval_mins=60,    # Run compaction every 60 minutes
)
 
await compactor.start()

Tagging Checkpoints

Tag a checkpoint to exclude it from compaction:


async for msg in query("Process files...", options=options):
    if msg.is_checkpoint and msg.metadata.get("milestone"):
        # Tag this checkpoint so it's never pruned
        await msg.tag(label="milestone-complete")

Tagged checkpoints are preserved regardless of max_checkpoints. Use tags for meaningful milestones (e.g., “phase 1 complete”, “all files indexed”).

Compaction Reference

Parameter	Default	Description
`max_checkpoints`	`50`	Maximum untagged checkpoints to retain
`compaction_interval_mins`	`60`	How often to run the compaction pass

Session State Garbage Collection

The GarbageCollector removes expired checkpoints, stale Redis keys, and old task results on a schedule.


from isa_agent_sdk.services.persistence import GarbageCollector
 
gc = GarbageCollector(
    checkpoint_ttl_days=30,     # Delete checkpoints older than 30 days
    redis_key_ttl_hours=48,     # Evict stale Redis session keys after 48h
    task_result_ttl_days=7,     # Remove task result records after 7 days
)
 
# Dry run — see what would be deleted without deleting
report = await gc.dry_run()
print(f"Would delete: {report.checkpoint_count} checkpoints, {report.redis_key_count} Redis keys")
 
# Run actual GC
await gc.run()

GC Configuration Reference

Parameter	Default	Description
`checkpoint_ttl_days`	`30`	Checkpoints older than this are deleted
`redis_key_ttl_hours`	`48`	Stale Redis session keys evicted after this
`task_result_ttl_days`	`7`	Task result records older than this are deleted

Run GC as a scheduled background task:


import asyncio
 
async def schedule_gc():
    gc = GarbageCollector()
    while True:
        await gc.run()
        await asyncio.sleep(24 * 3600)  # Daily

Warm Restart

When an agent resumes from a checkpoint, it normally re-fetches all MCP tool schemas from scratch — adding latency proportional to the number of registered MCP servers. Warm restart caches schemas in Redis so resume is near-instant.

How It Works

On first run: tool schemas are fetched from MCP servers and cached in Redis under mcp:schema:{session_id}
On resume: schemas are loaded from Redis cache, skipping MCP server round-trips
Cache is invalidated when: an MCP server reconnects, a schema version changes, or the session GC removes the key

Enabling Warm Restart

Warm restart is enabled by default when REDIS_URL is configured. To disable:


MCP_SCHEMA_CACHE_ENABLED=false

Cache Configuration

Variable	Default	Description
`MCP_SCHEMA_CACHE_ENABLED`	`true`	Enable/disable warm restart caching
`MCP_SCHEMA_CACHE_TTL_HOURS`	`24`	How long to keep cached schemas

Best Practices

Use explicit session IDs for tasks you might need to resume
Set appropriate checkpoint frequency based on task duration
Handle checkpoint events in streaming for visibility
Use COLLABORATIVE mode for long-running tasks
Configure production backend (session_service or postgres)
Tag milestone checkpoints to protect them from compaction
Run GC daily to prevent unbounded checkpoint accumulation
Enable warm restart (default) for fast resume on long-running sessions

Environment Variables

Variable	Description	Default
`CHECKPOINTER_BACKEND`	Backend type	`session_service`
`DATABASE_URL`	PostgreSQL connection	-
`SESSION_SERVICE_URL`	Session service URL	From config
`ENVIRONMENT`	Environment name	`dev`

Next Steps

Human-in-the-Loop - Combine with HIL
Options - Configure execution modes
Streaming - Handle checkpoint events