Metadata-Version: 2.5
Name: aictrl-langgraph
Version: 0.5.0
Summary: Graph and execution monitoring for LangGraph workflows
Project-URL: Repository, https://github.com/aictrl-azenohi/ai-control-plane-pro
Project-URL: Documentation, https://github.com/aictrl-azenohi/ai-control-plane-pro/tree/main/packages/aictrl-langgraph
Project-URL: Issues, https://github.com/aictrl-azenohi/ai-control-plane-pro/issues
License-Expression: Apache-2.0
License-File: LICENSE
Keywords: agents,langgraph,monitoring,observability
Classifier: Development Status :: 4 - Beta
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Typing :: Typed
Requires-Python: <3.15,>=3.11
Requires-Dist: httpx<1,>=0.28
Requires-Dist: langchain-core<2,>=1.6
Requires-Dist: langgraph<2,>=1.2
Requires-Dist: regex==2026.9.29
Description-Content-Type: text/markdown

# AI Control Plane · LangGraph SDK

Connect a compiled LangGraph `StateGraph` to graph and run monitoring with one
Python package. The SDK registers the topology, reports run and node lifecycle
changes, and keeps the application connected with a heartbeat.
Optional policy guards enforce assignments configured in the graph workspace.

Python 3.11–3.14. Supports LangGraph `>=1.2,<2`, LangChain Core `>=1.6,<2` and
HTTPX `>=0.28,<1`; CI tests the lowest and latest allowed versions. License: Apache-2.0.

## Install

```sh
uv add aictrl-langgraph        # or: pip install aictrl-langgraph
```

Python 3.11–3.14. Pin a compatible range in applications, for example
`aictrl-langgraph>=0.5,<0.6`, while the SDK is in `0.x`. Releases and notes:
https://github.com/aictrl-azenohi/ai-control-plane-pro/releases

## Install from a local build

From the repository root:

```sh
uv build --package aictrl-langgraph --out-dir dist/sdk
```

In the agent application's environment:

```sh
python -m pip install /path/to/dist/sdk/aictrl_langgraph-0.5.0-py3-none-any.whl
```

The wheel is standalone; it does not require the repository or the backend package.
Released versions are on PyPI (see Install above).

## Provision the application once

An administrator creates an application using the monitoring API. This helper is
included in the same package:

```python
import os
from aictrl_langgraph import register_application

credentials = register_application(
    base_url=os.environ["AICTRL_URL"],
    name="Support workflow",
    management_token=os.environ["MONITORING_ACCESS_TOKEN"],
)
# Store credentials.api_key in your secret manager as AICTRL_API_KEY.
# Save credentials.application_id for administration. Do not log the key.
```

Omit `management_token` only for the backend's explicit local development mode.
Do not provision on every process start. Registration is not retried because a
lost response can already have created an application. If this happens, inspect
the application list and rotate that application's credentials through the API.

Runtime monitoring needs `AICTRL_URL` and the application-scoped `AICTRL_API_KEY`.
Model-backed policies additionally need their provider configuration (see below).
`AICTRL_URL` is the server URL (for example `http://localhost:8100` locally), optionally
ending in `/api/v1`. Use HTTPS for remote connections. The SDK does not load `.env`
files. A management token is not an ingestion key.

## Integrate your graph

```python
from aictrl_langgraph import LangGraphMonitor

# builder is your application's existing StateGraph.
with LangGraphMonitor.from_env() as monitor:
    graph = monitor.instrument(builder.compile())
    result = graph.invoke(inputs)
    monitor.flush()
```

`instrument` returns a copy of the compiled graph with monitoring callbacks added.
Existing callbacks, configuration, checkpointing, runtime context and graph methods
remain available. Use the returned graph for `invoke`, `batch`, `stream`, `ainvoke`,
`abatch` and `astream`. Use one monitor per root graph and application connection;
create a new monitor for a new graph version. Instrument during startup.

Async applications can keep network registration and shutdown off the event loop:

