midchain-governance 0.1.0

Selective re-verification of safety/compliance constraints across multi-step AI processes

midchain-governance

A small library for re-verifying safety/compliance constraints across a multi-step AI process — not just at the entry point.

The problem this addresses

Most guardrail setups check a request once, at the start. But in a multi-step AI pipeline (agent chains, model failover, tool-calling sequences), a violation can be introduced after that check — a later step's output, not the original prompt, is what actually reaches the user. An entry-only check can't see that.

What this is — and isn't

Is: a small, tested implementation of interval-based re-verification, with a specific guarantee: whatever output is about to be delivered is always checked, regardless of how many steps ran or were planned.

Isn't: a novel idea. Re-checking constraints across multi-step AI processes is an active area — see docs/CITATIONS.md for directly related published work (compliance gating, constraint drift across agent delegation, context-compaction-driven constraint decay). This library doesn't propose a new mechanism class; it's a specific, tested, reusable implementation of an established idea, plus a simulation exploring the catch-rate/latency tradeoff between three concrete policies.

See docs/PAPER_DRAFT.md for the full writeup, including honest limitations, and experiments/RESULTS.md for what the simulation actually found (and didn't).

Why use this

Three concrete advantages over what most guardrail setups do by default:

1. Most guardrail setups only check the user's prompt. If an agent calls several tools in sequence, or a request fails over between models, whatever comes back from step 2 or the second model attempt often never gets checked — only the final output does, if that. This library closes that gap deliberately, by design, not as an afterthought.
2. "Check everything" is expensive; "check once" is unsafe. The selective policy gives you a middle setting that, in simulation under the assumptions in experiments/RESULTS.md, recovers most of full re-checking's catch-rate benefit at roughly half the latency cost — so you don't have to choose between safe and fast for every request. Re-measure on your own guardrails before treating those numbers as production forecasts.
3. The delivery guarantee is unconditional. Whatever actually reaches the end user gets checked, regardless of which step or which failover candidate produced it — even in a process where you don't know in advance how many steps will run. This is the guarantee with a real bug fix and regression test behind it (see CHANGELOG.md), not just a design intention.

Install

pip install -e .

Or copy src/midchain_governance/ into your project. Optional: pip install -e ".[dev]" for pytest.

Quick start

from midchain_governance import GovernanceGate, run_sequential_pipeline

gate = GovernanceGate(interval=2, always_check_final=True)

def my_compliance_check(output: str) -> bool:
    # wire this to your real moderation/guardrail API
    return "SECRET" not in output

steps = [lambda: "step 1", lambda: "step 2", lambda: "step 3"]
result = run_sequential_pipeline(steps, check_fn=my_compliance_check, gate=gate)

For failover / retry logic (candidates are alternatives, not a mandatory sequence — the first passing one wins):

from midchain_governance import run_failover_chain

candidates = [lambda: call_model_a(), lambda: call_model_b()]
result = run_failover_chain(candidates, check_fn=my_compliance_check)

See examples/basic_usage.py for a runnable version of both.

Real-time example: LLM provider failover with compliance checking

A realistic setup — try OpenAI, fail over to Anthropic if the response is blocked or the provider errors. Uses litellm only as an illustrative client (optional; not a package dependency) plus a moderation-style check:

import litellm
from midchain_governance import GovernanceGate, run_failover_chain

def call_openai():
    resp = litellm.completion(
        model="gpt-4o",
        messages=[{"role": "user", "content": user_prompt}],
    )
    return resp.choices[0].message.content

def call_anthropic():
    resp = litellm.completion(
        model="claude-sonnet-4-6",
        messages=[{"role": "user", "content": user_prompt}],
    )
    return resp.choices[0].message.content

def real_compliance_check(output: str) -> bool:
    # Swap this for whatever you actually use: a moderation endpoint,
    # a PII/regex scanner, a policy-tuned classifier, or an
    # LLM-as-judge prompt. This example uses litellm's moderation call.
    result = litellm.moderation(input=output)
    return not result.results[0].flagged

gate = GovernanceGate(interval=2, always_check_final=True)

response = run_failover_chain(
    candidate_fns=[call_openai, call_anthropic],
    check_fn=real_compliance_check,
    gate=gate,
)

If OpenAI's response is flagged, this automatically tries Anthropic next — the request doesn't abort, and whichever response actually gets returned has been checked, guaranteed, regardless of which provider produced it.

Real-time example: fixed multi-step agent pipeline

For a pipeline where every request walks the same steps (e.g. fetch data → analyze → draft a response), each with its own agent/model call:

from midchain_governance import GovernanceGate, run_sequential_pipeline

gate = GovernanceGate(interval=2, always_check_final=True)

steps = [
    lambda: fetch_agent.run(query),
    lambda: analyze_agent.run(fetched_data),
    lambda: writer_agent.run(analysis),
]

result = run_sequential_pipeline(steps, check_fn=real_compliance_check, gate=gate)

With interval=2, this checks after step 1 (entry), step 2, and always step 3 (final) — so a violation introduced by the analyze step gets caught before the writer agent ever sees it, not just after the final draft is produced.

Three checking policies

Config	Behavior
GovernanceGate(interval=<very large>, always_check_final=False)	Entry-only ("once")
GovernanceGate(interval=1)	Every step ("every_hop")
GovernanceGate(interval=K, always_check_final=True)	Selective: entry, every Kth step, and always the delivered output

Why "selective" over "every step"

In simulation (see experiments/), interval-based selective checking recovered most of full re-checking's catch-rate benefit at roughly half the latency cost. Full numbers, methodology, and — importantly — the assumptions those numbers depend on, are in experiments/RESULTS.md. These are simulation results, not production measurements. Don't cite the specific percentages as what you'll see on real traffic without re-measuring against your own guardrail's real detection rate.

A real bug this caught (kept as a regression test)

Early in development, the "final check" guarantee used the planned number of steps to decide when to force a check. In a variable-length process like failover — where most requests only make one attempt — that meant an early-succeeding step could skip the check entirely. Fixed by checking whatever's about to be delivered directly (before_delivery=True), independent of how many steps were planned. See tests/test_gate.py::test_regression_early_success_still_gets_final_check.

Structure

src/midchain_governance/   the library
tests/                     outcome-level tests (not just mechanism tests)
experiments/                the simulation, its results, and honest limitations
examples/                  runnable usage examples
docs/                      citations, limitations, full paper draft

Running tests

python3 -m unittest discover tests

License

MIT — see LICENSE.

Contributing

See CONTRIBUTING.md.