feat: add HTTP-proxy LangGraph checkpoint API

Agents no longer need a direct Postgres connection for LangGraph
checkpointing. Instead, checkpoint operations are proxied through
5 new backend endpoints under /checkpoints (get-tuple, put,
put-writes, list, delete-thread). Binary blob data is base64-encoded
for JSON transport.

Includes ORM models for the 4 checkpoint tables, Alembic migration,
repository with composite-PK queries, use case layer, Pydantic
schemas, and FastAPI routes.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: danielmillerp

agentex-ui/hooks/use-task-messages.ts

@@ -116,19 +116,27 @@ export function useSendMessage({
             throw new Error(response.error.message);
           }
 
-          queryClient.setQueryData<TaskMessagesData>(queryKey, data => ({
-            messages: data?.messages || [],
-            deltaAccumulator: data?.deltaAccumulator || null,
-            rpcStatus: 'pending',
-          }));
+          // Refetch messages and spans now that the agent has finished processing
+          await queryClient.invalidateQueries({ queryKey: taskMessagesKeys.byTaskId(taskId) });
+          queryClient.invalidateQueries({ queryKey: ['spans', taskId] });
 
-          return (
-            queryClient.getQueryData<TaskMessagesData>(queryKey) || {
-              messages: [],
-              deltaAccumulator: null,
-              rpcStatus: 'pending',
-            }
-          );
+          const finalMessages = await agentexClient.messages.list({
+            task_id: taskId,
+          });
+
+          const chronologicalMessages = finalMessages.slice().reverse();
+
+          queryClient.setQueryData<TaskMessagesData>(queryKey, {
+            messages: chronologicalMessages,
+            deltaAccumulator: null,
+            rpcStatus: 'success',
+          });
+
+          return {
+            messages: chronologicalMessages,
+            deltaAccumulator: null,
+            rpcStatus: 'success',
+          };
         }
 
         case 'sync': {

agentex/database/migrations/alembic/versions/2026_02_11_0802_add_langgraph_checkpoint_tables_d1a6cde41b3f.py

@@ -0,0 +1,84 @@
+"""add_langgraph_checkpoint_tables
+
+Revision ID: d1a6cde41b3f
+Revises: d024851e790c
+Create Date: 2026-02-11 08:02:10.739927
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = 'd1a6cde41b3f'
+down_revision: Union[str, None] = 'd024851e790c'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # checkpoint_migrations
+    op.create_table('checkpoint_migrations',
+    sa.Column('v', sa.Integer(), nullable=False),
+    sa.PrimaryKeyConstraint('v')
+    )
+
+    # checkpoints
+    op.create_table('checkpoints',
+    sa.Column('thread_id', sa.Text(), nullable=False),
+    sa.Column('checkpoint_ns', sa.Text(), server_default='', nullable=False),
+    sa.Column('checkpoint_id', sa.Text(), nullable=False),
+    sa.Column('parent_checkpoint_id', sa.Text(), nullable=True),
+    sa.Column('type', sa.Text(), nullable=True),
+    sa.Column('checkpoint', postgresql.JSONB(astext_type=sa.Text()), nullable=False),
+    sa.Column('metadata', postgresql.JSONB(astext_type=sa.Text()), server_default='{}', nullable=False),
+    sa.PrimaryKeyConstraint('thread_id', 'checkpoint_ns', 'checkpoint_id')
+    )
+    op.create_index('checkpoints_thread_id_idx', 'checkpoints', ['thread_id'], unique=False)
+
+    # checkpoint_blobs
+    op.create_table('checkpoint_blobs',
+    sa.Column('thread_id', sa.Text(), nullable=False),
+    sa.Column('checkpoint_ns', sa.Text(), server_default='', nullable=False),
+    sa.Column('channel', sa.Text(), nullable=False),
+    sa.Column('version', sa.Text(), nullable=False),
+    sa.Column('type', sa.Text(), nullable=False),
+    sa.Column('blob', sa.LargeBinary(), nullable=True),
+    sa.PrimaryKeyConstraint('thread_id', 'checkpoint_ns', 'channel', 'version')
+    )
+    op.create_index('checkpoint_blobs_thread_id_idx', 'checkpoint_blobs', ['thread_id'], unique=False)
+
+    # checkpoint_writes
+    op.create_table('checkpoint_writes',
+    sa.Column('thread_id', sa.Text(), nullable=False),
+    sa.Column('checkpoint_ns', sa.Text(), server_default='', nullable=False),
+    sa.Column('checkpoint_id', sa.Text(), nullable=False),
+    sa.Column('task_id', sa.Text(), nullable=False),
+    sa.Column('idx', sa.Integer(), nullable=False),
+    sa.Column('channel', sa.Text(), nullable=False),
+    sa.Column('type', sa.Text(), nullable=True),
+    sa.Column('blob', sa.LargeBinary(), nullable=False),
+    sa.Column('task_path', sa.Text(), server_default='', nullable=False),
+    sa.PrimaryKeyConstraint('thread_id', 'checkpoint_ns', 'checkpoint_id', 'task_id', 'idx')
+    )
+    op.create_index('checkpoint_writes_thread_id_idx', 'checkpoint_writes', ['thread_id'], unique=False)
+
+    # Pre-populate checkpoint_migrations so LangGraph sees all its
+    # internal migrations as already applied and skips setup().
+    op.execute(
+        sa.text(
+            "INSERT INTO checkpoint_migrations (v) VALUES (0),(1),(2),(3),(4),(5),(6),(7),(8),(9)"
+        )
+    )
+
+
+def downgrade() -> None:
+    op.drop_index('checkpoint_writes_thread_id_idx', table_name='checkpoint_writes')
+    op.drop_table('checkpoint_writes')
+    op.drop_index('checkpoint_blobs_thread_id_idx', table_name='checkpoint_blobs')
+    op.drop_table('checkpoint_blobs')
+    op.drop_index('checkpoints_thread_id_idx', table_name='checkpoints')
+    op.drop_table('checkpoints')
+    op.drop_table('checkpoint_migrations')

agentex/database/migrations/migration_history.txt

@@ -1,4 +1,5 @@
-24429f13b8bd -> d024851e790c (head), add_performance_indexes
+d024851e790c -> d1a6cde41b3f (head), add_langgraph_checkpoint_tables
+24429f13b8bd -> d024851e790c, add_performance_indexes
 a5d67f2d7356 -> 24429f13b8bd, add agent input type
 329fbafa4ff9 -> a5d67f2d7356, add unhealthy status
 d7addd4229e8 -> 329fbafa4ff9, change_default_acp_to_async

agentex/src/adapters/orm.py

@@ -5,6 +5,8 @@
     DateTime,
     ForeignKey,
     Index,
+    Integer,
+    LargeBinary,
     String,
     Text,
     func,
@@ -213,3 +215,56 @@ class DeploymentHistoryORM(BaseORM):
             "commit_hash",
         ),
     )
+
+
+# LangGraph checkpoint tables
+# These mirror the schema from langgraph.checkpoint.postgres so that
+# tables are created via Alembic migrations rather than at agent runtime.
+
+
+class CheckpointMigrationORM(BaseORM):
+    __tablename__ = "checkpoint_migrations"
+    v = Column(Integer, primary_key=True)
+
+
+class CheckpointORM(BaseORM):
+    __tablename__ = "checkpoints"
+    thread_id = Column(Text, nullable=False, primary_key=True)
+    checkpoint_ns = Column(Text, nullable=False, primary_key=True, server_default="")
+    checkpoint_id = Column(Text, nullable=False, primary_key=True)
+    parent_checkpoint_id = Column(Text, nullable=True)
+    type = Column(Text, nullable=True)
+    checkpoint = Column(JSONB, nullable=False)
+    metadata_ = Column("metadata", JSONB, nullable=False, server_default="{}")
+    __table_args__ = (
+        Index("checkpoints_thread_id_idx", "thread_id"),
+    )
+
+
+class CheckpointBlobORM(BaseORM):
+    __tablename__ = "checkpoint_blobs"
+    thread_id = Column(Text, nullable=False, primary_key=True)
+    checkpoint_ns = Column(Text, nullable=False, primary_key=True, server_default="")
+    channel = Column(Text, nullable=False, primary_key=True)
+    version = Column(Text, nullable=False, primary_key=True)
+    type = Column(Text, nullable=False)
+    blob = Column(LargeBinary, nullable=True)
+    __table_args__ = (
+        Index("checkpoint_blobs_thread_id_idx", "thread_id"),
+    )
+
+
+class CheckpointWriteORM(BaseORM):
+    __tablename__ = "checkpoint_writes"
+    thread_id = Column(Text, nullable=False, primary_key=True)
+    checkpoint_ns = Column(Text, nullable=False, primary_key=True, server_default="")
+    checkpoint_id = Column(Text, nullable=False, primary_key=True)
+    task_id = Column(Text, nullable=False, primary_key=True)
+    idx = Column(Integer, nullable=False, primary_key=True)
+    channel = Column(Text, nullable=False)
+    type = Column(Text, nullable=True)
+    blob = Column(LargeBinary, nullable=False)
+    task_path = Column(Text, nullable=False, server_default="")
+    __table_args__ = (
+        Index("checkpoint_writes_thread_id_idx", "thread_id"),
+    )

agentex/src/api/app.py

@@ -19,6 +19,7 @@
     agent_api_keys,
     agent_task_tracker,
     agents,
+    checkpoints,
     deployment_history,
     events,
     messages,
@@ -183,6 +184,7 @@ async def handle_unexpected(request, exc):
 fastapi_app.include_router(agent_api_keys.router)
 fastapi_app.include_router(deployment_history.router)
 fastapi_app.include_router(schedules.router)
+fastapi_app.include_router(checkpoints.router)
 
 # Wrap FastAPI app with health check interceptor for sub-millisecond K8s probe responses.
 # This must be the outermost layer to bypass all middleware.