```python
async with LangGraphMonitor.from_env() as monitor:
    graph = await monitor.ainstrument(builder.compile())
    result = await graph.ainvoke(inputs)
    async for chunk in graph.astream(inputs, stream_mode="updates"):
        consume(chunk)
    await monitor.aflush()
```

Stream modes and chunks are unchanged. Exhaust or explicitly close/aclose stream
iterators before closing the monitor. Closing a stream early or cancelling an
invocation records a failed run. Graph exceptions are preserved.

A runnable, LLM-free example is included in
[`examples/workflow.py`](examples/workflow.py). Run it only against an application
registered for testing; it creates real monitoring records.

## Run and graph semantics

- Each graph invocation gets a new monitoring run ID and starts at sequence 1.
  Parallel invocations are isolated. Parallel nodes and retry attempts share the
  invocation's ordered sequence.
- Interrupts finish that invocation as `PAUSED`. Resuming a checkpoint creates a
  **new monitoring run** for the resumed invocation. LangGraph's thread/checkpoint
  continuity remains intact; thread IDs and checkpoint contents are not sent.
- Branches and loops retain their node IDs. Give conditional edges an explicit
  `path_map` or a `Literal` return annotation so LangGraph can expose their topology.
  The SDK exports the public `get_graph(xray=False)` representation; it cannot
  discover undeclared runtime `Command` destinations. Declare node destinations
  when building such graphs.
- Nested subgraphs appear as their parent node on the static graph. With opt-in
  tracing, nested runnable, model and tool calls are separate child operations in
  the run inspector. They do not create extra monitoring runs or graph nodes.
  Cache hits that do not execute a node do not generate synthetic node events.
- Node IDs must match `[\w.:-]{1,100}`. Graphs allow at most 500 nodes and 2,000 edges.
  The immutable digest returned by the API is attached to every event.
- Monitoring callbacks observe execution without changing routing, prompts, retry
  policies or checkpoint state. Explicit policy guards can block node execution.

## Guard nodes with policies

Declare local evaluators with `Policy` and wrap each protected node with
`monitor.guard_node` **before** adding it to the StateGraph:

```python
from aictrl_langgraph import LangGraphMonitor, Policy, PolicyBlocked

policies = [
    Policy(
        "export_limit",
        "Export amount limit",
        "1",
        lambda state: state["amount"] <= 100,
        description="Allow exports up to 100 units.",
        operations=("data.export",),
    )
]
with LangGraphMonitor.from_env(policies=policies) as monitor:
    # builder, export_result and inputs belong to your application.
    builder.add_node(
        "export",
        monitor.guard_node(
            "export",
            export_result,
            operation="data.export",
        ),
    )
    graph = monitor.instrument(builder.compile())
    try:
        result = graph.invoke(inputs)
    except PolicyBlocked:
        handle_denial()
```

The catalog appears in **Policies**. Choose **Optimize**, assign a policy to a
diamond and **Apply**. Each guard fetches current assignments before calling the
node, so a saved change affects its next execution boundary. No assignment means
no evaluator runs, but the configuration read must still succeed. Bump the policy
version when its behavior changes; evaluator source code is never uploaded.

Exactly `True` allows execution; `ReviewRequired` pauses for human approval.
Denials and evaluator errors raise `PolicyBlocked` before the node function runs.
Unavailable configuration, incompatible active versions and review-service errors
raise its subclass `PolicyUnavailable`, allowing applications to distinguish a
connection/configuration problem from a policy refusal. Callbacks report both as
`BLOCKED`. Keep evaluators
free of side effects and do not mutate their input. Async node functions support
async evaluators; sync nodes reject an async evaluator. Wrappers preserve node
signatures and injected config/runtime context.

Guard only with ordinary sync/async node functions. Unwrapped nodes and side
effects already in progress are outside enforcement; parallel operations are not
rolled back. Do not cache protected nodes: cache hits skip their guards. A
checkpoint resume checks nodes that execute again. All versions using one
application key must have compatible graph/catalog metadata; incompatible active
bindings block until a compatible configuration is applied.

