3tears-langgraph

3tears-langgraph

Three-tier LangGraph checkpoint saver: L1 (SQLite) -> L2 (NATS KV) -> L3 (PostgreSQL).

L1 and L2 are optional cache layers that degrade gracefully on failure. L3 (PostgreSQL) is the source of truth, reached through the AsyncQueryExecutor protocol so the same saver serves trusted services (direct asyncpg pool) and sandboxed agents (NATS L3 proxy).

L3 failures reach the caller, with one channel-scoped carve-out: aput_writes is called from LangGraph's executor teardown, where raising kills a turn that has already answered, so a failed crash-recovery write is logged and degraded. Writes on LangGraph's control channels (__interrupt__, __resume__, __error__, __scheduled__) still raise: losing one changes what the run does, and a lost __interrupt__ would silently skip a human-approval gate. A control-channel write also overwrites an earlier write on the same channel, matching the reference saver, so a second Command(resume=...) replaces the first rather than being dropped.

Installation

pip install 3tears-langgraph

Usage

from threetears.langgraph import (
    AsyncpgPoolAdapter,
    CheckpointScope,
    ThreeTierCheckpointSaver,
)

# Trusted service with direct asyncpg.Pool: wrap once
saver = ThreeTierCheckpointSaver(
    executor=AsyncpgPoolAdapter(pool),
    scope=CheckpointScope.for_customer(customer_id),
)

# Sandboxed agent: NatsProxyL3Backend already implements
# AsyncQueryExecutor, pass it straight through
saver = ThreeTierCheckpointSaver(
    executor=nats_l3_backend,
    scope=CheckpointScope.for_customer(customer_id),
)

graph = builder.compile(checkpointer=saver)

Scope is required

scope has no default. A saver either names the customer whose checkpoints it addresses, or says in writing that it deliberately names none:

saver = ThreeTierCheckpointSaver(
    executor=AsyncpgPoolAdapter(pool),
    scope=CheckpointScope.unscoped(reason="single-tenant deployment"),
)

CheckpointScope.for_customer(...) folds the customer into the stored thread_id, and therefore into the L3 bound parameter, the L2 bucket key, and the L1 thread key — a saver scoped to one customer cannot name another customer's row at any tier. It also unlocks adelete_customer_threads(), the whole-tenant purge, which refuses on an unscoped saver.

CheckpointScope.unscoped(...) is a legitimate answer, not a placeholder: it produces byte-identical keys and statements to a pre-tenancy saver, so an existing deployment adopts the required parameter by adding this one argument and migrating no data. The reason is mandatory, logged at WARNING on construction, and greppable in source, so "which deployments still run unscoped, and why" has an answer.

Adopting a real customer later is a data change rather than a code change: existing rows live under a bare thread id and a scoped saver will not find them, so they must be re-keyed (UPDATE checkpoints SET thread_id = $customer || '/' || thread_id, likewise checkpoint_writes, plus L2 invalidation). No re-key script ships here and none can — which customer owns which thread lives in the host's own tables, which this library has never seen.

Person erasure: anonymize, never delete

result = await saver.aanonymize_threads(conversation_ids)

Rewrites every stored checkpoint and pending write of those threads in place: a human message's name (the sender's display name) and every value under IDENTIFYING_METADATA_KEYS (external_user_name, external_user_id) become ANONYMIZED_MARKER. Message text, every id, and the blob format are unchanged, so the graph still loads and resumes. Idempotent, paged by batch_size, and it evicts the threads' L1 and shared L2 bundles, raising if an eviction fails. An unknown metadata key is kept: the metadata channel is working state the graph reads on resume. Run it on threads that are not in a live turn.

Middleware

The package ships platform-level AgentMiddleware for langchain.agents.create_agent — the framework-aligned successor to the old hand-rolled AgentNodeHook / ToolNodeHook protocols. Consumer-specific policy lives in each consumer as its own middleware; only the reusable platform seams live here:

