Extending checkpoint schema with user_id/agent_id for cross-thread querying — trade-offs vs Thread Registry?

[santosh-nallur]

Hi Team,

I want to gather opinion on one of our requirements related to LangGraph. I found a similar discussion here where this is discussed at length and provides useful insights, but I have more specific questions.

Use cases:

1. Loading all the previous conversations of user(s) using checkpoints
2. Re-hydrating an older conversation upon selection
   (You can imagine 1 & 2 as ChatGPT-style app)
3. HITL & Pause-Resume, and any other typical usecases for checkpoints.

The core problem:

For these use-cases, we need identifiers beyond the standard checkpoint keys (thread_id, checkpoint_ns, checkpoint_id). Specifically, to support use case #1 — list all conversations for a user — we need to query across threads, which requires a user/agent identifier since there is no cross-thread index otherwise.

Option 1 — Thread Registry:

A separate thread registry (analogous to LangGraph Platform’s Thread entity) where each thread carries metadata such as user_id, agent_id, etc. This cleanly separates concerns: the checkpointer owns state snapshots, the registry owns identity and lifecycle metadata.

Trade-off: introduces a two-step flow (create thread → then configure checkpoint with the returned thread_id), adds infrastructure complexity (especially for non-platform deployment without an agent server or deployment infra taking care of this requirement), and requires keeping thread and checkpoint lifecycles in sync.

Option 2 — Extend the checkpoint table:

Rather than managing a separate thread registry, store the required values directly in the checkpoints. Essentially enhancing the table design to have additional columns:

thread_id TEXT NOT NULL,
checkpoint_ns TEXT NOT NULL DEFAULT ‘’,
checkpoint_id TEXT NOT NULL, – ULID, lexicographically sortable newest-last
parent_checkpoint_id TEXT,
type TEXT,
checkpoint BYTEA,
metadata JSONB, – LangGraph internal use (source, step, writes)
expires_at TIMESTAMPTZ, – optional: TTL for automatic expiry
user_id TEXT,
agent_id TEXT,
PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id)

The agent would supply a config as follows:

config = {
    "configurable": {
        "thread_id": "<thread_id>",
        "agent_id": "my-agent",
        "user_id": "my-user"
    }
}

Note: the existing metadata column is for LangGraph’s internal use (source, step, writes). The proposed user_id/agent_id columns are application-level identifiers.

Context — A2A agents:

Most of our agents are behind an A2A server (Google’s A2A protocol). For each request A2A generates a context_id; agents typically use context_id as thread_id or append more details to the context_id itself. Storing user_id and agent_id alongside checkpoints is therefore beneficial for queries such as — list conversations for user, get conversation details.

Option 2 — Pros:

1. Simpler from the consumption side — no need to make two calls (1. create thread first, then 2. configure checkpoint with the returned thread_id). Get started with configuring checkpoints straight away.
2. The agent can generate its own thread_id or use context_id from A2A. Storing user_id/agent_id with checkpoints enables queries such as — list conversations for user, get conversation details.
3. TTL lives with the checkpoint itself. No separate per-thread TTL or lifecycle sync required.

Option 2 — Cons:

1. Multiple snapshots are taken during agent execution; all snapshots share the same user_id/agent_id, introducing duplication (no normalization).
2. Limited filtering capabilities — only two fixed identifiers. This can be mitigated by introducing an agent_metadata JSONB column to support additional key-value pairs for future filtering needs (keeping the existing metadata column for LangGraph’s internal use).
3. Potential performance issues due to filtering on non-PK columns across large tables — can be mitigated by indexing user_id and agent_id.

Questions:

1. Has anyone implemented Option 2 or a variant in production? What are the query performance implications of filtering on non-PK columns across large checkpoint tables?
2. Is there a cleaner way to support cross-thread user/agent queries without Option 1’s two-step flow ? or would you say option-1 is the best way forward
3. Any thoughts on agent_metadata JSONB vs fixed columns — which approach has held up best in practice?

Thanks!

[keenborder786]

Hello @santosh-nallur
Short answer: Use a thin thread registry instead of putting user_id and agent_id on every checkpoint row.

Checkpoints are versioned snapshots, one per step. If you store user and agent IDs on each row, you duplicate data and “list my conversations” becomes an expensive scan over a large table, even with indexes.

LangGraph’s metadata column is meant for run internals like step, writes, and source. Using it for app-level identity mixes concerns and still doesn’t give you one clean row per conversation.

You can avoid a clunky two-step flow. Generate thread_id yourself (UUID or A2A context_id) and pass it in config["configurable"] on the first invoke. On that first message, upsert one registry row with thread_id, user_id, agent_id, title, updated_at, and optional expires_at. From the agent’s side that’s basically one step; the registry is just a small indexed table.

For queries: list conversations from the registry (WHERE user_id = ?). To resume, take thread_id from the registry and use normal checkpoint APIs like get_state or invoke with the same config. Handle TTL at the thread level and delete checkpoints by thread_id.

For schema design, use fixed columns like user_id and agent_id for filters you’ll run often. Use JSONB for optional or evolving fields, and add a GIN index only if you actually query inside it.

Bottom line: keep the checkpointer focused on state and the registry on identity and lifecycle. A minimal registry is the pattern that holds up in production. With A2A, mapping context_id to thread_id plus registry metadata is a common setup.