## Delivery and shutdown

```python
monitor = LangGraphMonitor(
    base_url="https://control.example.com",
    api_key=application_key,
    timeout=5.0,
    max_retries=2,
    queue_capacity=10_000,
    heartbeat_interval=10.0,
    max_active_runs=1_000,
)
```

Callbacks enqueue metadata and, when explicitly enabled, bounded captured content.
Serialization runs inline; network delivery runs in one background worker. FIFO
batches contain at most 100 events and split below the 1 MiB API request limit. Transient network errors and HTTP
408/429/500/502/503/504 are retried with the **same event IDs, timestamps and body**.
Other errors, including authentication failures and sequence conflicts, are not
retried. Redirects are not followed. Request timeouts and retry counts are bounded.

If delivery retries are exhausted or the queue fills, telemetry for that monitor
stops and `monitor.error` records the failure. It does not silently skip an event
and continue with an invalid sequence. The application keeps running, a safe
warning is logged, and `flush()` / `close()` raise `MonitoringError`. Opt-in guards
still require a successful configuration read before execution; an unavailable
connection denies the operation even if telemetry can no longer report it.
Telemetry failure leaves the HTTP client available for fresh policy reads and
human-review decisions once the service returns. Only explicit monitor closure
closes that client. Create a new monitor to restore telemetry after the underlying
problem is corrected. The affected history can remain incomplete; completion is
never fabricated.

`flush(timeout=30)` waits for queued events to be acknowledged. A flush timeout
does not discard events. `close(timeout=30)` stops new telemetry and drains the
queue; use it after all graph invocations and streams finish. `aflush` and `aclose`
are the asynchronous equivalents. Context managers close automatically and do
not replace an exception already raised by the application with a delivery error.

Delivery is in-memory and best effort. Process crashes or forced termination can
lose unsent events. There is no disk spool, cross-process queue or recovery of
in-flight invocations after a process restart. Create monitors after worker fork;
monitor instances belong to one process.

## Data collected

By default, only graph node IDs/labels, topology, opaque monitoring run/event IDs, graph digest,
lifecycle types, sequence numbers and UTC timestamps are sent. With policy guards,
policy IDs/names/descriptions/versions, operation restrictions and hook metadata are
also sent, and the SDK reads assignment IDs. Inputs, outputs,
state values, messages, prompts, tool arguments, exception text, checkpoint data
and callback metadata are excluded from lifecycle-only monitoring. Opt-in tracing
changes this contract as described below. Model-backed
policy evaluators explicitly send guarded input to their configured model endpoint.
Custom Python sends it through the authenticated control-plane execution API to
the isolated runner. HITL decision memory separately persists bounded reviewed
context in the control plane and compares it using the backend Ollama. Node names
and edge labels are visible
in the UI, so use structural labels rather than user content. Credentials and
server response bodies are excluded from SDK diagnostics.

## Detailed run tracing (0.5)

Upgrade the backend before enabling this extension; older backends reject operation
events. Existing integrations remain lifecycle-only unless `trace` is supplied:

```python
from aictrl_langgraph import LangGraphMonitor, TraceConfig

with LangGraphMonitor.from_env(trace=TraceConfig(capture_content=True)) as monitor:
    graph = monitor.instrument(builder.compile())
    result = graph.invoke(inputs)
```

`TraceConfig()` records operation names, timing, parent relationships, outcomes and
reported token usage but omits content. `capture_content=True` additionally records
node inputs/state updates, model messages and invocation parameters, retrieval
queries/documents, tool arguments/results, run metadata/tags and exception type/message. No raw
HTTP headers, process environment or arbitrary object representations are collected.
The inspector distinguishes disabled, captured, redacted, truncated and missing values.

For a custom HTTP/model/MCP client, wrap the actual call inside an instrumented node:

