Metadata-Version: 2.5
Name: 3tears-langgraph
Version: 0.30.1
Summary: Three-tier LangGraph checkpoint saver: L1 SQLite -> L2 NATS KV -> L3 PostgreSQL
Project-URL: Repository, https://github.com/pacepace/3tears
Author: pace
License-Expression: MIT
License-File: LICENSE
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Database
Classifier: Topic :: Software Development :: Libraries
Classifier: Typing :: Typed
Requires-Python: >=3.14
Requires-Dist: 3tears-media-contracts<0.31.0,>=0.30.0
Requires-Dist: 3tears-observe<0.31.0,>=0.30.0
Requires-Dist: 3tears<0.31.0,>=0.30.0
Requires-Dist: asyncpg
Requires-Dist: langchain-core>=1.0
Requires-Dist: langchain>=1.0
Requires-Dist: langgraph-checkpoint>=4.0
Requires-Dist: langgraph>=1.0
Requires-Dist: uuid-utils
Description-Content-Type: text/markdown

# 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

```bash
pip install 3tears-langgraph
```

## Usage

```python
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:

```python
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.

## Middleware

The package ships platform-level [`AgentMiddleware`](https://docs.langchain.com/oss/python/langchain/middleware) 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.

```python
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 `Protocol`s 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`](../../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.

```python
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"]`.