I hope this helps!!!

[santosh-nallur]

Hello @keenborder786 ,

Thanks, this is very helpful.

Regarding : “Generate thread_id yourself (UUID or A2A context_id) and pass it in config["configurable"] on the first invoke. On that first message, upsert one registry row with thread_id, user_id, agent_id, title, updated_at, and optional expires_at. From the agent’s side that’s basically one step; the registry is just a small indexed table.”

I could think of a following a hybrid pattern that keeps client usage simple while preserving a forward path to explicit thread APIs if required in future. However, I’m not sure what do you mean by "Generate thread_id yourself (UUID or A2A context_id) and pass it in config["configurable"] on the first invoke. On that first message" **How do we determine the first invoke / first message in the checkpoint library ? **

1. Keep checkpointing focused on state snapshots and pending writes.
2. Add a minimal thread registry (one row per thread) for conversation discovery and filtering.
3. Internally call a single upsert operation on successful checkpoint writes (both snapshot and writes paths), so clients still make one logical call with configurable thread_id, user_id, and agent_id.
4. Use upsert semantics for lifecycle:
5. Insert when thread_id is new (created_at plus identity fields).
6. Update when thread_id exists (refresh updated_at and expires_at).
7. Keep first-write-wins for identity fields (i.e. no update on identified fields like agent_id, user_id).
8. Use registry for list_conversations and thread-level filters; use checkpoint APIs for resume, history, and state hydration.

Background:

Our checkpointer is not based on backend store, we have an OData API that internally uses a store.

Sample implementation:

class MyCustomSaver(BaseCheckpointSaver[str]):
    def _upsert_thread_registry(self, cfg):
        body = {
            "threadId": cfg["thread_id"],
            "agentId": cfg.get("agent_id", ""),
            "invokerId": cfg.get("invoker_id", "") or cfg.get("user_id", ""),
            "updatedAt": utc_now(),
            "expiresAt": self._expires_at(),  # None if TTL disabled
        }
        self._sync_retry.post(self._thread_registry_upsert_url(), json=body).raise_for_status()

    async def _aupsert_thread_registry(self, cfg):
        body = {
            "threadId": cfg["thread_id"],
            "agentId": cfg.get("agent_id", ""),
            "invokerId": cfg.get("invoker_id", "") or cfg.get("user_id", ""),
            "updatedAt": utc_now(),
            "expiresAt": self._expires_at(),
        }
        r = await self._retry.post(self._thread_registry_upsert_url(), json=body)
        r.raise_for_status()

    def put(...):
        self._sync_retry.post(self._snapshots_url(), json=snapshot_body).raise_for_status()
        self._upsert_thread_registry(cfg)    # additional call for upsert

    def put_writes(...):
        self._sync_retry.post(self._upsert_writes_url(), json=writes_body).raise_for_status()
        self._upsert_thread_registry(cfg)   # additional call for upsert

    async def aput(...):
        (await self._retry.post(self._snapshots_url(), json=snapshot_body)).raise_for_status()
        await self._aupsert_thread_registry(cfg)  # additional call for upsert

    async def aput_writes(...):
        (await self._retry.post(self._upsert_writes_url(), json=writes_body)).raise_for_status()
        await self._aupsert_thread_registry(cfg) # additional call for upsert

list is not impacted, but delete_thread would remove the thread + corresponding checkpoints. This however, has a down side of two calls for put and put_writes - 1. checkpoint and 2. upsert thread

Client/ Agent usage remains as follows:

# 1) Build saver and compile graph once
checkpointer = MyCustomSaver(
    base_url="https://<service-host>",
    token_url="https://<token-url>",
    client_id="<client-id>",
    client_secret="<client-secret>",
    ttl_seconds=3600,
)
graph = builder.compile(checkpointer=checkpointer)

# 2) One-call client config + invoke
config = {
    "configurable": {
        "thread_id": "ctx-001",          # required
        "agent_id": "chat-agent-v2",    # optional
        "user_id": "user-123456",    # optional (preferred key today)
        # "user_id": "user-i123456",      # if you add alias support
    }
}

result = graph.invoke(
    {"messages": [{"role": "user", "content": "Plan my trip to Paris"}]},
    config=config,
)

Thread registry table would contain fields:
thread_id - unique client supplied id such as context_id from A2A)
agent_id,user_id, description,
metadata - additional JSON fields for filtering ,
created_at, updated_at, expires_at

Future compatibility:
Similar to langgraph platform we can later expose explicit create_thread with title/description and metadata, and pass that thread_id into checkpointer config . This is additive and backward compatible because the same thread registry model and upsert/store semantics remain in place.

What do you think ?

[keenborder786]

There is no “first invoke” in LangGraph. A new thread_id means a new conversation; reusing it means continuation. Your registry upsert decides that with insert vs update.

Your hybrid design is the right approach. Upsert the registry from put / put_writes so clients keep a single call with thread_id, user_id, and agent_id. Use first-write-wins for identity fields and refresh updated_at / expires_at on later writes.

The extra registry call per write is fine. If it becomes noisy, upsert only on put (snapshots), not every put_writes.

Read user_id / agent_id from config["configurable"], not checkpoint metadata. A future create_thread API fits cleanly on top of the same registry. I’d ship this.