```python
with monitor.operation(
    "tool", "lookup_order", arguments, attributes={"transport": "mcp"}
) as operation:
    result = await client.call_tool("lookup_order", arguments)
    operation.output(result)
```

Use `kind="model"` for model clients, `kind="retriever"` for RAG and `kind="chain"`
for internal checks or application sub-operations. Supply the final
messages and tool definitions as inputs/attributes at the actual request boundary.
For a model response, `operation.output(result, total_tokens=reported_tokens)` can
report measured usage. Never estimate missing usage as zero. Manual scopes nest
and remain isolated across concurrent async tasks. Outside an instrumented graph,
or with tracing disabled, the scope executes normally without producing telemetry.
Do not wrap a call already observed by LangChain unless the extra span is intentional.

With tracing enabled, `guard_node` records each evaluated policy as a child chain
operation with its policy ID/version, binding version and evaluation digest.
Content capture includes the evaluated input and allow/deny/require-review result
after decision memory. Requiring review is not human approval; the review service
still controls release. Evaluator exceptions are recorded as failures, and policies
skipped after a denial are not fabricated. A manual scope retains an explicitly
recorded partial output when it later raises, alongside its error.

Capture defaults to 32 KiB per field (configurable from 256 bytes to 64 KiB), depth
16 and 500 entries per collection. Strings, JSON values, LangChain messages and
retrieval documents are supported. Cycles, unsupported objects and exhausted limits
are marked partial; no `repr()` fallback is used. Known credential keys and token
patterns, plus the monitoring credential, are filtered before queuing. This is not
an exhaustive PII/secret detector. `TraceConfig(redact=my_string_redactor, ...)`
adds application-specific filtering; a failed redactor omits the affected string.
Use structural operation names. Readers with workspace monitoring access can read
captured content; there is no separate field-level entitlement or automatic retention
policy. Configure capture and retention for the application's data.

Tracing observes calls made through callbacks/manual scopes, not hidden provider
internals or unrelated network requests. Node state is distinct from model context:
inspect a model operation's input for the messages actually passed to that model.
Provider transformations after the callback require a manual client adapter. Cached
model results are not new dispatched calls. Streaming captures the final assembled
response, not one telemetry event per token. Resumed invocations retain the existing
separate-run behavior. Only callbacks actually received are reported as completed;
open operations become incomplete when the run terminates. A process crash can leave
a run running and lose unsent events because delivery remains in memory.

## Development

```sh
uv run pytest packages/aictrl-langgraph/tests services/backend/tests/test_langgraph_sdk.py
make verify
make sdk-check
```

`make sdk-check` builds the wheel and source distribution, checks their metadata,
installs the wheel into an isolated environment and executes a real LangGraph run
outside the repository. CI performs isolated installation checks on Python
3.11, 3.12, 3.13 and 3.14. These checks do not upload or publish anything.

## Human-in-the-loop policies

Return `ReviewRequired` from a policy evaluator to require an editor's decision:

```python
from aictrl_langgraph import Policy, ReviewRequired

review = Policy(
    id="pii_review",
    name="Approve personal data release",
    version="1",
    operations=("chat.respond",),
    check=lambda state: ReviewRequired(state["request_id"]) if state["sensitive"] else True,
)
```

Declare this policy on the monitor, guard the response node, compile with a
LangGraph checkpointer, and instrument as usual. Assign it in the control-plane
policy editor. `request_id` must be an opaque unique identifier retained in the
checkpoint; never put user content in it. The guarded input must be JSON-compatible.
The SDK binds the review to the entire input using a keyed HMAC, scoped to the
application, node, policy and active graph/catalog/binding versions.

On pause, LangGraph's `__interrupt__` result contains a safe payload:
`{"type": "action_required", "review_ids": [...]}`. Forward that status to the
user while keeping the response on the backend. Poll `monitor.review_status(id)`
or `await monitor.areview_status(id)`: results are `pending`, `approved`, `rejected`
or `expired`. When all required reviews are approved, resume the same checkpoint
with `Command(resume=True)`. The guard fetches the service decision again and never
trusts the resume value. Reject/expiry/read failures block; modified input cannot
reuse the request identity. A block from any assigned evaluator wins over review.

