Skip to main content

Split governance

Status: Beta Companion docs: Inline model calls · Verdicts and modes · Chat and Proxy API Source of truth: Chat and Proxy API. Both SDK checks call POST /v1/govern.

Choose this when​

Choose split governance when you must keep your provider client or cannot route the provider request through the Trinitite model proxy. Your code checks the conversation before and after the provider call.

Python flow​

Pass the full conversation to the input check. If Trinitite returns a masked body, send that body to the provider.

from openai import OpenAI
from trinitite import Trinitite

tr = Trinitite(env="prod")
provider = OpenAI()

policy = "Allow support help. Block credential disclosure. Mask personal data."
conversation = [
    {"role": "system", "content": "You are a support assistant."},
    {"role": "user", "content": "Summarize my account notes."},
]

input_result = tr.govern.input(
    instructions=policy,
    input=conversation,
    guardian="trinitite-base",
    masking=["pii"],
)

if input_result.verdict == "blocked":
    raise RuntimeError(input_result.violations)

if not input_result.flow_id:
    raise RuntimeError("Input governance did not return a flow_id")

provider_messages = input_result.masked_body or conversation
provider_response = provider.chat.completions.create(
    model="gpt-4o",
    messages=provider_messages,
)

reply = provider_response.choices[0].message.content or ""
output_result = tr.govern.output(
    flow_id=input_result.flow_id,
    instructions=policy,
    input=provider_messages + [{"role": "assistant", "content": reply}],
    guardian="trinitite-base",
    upstream={
        "provider": "openai",
        "model": "gpt-4o",
        "usage": {
            "prompt_tokens": provider_response.usage.prompt_tokens,
            "completion_tokens": provider_response.usage.completion_tokens,
            "total_tokens": provider_response.usage.total_tokens,
        },
    },
)

if output_result.verdict == "blocked":
    raise RuntimeError(output_result.violations)

if output_result.flow_id:
    unmasked = output_result.unmask(reply)
    reply = unmasked.body["text"]

print(reply)

The output check must receive the conversation plus the provider reply as the final assistant turn. Pass upstream usage so the audit record can attribute provider and token usage.

Request lifecycle​

Result and failure behavior​

Both calls return GovernanceResult subclasses. Read verdict, violations, diff, receipt, and flow_id on each result.

  • Input and output verdicts normalize to passed, corrected, blocked, masked, HiTL, or failed.
  • input_result.masked_body is the provider-safe copy when input masking succeeds.
  • The output call evaluates only. It does not rehydrate the provider response.
  • A production client rejects force_verdict before sending a request.
  • HTTP failures remain visible through result.status, result.ok, and result.body.
  • A transport error that exhausts retries raises ControlPlaneError.

If the input result has no flow_id, there is no masking flow to pass into the required output call. Stop and handle that result instead of inventing an id.

Identity, masking, and receipts​

The same flow_id binds input masking, output masking, and later unmasking. Forward masked_body, not the original conversation, whenever it is present.

The optional context argument can carry correlation context on both checks. Keep the same context on both sides of the provider call:

context = {"correlation_id": "support-4821"}

Use the same context=context argument in tr.govern.input(...) and tr.govern.output(...).

Each check can return its own receipt block. Verify a receipt only when its id is present:

for result in (input_result, output_result):
    if result.receipt and result.receipt.get("id"):
        assert result.receipt.verify().ok

Next steps​

Last updated on