• PromptCachingMiddleware (wrap_model_call) — annotates a leading bare-string system message with Anthropic cache_control={"type": "ephemeral"} when the model supports it, then normalizes cache-hit/creation counters onto usage_metadata["cache_usage"]. Non-Anthropic adapters degrade silently to bare-string system messages.
• ToolResultOffloadMiddleware (wrap_tool_call) — when a ToolResultOffloader is injected on config["configurable"] and a tool result exceeds offload_threshold_chars, stores the full content out-of-band and shows the model "<summary>\n\n[ctx:<handle>]" (the structured artifact is preserved). Opt-in: no offloader ⇒ byte-for-byte no-op.
• ObjectCatalogMiddleware (wrap_tool_call) — when a tool returns an ObjectHandle in its result artifact and an ObjectCataloger is injected, persists a catalog record under the verified call identity. Soft-fail side-effect: a catalog error never breaks the tool result.
• Two summarizers. Choose by whether the conversation's history must survive:
  • SummarizationMiddleware (before_model) — once the message COUNT passes trigger_messages, rewrites the checkpointed messages channel to [summary, *last keep_messages]. The older messages are gone from the checkpoint.
  • RollingSummaryMiddleware (awrap_model_call, async only) — NON-destructive. Give it store (built per turn) or store_for(request) (an agent compiled once). The checkpointer keeps every message, and only the model's input is trimmed, once the messages since the last fold pass a TOKEN budget. The rolling summary and a cursor (the last folded message's id) live behind a SummaryStore, whose save is a compare-and-swap. It folds at most once per turn, never stores a failed or empty summary over the prior one, never leaves an orphaned tool result at the head of the kept tail, and marks its summary call NOSTREAM_TAG plus metadata["threetears.usage.purpose"] = "summarization". threetears.conversations.ConversationSummaryStore is the store over a conversations row, and dispatch_conversation_summarized is its on_summarized.

from langchain.agents import create_agent

from threetears.langgraph import (
    ObjectCatalogMiddleware,
    PromptCachingMiddleware,
    ToolResultOffloadMiddleware,
)

agent = create_agent(
    model=chat_anthropic,
    tools=tools,
    middleware=[
        PromptCachingMiddleware(),
        ToolResultOffloadMiddleware(),
        ObjectCatalogMiddleware(),
    ],
)
# after a run, PromptCachingMiddleware has stamped:
# message.usage_metadata["cache_usage"]
# == {"cache_read_input_tokens": ..., "cache_creation_input_tokens": ..., "cached_tokens": ...}

The offload / catalog contracts (ToolResultOffloader, ObjectCataloger) are pure structural Protocols exported from threetears.langgraph.offload / threetears.langgraph.catalog; a consumer injects a concrete implementation on config["configurable"] (e.g. tool_result_offloader, object_cataloger) without the package taking any dependency on the consumer's context store.

See 3tears/docs/prompt-caching.md for the full caching contract, summarization interaction, downstream wiring checklist, and a worked example.

Streaming

The package ships StreamingResponse, a transport-agnostic primitive that owns the lifecycle of one streaming response: start -> any number of emit_token / emit_tool_call_* -> mutually-exclusive end (success) or error (failure) terminal. run_graph(compiled_graph, state, config) consumes a LangGraph astream_events(version="v2") loop with the start/end ordering managed; on graph exception it fires error(code="AGENT_FAILED", ...) and re-raises so the caller still sees the failure on the synchronous path.

The wire vocabulary is fixed: StreamStartEvent / StreamTokenEvent / StreamEndEvent / StreamErrorEvent / ToolCallStartEvent / ToolCallEndEvent / ToolCallProgressEvent, dispatched via the StreamEvent discriminated union and the parse_stream_event(payload) adapter. The transport seam is the StreamTransport Protocol -- one method, async def publish(self, payload: bytes) -> None. Any wire (NATS subject, websocket, chunked HTTP body) satisfies it.

from threetears.langgraph import StreamingResponse, StreamTransport

class WebSocketStreamTransport:
    """example transport for a websocket consumer."""
    def __init__(self, ws): self._ws = ws
    async def publish(self, payload: bytes) -> None:
        await self._ws.send_bytes(payload)

stream = StreamingResponse(
    transport=WebSocketStreamTransport(ws),
    correlation_id=correlation_id,
    conversation_id=conversation_id,
    start_time_monotonic=request_start,
)
final_state = await stream.run_graph(compiled_graph, state, config)

A reference adapter can bind the primitive to a per-correlation-id stream subject via nc.publish_raw. Tool-call observation envelopes flow through ToolCallProgressHook reading the active StreamingResponse from config["configurable"]["streaming_response"].