Metadata-Version: 2.4
Name: tenet-langchain
Version: 0.3.0
Summary: Tenet PII governance integration for LangChain agents
Project-URL: Homepage, https://github.com/tenetlabsdev/tenet
Project-URL: Repository, https://github.com/tenetlabsdev/tenet
Project-URL: Documentation, https://github.com/tenetlabsdev/tenet/tree/main/packages/clients/python/tenet-langchain
Project-URL: Issues, https://github.com/tenetlabsdev/tenet/issues
License-Expression: Apache-2.0
License-File: LICENSE
Keywords: compliance,governance,langchain,pii,redaction
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Security
Requires-Python: <3.14,>=3.11
Requires-Dist: langchain>=1.2.13
Requires-Dist: tenet-client>=0.3.0
Provides-Extra: dev
Requires-Dist: pytest-asyncio>=0.21; extra == 'dev'
Requires-Dist: pytest>=7.0; extra == 'dev'
Requires-Dist: respx>=0.21; extra == 'dev'
Requires-Dist: ruff>=0.1.0; extra == 'dev'
Description-Content-Type: text/markdown

# tenet-langchain

LangChain integration for the [Tenet](https://tenetlabs.com) compliance SDK.
Built on top of the framework-agnostic [`tenet-client`](../tenet-client)
package — install this one if you're building a LangChain agent; install
`tenet-client` directly if you're not.

```bash
pip install tenet-langchain
```

## You own the agent lifecycle

Tenet evaluates and returns a *verdict*. Your application *policy*
decides what to do with it — allow, review, block, override the call,
fail-open, hand off to a human reviewer. Modeled loosely on Galileo
Protect's stage/action split: the centralized judge produces the verdict,
your local policy maps it to a runtime action.

That means you keep calling `langchain.agents.create_agent` yourself.
Tenet ships as a middleware you drop into the `middleware=[...]` list,
not a façade that owns agent creation.

## Three control levels

| Level | Surface | When to use |
|---|---|---|
| **1. Raw client** | [`TenetCloudJudgeClient`](../tenet-client) + [`evaluate_phase`](../tenet-client/src/tenet_client/cloud_client.py) | Custom agents (no LangChain), FastAPI middleware, queue workers, anywhere you want full control. |
| **2. LangChain middleware** | [`TenetJudgeMiddleware`](src/tenet_langchain/tenet_judge_middleware.py) + [`JudgePolicy`](../tenet-client/src/tenet_client/policy.py) | The primary path. Drop into a `create_agent(...)` middleware list; policy callbacks decide every outcome. |
| **3. Convenience helper** | [`wrap()`](src/tenet_langchain/wrap.py) (deprecated) | Preserved for alpha integrations. Owns the agent lifecycle for you, which removes control over middleware order and policy dispatch. New code should use level 2. |

The local-server path ([`TenetMiddleware`](src/tenet_langchain/middleware.py))
remains unchanged for self-hosted Tenet desktop / on-prem deployments.

## Primary path — `create_agent` + `TenetJudgeMiddleware` + `JudgePolicy`

```python
from langchain.agents import create_agent
from tenet_client import (
    JudgeAction,
    JudgePolicy,
    TenetCloudJudgeClient,
)
from tenet_langchain import TenetJudgeMiddleware

judge = TenetCloudJudgeClient.from_env()

policy = JudgePolicy(
    on_block=lambda ctx, verdict: JudgeAction.tool_error(
        verdict.safe_message or "Blocked by Tenet."
    ),
    on_review=lambda ctx, verdict: JudgeAction.allow(metadata={"tenet_review": True}),
    on_unavailable=lambda ctx, error: JudgeAction.raise_error(error),
    on_timeout=lambda ctx, error: JudgeAction.raise_error(error),
)

agent = create_agent(
    model="claude-sonnet-4-5-20250514",
    tools=[],
    middleware=[
        TenetJudgeMiddleware(judge, policy=policy),
    ],
)
```

`JudgePolicy()` with no callbacks gives you the safe defaults — block →
`tool_error`, review → soft-allow with metadata, unavailable / timeout →
re-raise (fail-closed). Override only what you want to change.

### `from_env()` environment variables

`TenetCloudJudgeClient.from_env()` needs two variables — the credentials
Tenet provisions for your account:

- `TENET_CLIENT_ID`
- `TENET_CLIENT_SECRET`

Both are required; `from_env()` raises `ValueError` naming any that are
missing. Everything else (cloud URL, Auth0 audience, Auth0 issuer)
points at Tenet-operated infrastructure and uses sensible production
defaults — you should not need to set them.

Optional tuning knobs, if you want them: `TENET_JUDGE_ID` (defaults to
`hr-1`; set `healthcare-1` for the PHI / clinical-safety judge) and
`TENET_JUDGE_TIMEOUT_SECONDS` (defaults to `3.0`).

If you'd rather pass credentials explicitly, call `TenetCloudJudgeClient(...)`
directly with the same keyword arguments.

### Per-tool and per-phase policies

Stricter rules for high-risk tools, softer rules for low-risk ones:

```python
from tenet_client import JudgeAction, JudgePolicy
from tenet_langchain import TenetJudgeMiddleware

policy = JudgePolicy(
    per_tool={
        "send_email": JudgePolicy(
            on_block=lambda ctx, v: JudgeAction.tool_error("Cannot send PHI."),
        ),
        "search": JudgePolicy(
            on_block=lambda ctx, v: JudgeAction.allow(),
        ),
    },
    per_phase={
        "tool_post": JudgePolicy(
            on_block=lambda ctx, v: JudgeAction.override_output(
                "Sorry, I can't share that."
            ),
        ),
    },
)
```

Per-tool wins over per-phase wins over the top-level callbacks.

### Recipes

**Block hard.** Default. Returns a `ToolMessage(status="error")` to the agent.

```python
from tenet_client import JudgeAction, JudgePolicy

policy = JudgePolicy(
    on_block=lambda ctx, v: JudgeAction.tool_error(v.safe_message or "blocked"),
)
```

**Review only — never block, just tag for audit.**

```python
from tenet_client import JudgeAction, JudgePolicy

policy = JudgePolicy(
    on_block=lambda ctx, v: JudgeAction.allow(metadata={"flagged": True}),
    on_review=lambda ctx, v: JudgeAction.allow(metadata={"flagged": True}),
)
```

**Fail-open dev mode.** When the cloud is flaky in dev.

```python
from tenet_client import JudgeAction, JudgePolicy

dev_policy = JudgePolicy(
    on_unavailable=lambda ctx, err: JudgeAction.allow(),
    on_timeout=lambda ctx, err: JudgeAction.allow(),
)
```

**Human review handoff.**

```python
from tenet_client import JudgeAction, JudgePolicy

def queue_for_review(ctx, verdict):
    queue.push({
        "session_id": ctx.session_id,
        "tool_name": ctx.tool_name,
        "verdict": verdict.model_dump(),
    })
    return JudgeAction.tool_error("Waiting on human review.")

policy = JudgePolicy(on_block=queue_for_review)
```

**Human-in-the-loop review prompt.**

Use this when a `review` verdict should pause the LangChain agent and ask
the end user whether to proceed. This uses LangGraph interrupts under the
hood, so the agent needs a checkpointer and each run needs a stable
`thread_id`. You do not need to build a custom LangGraph graph; LangChain
agents created with `create_agent(...)` are already graph-backed.

```python
from typing import Any

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command, interrupt

from tenet_client import (
    JudgeAction,
    JudgeContext,
    JudgePolicy,
    JudgeVerdict,
    TenetCloudJudgeClient,
)
from tenet_langchain import TenetJudgeMiddleware


def ask_user_on_review(ctx: JudgeContext, verdict: JudgeVerdict) -> JudgeAction:
    answer: Any = interrupt(
        {
            "kind": "tenet_review",
            "question": "Tenet recommends human review. Proceed with this tool call?",
            "tool_name": ctx.tool_name,
            "tool_input": ctx.tool_input,
            "reason": verdict.reason,
            "choices": [
                {"id": "proceed", "label": "Yes, proceed"},
                {"id": "reject", "label": "No, choose another action"},
            ],
        }
    )

    if isinstance(answer, dict) and answer.get("proceed") is True:
        return JudgeAction.allow(metadata={"tenet_review_approved": True})

    if isinstance(answer, dict):
        message = answer.get("message")
    else:
        message = None

    return JudgeAction.tool_error(
        message or "The user rejected this tool call after Tenet review."
    )


judge = TenetCloudJudgeClient.from_env()
policy = JudgePolicy(on_review=ask_user_on_review)

agent = create_agent(
    model="claude-sonnet-4-5-20250514",
    tools=[send_claim_to_payer],
    middleware=[TenetJudgeMiddleware(judge, policy=policy)],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "case-123"}}

result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "Submit this prior-auth packet to the payer.",
            }
        ]
    },
    config=config,
    version="v2",
)

# If Tenet returns review, expose result.interrupts[0].value to your UI.
# The user chooses yes/no, then you resume with the same thread_id.
if result.interrupts:
    user_selected_yes = True

    if user_selected_yes:
        resume_payload = {"proceed": True}
    else:
        resume_payload = {
            "proceed": False,
            "message": "Do not submit yet. Ask me for the missing CPT code first.",
        }

    agent.invoke(
        Command(resume=resume_payload),
        config=config,
        version="v2",
    )
```

Keep the `on_review` callback idempotent. LangGraph restarts the node
that called `interrupt(...)` when the run resumes, so any side effects
before the interrupt can happen more than once.

**Safe override response — replace the tool output instead of erroring.**

```python
from tenet_client import JudgeAction, JudgePolicy

policy = JudgePolicy(
    on_block=lambda ctx, v: JudgeAction.override_output(
        "I can't share that information. Please contact support."
    ),
)
```

**Retry nudge — let the agent self-correct.**

A `retry_nudge` action returns a `ToolMessage(status="error")` carrying
corrective guidance to the agent loop. The agent reads the guidance on
its next iteration and re-issues the tool call, typically with different
arguments. Distinct from `tool_error` (terminal denial) and from
`override_output` (replacement output). Forge research found this
mechanic produced large completion-accuracy lifts on identical model
weights, so it ships as a first-class `JudgeAction` kind.

```python
from tenet_client import JudgeAction, JudgePolicy

policy = JudgePolicy(
    on_block=lambda ctx, v: JudgeAction.retry_nudge_from_verdict(v),
)
```

The canonical constructor assembles the guidance from `verdict.reason`
and `verdict.safe_message` via a documented two-line template:

```
Cannot proceed as requested: {reason}
Suggested correction: {safe_message_or_reason}
```

Pass `template=...` to `retry_nudge_from_verdict` to override.

**Severity-routed recipe.**

For partner-configurable severity → action mapping, use the
`severity_routed_policy` recipe. Default mapping:

| Severity     | Action                              |
|--------------|-------------------------------------|
| `low`        | `retry_nudge_from_verdict`          |
| `medium`     | `retry_nudge_from_verdict`          |
| `high`       | `tool_error`                        |
| `critical`   | `tool_error`                        |
| `None`       | `tool_error` (preserves v1 default) |

Each bucket is keyword-configurable. `human_review_severities` is empty
by default and, when populated, takes precedence over `tool_error`:

```python
from tenet_client.recipes import severity_routed_policy

policy = severity_routed_policy(
    retry_nudge_severities=("low", "medium"),
    tool_error_severities=("high", "critical"),
    human_review_severities=(),
)
```

Pass through any other `JudgePolicy` kwargs (`on_unresolved`,
`on_unavailable`, `per_tool`, ...) via the same call.

**`on_unresolved` — workflow correction, not a compliance outcome.**

A tool can return a valid but empty result — "no records found", empty
list, null. The middleware detects these heuristically (see below) and
routes the case to `JudgePolicy.on_unresolved` independently of the
verdict's `allow|review|block`. The default callback returns
`retry_nudge_from_verdict(v)` so the agent gets a workflow correction
prompt; override to taste:

```python
from tenet_client import JudgeAction, JudgePolicy

policy = JudgePolicy(
    on_unresolved=lambda ctx, v: JudgeAction.retry_nudge(
        "No matches. Broaden the query or try a synonym."
    ),
)
```

### Resolution-status detection

After every tool call, `TenetJudgeMiddleware` classifies the return as
`resolved` / `unresolved` / (`error` — reserved for a future wave) and
threads the value to the judge so server-side rules can branch on it.
The default heuristics (case-insensitive substring match on the
stringified output):

- empty list `[]`, empty dict `{}`, empty string, or literal `"None"` → `unresolved`
- `"not found"`, `"no results"`, `"no match"`, `"0 results"` → `unresolved`
- anything else → `resolved`

Opt out per tool or globally on the middleware constructor:

```python
mw = TenetJudgeMiddleware(
    judge,
    policy=policy,
    resolution_detection=True,                              # default
    resolution_detection_excluded_tools={"send_email"},     # skip these tools
)
```

When opted out, the middleware threads `resolution_status=None`, which
preserves the pre-Phase-4.5 wire envelope and dispatch byte-for-byte.

### Retry-nudge budget + telemetry

To prevent runaway loops where the agent keeps re-issuing the same
broken call, the middleware tracks consecutive retry-nudges per tool
and escalates to `tool_error` once the budget is hit:

```python
mw = TenetJudgeMiddleware(
    judge,
    policy=policy,
    retry_nudge_max_consecutive=3,   # default; escalate on the 4th
)
```

Three counters live on `mw.telemetry`, each keyed by `(surface, tool)`:

- `mw.telemetry.emitted` — every retry-nudge synthesized
- `mw.telemetry.succeeded` — every chain that self-corrected (a non-nudge
  tool call followed a nudge on the same tool)
- `mw.telemetry.exhausted` — every chain that hit the budget and was
  escalated to `tool_error`

Read these directly to wire into whatever observability stack you have;
the middleware ships no external dependency.

### Verdict-first streaming

The judge exposes an SSE surface (`POST /eeo/decide/stream`) that emits the
verdict ~4× sooner than the full response. Opt in with `stream=True` and run
the agent with an **async** invocation (`ainvoke` / `astream`) — the sync
hooks have no streaming variant and ignore the flag.

```python
mw = TenetJudgeMiddleware(judge, policy=policy, stream=True)
```

The middleware branches on the stream's `route` event:

- `act_at_verdict` (route_to_human): the policy is dispatched the instant the
  `verdict` event lands — before the safe-message / explanation / rewrite body
  streams in — so a routing decision is available at ~TTV. The body is still
  drained and folded onto the verdict context for the HITL payload.
- `wait_for_done` (include / exclude / block): the rewrite or explanation *is*
  the actionable payload, so the stream drains fully and the policy is
  dispatched once on the complete verdict.

Behaviour and return types are otherwise identical to the non-streaming path —
it is a drop-in, A/B-able latency optimization. (For the raw client, the
equivalent toggles are `use_stream=True`, plus `evaluate_stream(...)` for a
verdict-first event iterator.)

### Multi-turn conversation history

The judge is stateless per call. On multi-turn BFOQ flows, sending only the
latest turn can read as discriminatory out of context. Set `send_history=True`
to send the **prior recruiter turns** (every prior `HumanMessage`, oldest
first) as additional `{"role": "user"}` messages ahead of the latest turn:

```python
mw = TenetJudgeMiddleware(judge, policy=policy, send_history=True)
```

Assistant turns are **never** sent — they carry no recruiter intent. Off by
default, in which case the wire envelope is byte-for-byte the single-turn shape.

### Langfuse session grouping

Every judge call carries a conversation `session_id` so the server groups a
conversation's traces into one [Langfuse](https://langfuse.com) session. The
middleware uses your LangGraph `thread_id` when a checkpointer is configured
(so each conversation groups on its own), falling back to the middleware's
`session_id` constructor arg otherwise — the same conversation identity the
verdict cache and turn-approval features scope on. No configuration required;
to pin a fixed id for an out-of-graph deployment, pass
`TenetJudgeMiddleware(judge, session_id="…")`.

### Support correlation — `X-Tenet-Canonical-Id`

Every judge call is tagged with a 32-hex `canonical_id` — the same key the
server emits to Langfuse (trace id), Metronome (transaction id), and OTel
(`tenet.canonical_id`). It is returned on the `X-Tenet-Canonical-Id` response
header and the `request_accepted` / `done` stream events, surfaced on
`CloudJudgeResponse.canonical_id`, and folded onto the verdict context as
`ctx.metadata["tenet_canonical_id"]`.

**Runbook:** log the `canonical_id` on every judge call so support can grep the
same key across Langfuse and Metronome when triaging a decision. The middleware
emits it at `DEBUG` (`logging.getLogger("tenet_langchain.tenet_judge_middleware")`).
**Never** show it to the end user — it is an internal correlation key, not a
user-facing reference.

## Level 1 — raw client, no LangChain

For custom agents, FastAPI middleware, queue workers — anywhere you
don't want LangChain in the call path:

```python
from tenet_client import (
    JudgeAction,
    JudgePolicy,
    TenetCloudJudgeClient,
)

judge = TenetCloudJudgeClient.from_env()
policy = JudgePolicy()

verdict, action = judge.evaluate_phase(
    policy=policy,
    phase="tool_pre",
    tool_name="search",
    tool_input={"q": "patient lookup"},
)

if action.kind == "allow":
    result = run_search()
elif action.kind == "tool_error":
    result = action.message
elif action.kind == "raise_error":
    raise action.error
```

`evaluate_phase_async` is the async sibling.

## Level 1 — `@judged` for standalone tools

```python
from tenet_client import (
    JudgeAction,
    JudgePolicy,
    JudgeBlocked,
    TenetCloudJudgeClient,
    judged,
)

judge = TenetCloudJudgeClient.from_env()
policy = JudgePolicy(
    on_block=lambda ctx, v: JudgeAction.override_output(
        "I cannot share that data."
    ),
)

@judged(judge, policy=policy)
def get_patient_chart(patient_id: str) -> str:
    return _fetch_chart(patient_id)

# When the judge blocks, the override_output replaces the return value.
chart = get_patient_chart("MRN12345")
```

## Cold-start avoidance

The first `evaluate()` call from a fresh `TenetCloudJudgeClient` pays a
TLS handshake to Auth0, the Auth0 token mint, and a TLS handshake to
the cloud judge — typically 600–900 ms vs. ~50–150 ms once warm. Call
`warmup()` (or `warmup_async()`) at process start:

```python
from tenet_client import TenetCloudJudgeClient

judge = TenetCloudJudgeClient.from_env()
judge.warmup()
```

`warmup()` is idempotent and doubles as a credential smoke test.

## Local server — `TenetMiddleware`

If you're running the Tenet desktop app or an on-prem server (default
`http://127.0.0.1:19990`) instead of the cloud judge:

```python
from langchain.agents import create_agent
from tenet_langchain import TenetMiddleware

agent = create_agent(
    model="claude-sonnet-4-5-20250514",
    tools=[],
    middleware=[TenetMiddleware(tenet_ids=["hipaa-safe-harbor"])],
)
```

Right surface for self-hosted deployments where data must not leave
your network.

## Migration from `wrap()`

`wrap()` still works but emits a `DeprecationWarning`. The translation
is mechanical:

```python
# Before
from tenet_langchain import wrap

agent = wrap(
    model=model,
    tools=tools,
    judge=judge,
    fail_open=False,
)

# After
from langchain.agents import create_agent
from tenet_client import JudgePolicy
from tenet_langchain import TenetJudgeMiddleware

agent = create_agent(
    model=model,
    tools=tools,
    middleware=[TenetJudgeMiddleware(judge, policy=JudgePolicy())],
)
```

The default `JudgePolicy()` is fail-closed — same as the old
`fail_open=False`. To restore the old `fail_open=True` behavior, set
`on_unavailable=lambda ctx, err: JudgeAction.allow()`.

## Examples

- [`examples/healthcare_agent/`](examples/healthcare_agent/) — end-to-end
  LangChain healthcare agent gated by the `healthcare-1` cloud judge.
- [`examples/human_review_interrupt.py`](examples/human_review_interrupt.py) —
  policy-level `review` verdict handling with a LangGraph interrupt and
  yes/no resume payload.

## License

Apache-2.0.