The **Reviews** panel uses policy-editor access for decisions. Approvals expire
after ten minutes, and graph/catalog/binding changes invalidate them. Request and
policy metadata, keyed input digests and timestamps are stored separately from
lifecycle events. With a backend advertising HITL decision memory, the SDK also
sends bounded reviewed input (excluding the top-level request ID) as memory context.
Human decisions automatically become examples for this exact policy use, node and
application. Resume still creates a new monitoring run. The application owns checkpoint retention,
request-to-conversation correlation, delivery idempotency and disconnect handling.
See the [SQLite chat example](../../apps/sqlite_chat/README.md) for a complete flow.

### Centrally authored rules (0.3)

The policy editor and control catalogue can publish data-only executable rules.
Version 0.3 fetches those definitions at every guarded node boundary and evaluates
them locally with the same compiler/scanner used by editor previews. Local callback
policies remain supported. Any hard denial wins before human-review requests are
created. Missing/incompatible configuration and unknown required scan results fail
closed. Removing an assignment or disabling a baseline takes effect at the next
boundary; configuration changes invalidate previous approvals.

Authored `require_approval` rules require an opaque, unique-per-request `request_id`
in the JSON-compatible node state and a checkpointer. Do not reuse a conversation ID
for separate user requests. Resume values never grant permission.

PII scans cover the whole guarded input (including conversation history), only
supported formats, and not future node output. Place the hook **after** generating
an answer and **before** releasing it, as in SQLite chat. Typed message objects
need JSON-compatible state at the guarded boundary; unsupported objects make a
required scan unavailable. See [authoring and limitations](../../docs/development.md#runtime-contract-and-limitations).


### Three-step YAML policy wizard (0.4)

SDK 0.4 additionally evaluates `aictrl.policy/v2` documents, published globally or
assigned from policy groups. New `guard_node(..., scope="tool")` metadata supports
agent (default), tool, MCP and RAG scopes. Always supply an explicit operation.
Global policies apply only to matching scope/operations; node assignments must
match the hook. Mandatory applied policies cannot be removed through assignments.

Predefined deterministic policies execute local field/text/range checks, PII scanning
or bounded regex. Custom Python is sent with the input to the control plane’s
authenticated execution endpoint and run in a disposable isolated container. No
Docker access is needed in the agent process. Exceptions/non-boolean results/timeouts
deny. Python can use arbitrary algorithms and installed libraries; the deployment
owns dependencies, and sandbox jobs cannot access the network or host filesystem. Probabilistic
policies send input to Jev (`JEV_API_KEY`, optional `JEV_BASE_URL`); Full AI sends
input to Ollama (`POLICY_OLLAMA_URL`, default `http://127.0.0.1:11434`). Configure
these in the **agent process** as well as the backend that runs wizard tests.
Model name and criteria come from the saved policy; credentials/endpoints never do.
Unknown, malformed or unavailable evaluations deny. A pending/denied/expired review
cannot be bypassed by a changed model answer on replay. Mandatory Human in the loop
still allows an authorized human approval bound to the exact input/configuration.

Upgrade all affected SDK applications before activating v2 policies. Older clients
fail closed when a selected policy requires v2. SDK 0.5 also includes automatic HITL
memory, ordered node policies and opt-in detailed run tracing (below).

### Automatic HITL memory (0.5)

Authored Human in the loop policies automatically perform an additional validation
against prior human decisions for the same application, graph node and policy use.
Empty memory keeps the original policy result; uncertain memory requires a human.
No `MemoryPoint` or route activation is needed. Context is stored with reviews and
compared using the backend Ollama for similar cases. See
[decision memory](../../docs/decision-memory.md) for scope, data handling and limits.
