diff --git a/README.md b/README.md
index 77ac775..6dcfd84 100644
--- a/README.md
+++ b/README.md
@@ -1,7 +1,30 @@
-
- AgentAssert
- Formal Behavioral Contracts for AI Agents
-
+# AgentAssert — runtime behavioral contracts for AI agents
+
+Define rules in YAML and check structured agent state in Python. A hard
+violation can raise `ContractBreachError` before your application continues.
+Your application supplies the signals and places the check at the action boundary.
+
+**[Install and docs](https://agentassert.com/getting-started)** · **[Working proof](docs/QUICK_PROOF.md)** · **[Research](https://agentassert.com/research)**
+
+```bash
+python -m pip install "agentassert-abc[yaml,math]"
+```
+
+Requires Python 3.12+. Clone this repository to run the self-contained example:
+
+```bash
+python examples/00_quick_proof.py
+# Allowed state: 0 hard violations
+# Unapproved state: ContractBreachError
+```
+
+This synthetic example checks a supplied `release.approved` boolean. It does
+not detect PII, assess security, or certify arbitrary agent behavior. AgentAssert
+supports AI Reliability Engineering through explicit contracts and observable checks.
+
+[Research on the dedicated site](https://agentassert.com/research) · [Qualixar overview](https://qualixar.com/products/agentassert) · [Author and research context](https://varunpratap.com/products/agentassert). If this check helps your workflow, [star the repository](https://github.com/qualixar/agentassert-abc); using the package does not require a star.
+
+
@@ -22,9 +45,9 @@
---
-AgentAssert is the **formal behavioral specification and runtime enforcement engine** for autonomous AI agents. Define what your agent must and must not do in a YAML contract, then enforce those rules at runtime with mathematical guarantees.
+AgentAssert is the **formal behavioral specification and runtime enforcement engine** for autonomous AI agents. Define what your agent must and must not do in a YAML contract, then check supplied signals at runtime. The papers below state the assumptions and scope of the mathematical results.
-It is the only framework combining all **6 pillars** of rigorous agent governance:
+The framework combines **6 components** of agent governance:
1. **ContractSpec DSL** -- YAML-based behavioral specification with 14 operators
2. **Hard/Soft Constraints** -- Formal separation with graduated enforcement and recovery
diff --git a/docs/QUICK_PROOF.md b/docs/QUICK_PROOF.md
new file mode 100644
index 0000000..cde901c
--- /dev/null
+++ b/docs/QUICK_PROOF.md
@@ -0,0 +1,20 @@
+# A small runtime contract proof
+
+Install Python 3.12+ and `python -m pip install "agentassert-abc[yaml,math]"`.
+From a repository clone, run `python examples/00_quick_proof.py`.
+The example contains its YAML contract; it does not depend on a relative contract file.
+
+Expected output:
+
+```text
+Allowed state: 0 hard violations
+Unapproved state: ContractBreachError
+```
+
+Both branches were executed against source version 0.7.1 on 1 October 2026.
+The caller supplies the boolean signal. `check` returns violations;
+`check_and_raise` raises for the false signal. The application must call the
+check before the protected action. This is a synthetic behavior proof, not an
+accuracy, latency, PII detection or security guarantee.
+
+See [enforcement coverage](enforcement-coverage.md) for adapter and host boundaries.
diff --git a/examples/00_quick_proof.py b/examples/00_quick_proof.py
new file mode 100644
index 0000000..ace5740
--- /dev/null
+++ b/examples/00_quick_proof.py
@@ -0,0 +1,33 @@
+"""Synthetic structured-signal proof; no network or agent framework required."""
+import agentassert_abc as aa
+from agentassert_abc.integrations.generic import GenericAdapter
+
+
+def main() -> None:
+ contract = aa.loads("""contractspec: "0.1"
+kind: agent
+name: synthetic-release-check
+description: Checks a caller-supplied approval flag, not a security classifier.
+version: "1.0.0"
+invariants:
+ hard:
+ - name: release-approved
+ check:
+ field: release.approved
+ equals: true
+""")
+ adapter = GenericAdapter(contract)
+ allowed = adapter.check({"release.approved": True})
+ if allowed.hard_violations != 0:
+ raise AssertionError("The approved synthetic state must pass")
+ print("Allowed state: 0 hard violations")
+ try:
+ adapter.check_and_raise({"release.approved": False})
+ except aa.ContractBreachError:
+ print("Unapproved state: ContractBreachError")
+ else:
+ raise AssertionError("The unapproved synthetic state must raise")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/src/agentassert_abc/certification/factor_reliability.py b/src/agentassert_abc/certification/factor_reliability.py
index 4413df2..0277452 100644
--- a/src/agentassert_abc/certification/factor_reliability.py
+++ b/src/agentassert_abc/certification/factor_reliability.py
@@ -54,6 +54,7 @@
from __future__ import annotations
import dataclasses
+from typing import Any, cast
import numpy as np
from scipy.integrate import quad
@@ -185,10 +186,12 @@ def gaussian_copula_all_success(
if m == 2:
rho = float(np.clip(R[0, 1], -_RCLIP, _RCLIP))
cov = [[1.0, rho], [rho, 1.0]]
- return float(multivariate_normal.cdf(a, mean=[0.0, 0.0], cov=cov))
+ return float(multivariate_normal.cdf(a, mean=[0.0, 0.0], cov=cast("Any", cov)))
if not _assume_psd:
R = _psd_retract_corr(R) # noqa: N806
- return float(multivariate_normal.cdf(a, mean=np.zeros(m), cov=R, allow_singular=True))
+ return float(
+ multivariate_normal.cdf(a, mean=np.zeros(m), cov=cast("Any", R), allow_singular=True)
+ )
# ---------------------------------------------------------------------------
@@ -196,9 +199,7 @@ def gaussian_copula_all_success(
# ---------------------------------------------------------------------------
-def shared_factor_all_success(
- marginals: object, loadings: object, q: int = 64
-) -> float:
+def shared_factor_all_success(marginals: object, loadings: object, q: int = 64) -> float:
"""One-factor all-success reliability via Gauss–Hermite (LLD-B Thm B.6).
Under the shared-factor model :math:`U_j = λ_j Ξ + \\sqrt{1 − λ_j^2}\\,ε_j`
@@ -284,8 +285,11 @@ def _integrand(xi: float) -> float:
if abs(lam[j]) > 1e-9 and abs(a[j] / lam[j]) < _QUAD_HALFWIDTH
)
value, abserr = quad(
- _integrand, -_QUAD_HALFWIDTH, _QUAD_HALFWIDTH,
- points=kinks or None, limit=200,
+ _integrand,
+ -_QUAD_HALFWIDTH,
+ _QUAD_HALFWIDTH,
+ points=kinks or None,
+ limit=200,
)
lo, hi = frechet_all_success_bounds(p)
if value < lo - _FR_TOL or value > hi + _FR_TOL:
diff --git a/src/agentassert_abc/certification/slepian_floor.py b/src/agentassert_abc/certification/slepian_floor.py
index d96d138..7040feb 100644
--- a/src/agentassert_abc/certification/slepian_floor.py
+++ b/src/agentassert_abc/certification/slepian_floor.py
@@ -37,6 +37,7 @@
from __future__ import annotations
import dataclasses
+from typing import Any, cast
import numpy as np
from scipy.stats import multivariate_normal, norm
@@ -103,7 +104,9 @@ def _rho_from_failure_cells(qa: float, qb: float, f11: float) -> float:
for _ in range(60):
mid = 0.5 * (lo + hi)
cov = [[1.0, mid], [mid, 1.0]]
- val = multivariate_normal.cdf([za, zb], mean=[0.0, 0.0], cov=cov, allow_singular=True)
+ val = multivariate_normal.cdf(
+ [za, zb], mean=[0.0, 0.0], cov=cast("Any", cov), allow_singular=True
+ )
if float(val) < f11:
lo = mid
else:
@@ -198,9 +201,16 @@ def slepian_model_floor(passes: object, eta_conf: float = 0.05) -> SlepianModelF
)
if m == 1:
return SlepianModelFloor(
- floor=float(p_lo[0]), observed=observed, eta_conf=eta_conf, m=m, n=n,
- rho_lower=((1.0,),), p_lo=(float(p_lo[0]),), is_model_bound=True,
- basis="single stage (Slepian floor vacuous at m=1)", assumptions=assumptions,
+ floor=float(p_lo[0]),
+ observed=observed,
+ eta_conf=eta_conf,
+ m=m,
+ n=n,
+ rho_lower=((1.0,),),
+ p_lo=(float(p_lo[0]),),
+ is_model_bound=True,
+ basis="single stage (Slepian floor vacuous at m=1)",
+ assumptions=assumptions,
)
# Failure-marginal boxes: q = 1 − p, so q_lo = 1 − p_hi and q_hi = 1 − p_lo.
q_lo = 1.0 - p_hi
@@ -217,9 +227,9 @@ def slepian_model_floor(passes: object, eta_conf: float = 0.05) -> SlepianModelF
# _assume_psd=True: corr_used is already PD and monotone-safe; do NOT let
# gaussian_copula_all_success re-apply the unsafe scale-toward-0 retraction
# (double-projection hazard, Opus 5 audit 2026-08-11).
- floor = float(np.clip(
- gaussian_copula_all_success(p_lo, corr_used, _assume_psd=True), 0.0, 1.0
- ))
+ floor = float(
+ np.clip(gaussian_copula_all_success(p_lo, corr_used, _assume_psd=True), 0.0, 1.0)
+ )
basis = "Thm B.7 exact: Gaussian orthant at the monotone (p_lo, ρ_lo) corner (Slepian)"
except DependenceError:
# Grok CRIT#1: an indefinite lower corner admits no elementwise-dominated
diff --git a/src/agentassert_abc/dependence/estimators.py b/src/agentassert_abc/dependence/estimators.py
index c0f93a2..0d1d08d 100644
--- a/src/agentassert_abc/dependence/estimators.py
+++ b/src/agentassert_abc/dependence/estimators.py
@@ -21,11 +21,12 @@
therefore means "fail together".
* All estimators are immutable and side-effect free; inputs are never mutated.
"""
+
from __future__ import annotations
import math
from dataclasses import dataclass
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any, cast
import numpy as np
from scipy.optimize import brentq
@@ -51,14 +52,15 @@
class CoFailureTable:
"""Immutable 2x2 contingency of two agents' failure indicators.
+ Fractional continuity-corrected cells are supported by tetrachoric fitting.
Cells count missions by ``(a_failed, b_failed)``:
``n11`` both failed, ``n10`` only a, ``n01`` only b, ``n00`` neither.
"""
- n11: int
- n10: int
- n01: int
- n00: int
+ n11: float
+ n10: float
+ n01: float
+ n00: float
def __post_init__(self) -> None:
for name in ("n11", "n10", "n01", "n00"):
@@ -90,7 +92,7 @@ def from_pairs(
return cls(n11=n11, n10=n10, n01=n01, n00=n00)
@property
- def n(self) -> int:
+ def n(self) -> float:
"""Total mission count."""
return self.n11 + self.n10 + self.n01 + self.n00
@@ -181,15 +183,15 @@ def tetrachoric(table: CoFailureTable) -> float:
"""
pa, pb, p11 = table.p_a, table.p_b, table.p11
if not (0.0 < pa < 1.0) or not (0.0 < pb < 1.0):
- raise DependenceError(
- "tetrachoric undefined for a degenerate marginal (0 or 1)"
- )
+ raise DependenceError("tetrachoric undefined for a degenerate marginal (0 or 1)")
tau_a = float(norm.ppf(pa))
tau_b = float(norm.ppf(pb))
def joint(rho: float) -> float:
cov = [[1.0, rho], [rho, 1.0]]
- return float(multivariate_normal.cdf([tau_a, tau_b], mean=[0.0, 0.0], cov=cov))
+ return float(
+ multivariate_normal.cdf([tau_a, tau_b], mean=[0.0, 0.0], cov=cast("Any", cov))
+ )
# Frechet feasibility: p11 must lie within the joint's attainable range.
lo, hi = -0.999999, 0.999999
@@ -198,7 +200,7 @@ def joint(rho: float) -> float:
return -1.0
if f_hi < 0: # even at rho=+1 the joint is below p11 -> clamp
return 1.0
- return float(brentq(lambda r: joint(r) - p11, lo, hi, xtol=1e-10))
+ return float(cast("float", brentq(lambda r: joint(r) - p11, lo, hi, xtol=1e-10)))
def tau_a_min_samples(eps: float, alpha: float) -> int:
diff --git a/src/agentassert_abc/enforce/bridge.py b/src/agentassert_abc/enforce/bridge.py
index 20e094e..6d9c9be 100644
--- a/src/agentassert_abc/enforce/bridge.py
+++ b/src/agentassert_abc/enforce/bridge.py
@@ -27,7 +27,7 @@
import threading
import uuid
from dataclasses import dataclass, field
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, TypedDict
from agentassert_abc.exceptions import ContractBreachError
from agentassert_abc.gateway.content.pii import apply_pii_redaction, evaluate_pii_filter
@@ -40,6 +40,12 @@
__all__ = ["EnforcementBridge", "ToolDecision", "ToolOutcome"]
+class _DecisionIds(TypedDict):
+ tool: str
+ session_id: str
+ contract_id: str
+
+
@dataclass(frozen=True)
class ToolDecision:
"""The verdict on a tool call that has not run yet.
@@ -396,7 +402,7 @@ def _count_deny(self) -> None:
with self._lock:
self._denied += 1
- def _ids(self, tool: str) -> dict[str, str]:
+ def _ids(self, tool: str) -> _DecisionIds:
return {
"tool": tool,
"session_id": self._session_id,
diff --git a/src/agentassert_abc/enforce/shims/agentscope.py b/src/agentassert_abc/enforce/shims/agentscope.py
index e2ff549..589ed60 100644
--- a/src/agentassert_abc/enforce/shims/agentscope.py
+++ b/src/agentassert_abc/enforce/shims/agentscope.py
@@ -155,6 +155,8 @@ def _rewrite(kwargs: dict[str, Any], holder: str | None, args: dict[str, Any]) -
if holder is None:
return {**kwargs, "input": args}
candidate = kwargs.get(holder)
+ if candidate is None:
+ return kwargs
if isinstance(candidate, dict):
return {**kwargs, holder: {**candidate, "input": args}}
# A structured block: mutate the copy's field, since we cannot rebuild an
diff --git a/src/agentassert_abc/experiments/logging_schema.py b/src/agentassert_abc/experiments/logging_schema.py
index 122af6f..5dd3a92 100644
--- a/src/agentassert_abc/experiments/logging_schema.py
+++ b/src/agentassert_abc/experiments/logging_schema.py
@@ -167,7 +167,7 @@ class MissionRecord:
mission_id: str
cluster_id: str
# Ledger 3f: narrow from str to Literal so static analysis catches invalid values.
- motif: Literal["series2", "series3", "parallel2", "quorum2of3", "hierarchy"]
+ motif: Literal["series2", "series3", "parallel2", "quorum2of3", "quorum3of4", "hierarchy"]
sharing_condition: Literal["same_model", "same_vendor", "different_vendor"]
route: tuple[str, ...]
components: tuple[ComponentRecord, ...]
@@ -184,7 +184,7 @@ def make(
mission_id: str,
cluster_id: str,
# Ledger 3f: Literal types narrow the parameter to valid values only.
- motif: Literal["series2", "series3", "parallel2", "quorum2of3", "hierarchy"],
+ motif: Literal["series2", "series3", "parallel2", "quorum2of3", "quorum3of4", "hierarchy"],
sharing_condition: Literal["same_model", "same_vendor", "different_vendor"],
route: tuple[str, ...],
components: tuple[ComponentRecord, ...],
diff --git a/src/agentassert_abc/experiments/motifs.py b/src/agentassert_abc/experiments/motifs.py
index 367b77b..46b3d6f 100644
--- a/src/agentassert_abc/experiments/motifs.py
+++ b/src/agentassert_abc/experiments/motifs.py
@@ -70,7 +70,7 @@
import dataclasses
import datetime
-from typing import TYPE_CHECKING, Final, Protocol
+from typing import TYPE_CHECKING, Final, Literal, Protocol, cast
from agentassert_abc.exceptions import AgentAssertError
from agentassert_abc.experiments.logging_schema import (
@@ -108,8 +108,7 @@ class ModelClient(Protocol):
satisfies this protocol. No inheritance required.
"""
- def generate(self, model: str, prompt: str) -> ModelResponse:
- ... # pragma: no cover
+ def generate(self, model: str, prompt: str) -> ModelResponse: ... # pragma: no cover
# ---------------------------------------------------------------------------
@@ -335,9 +334,7 @@ def _run_series(
resp = client.generate(model, prompt)
hard_ok = score(task, resp.text)
soft_ok = score_soft(task, resp.text)
- comps[node_id] = _make_comp(
- node_id, resp, hard_ok=hard_ok, soft_ok=soft_ok, scored=True
- )
+ comps[node_id] = _make_comp(node_id, resp, hard_ok=hard_ok, soft_ok=soft_ok, scored=True)
prev_output = resp.text
total_tokens += resp.input_tokens + resp.output_tokens
total_cost += resp.cost_usd
@@ -415,9 +412,7 @@ def _run_parallel_quorum(
total_cost += resp.cost_usd
passing_ids: list[str] = [
- bid
- for bid in branch_ids
- if branch_comps[bid].hard_ok and branch_comps[bid].soft_ok
+ bid for bid in branch_ids if branch_comps[bid].hard_ok and branch_comps[bid].soft_ok
]
quorum_met: bool = len(passing_ids) >= quorum_threshold
@@ -439,14 +434,13 @@ def _run_parallel_quorum(
# Handoffs: only passing branches emit handoffs to aggregator
handoffs: tuple[HandoffRecord, ...] = tuple(
- HandoffRecord(from_id=pid, to_id=agg_id, handoff_ok=True)
- for pid in passing_ids
+ HandoffRecord(from_id=pid, to_id=agg_id, handoff_ok=True) for pid in passing_ids
)
# Full component vector: all branches (in motif order) + aggregator
- all_comps: tuple[ComponentRecord, ...] = tuple(
- branch_comps[bid] for bid in branch_ids
- ) + (agg_comp,)
+ all_comps: tuple[ComponentRecord, ...] = tuple(branch_comps[bid] for bid in branch_ids) + (
+ agg_comp,
+ )
return realized_route, all_comps, handoffs, total_tokens, total_cost
@@ -505,7 +499,8 @@ def _run_hierarchy(
)
hard_ok_s = score(task, resp_s.text)
comps[sup_id] = _make_comp(
- sup_id, resp_s,
+ sup_id,
+ resp_s,
hard_ok=hard_ok_s,
soft_ok=score_soft(task, resp_s.text),
scored=True,
@@ -520,7 +515,8 @@ def _run_hierarchy(
)
hard_ok_w = score(task, resp_w.text)
comps[worker_id] = _make_comp(
- worker_id, resp_w,
+ worker_id,
+ resp_w,
hard_ok=hard_ok_w,
soft_ok=score_soft(task, resp_w.text),
scored=True,
@@ -535,7 +531,8 @@ def _run_hierarchy(
)
hard_ok_v = score(task, resp_v.text)
comps[verifier_id] = _make_comp(
- verifier_id, resp_v,
+ verifier_id,
+ resp_v,
hard_ok=hard_ok_v,
soft_ok=score_soft(task, resp_v.text),
scored=True,
@@ -558,9 +555,7 @@ def _run_hierarchy(
)
# Full component vector in motif.nodes order (deterministic)
- all_comps: tuple[ComponentRecord, ...] = tuple(
- comps[nid] for nid in motif.nodes
- )
+ all_comps: tuple[ComponentRecord, ...] = tuple(comps[nid] for nid in motif.nodes)
return motif.route, all_comps, handoffs, total_tokens, total_cost
@@ -627,9 +622,7 @@ def run_mission(
If *model_assignment* is missing a required generative node ID.
"""
if motif.name in ("series2", "series3"):
- route, comps, handoffs, tokens, cost = _run_series(
- motif, task, model_assignment, client
- )
+ route, comps, handoffs, tokens, cost = _run_series(motif, task, model_assignment, client)
elif motif.name in ("parallel2", "quorum2of3", "quorum3of4"):
route, comps, handoffs, tokens, cost = _run_parallel_quorum(
motif, task, model_assignment, client
@@ -640,19 +633,18 @@ def run_mission(
)
else:
raise MotifError(
- f"Unknown motif name {motif.name!r}. "
- f"Registered names: {sorted(MOTIF_LIBRARY)}"
+ f"Unknown motif name {motif.name!r}. Registered names: {sorted(MOTIF_LIBRARY)}"
)
- timestamp = (
- datetime.datetime.now(_UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
- )
+ timestamp = datetime.datetime.now(_UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
return MissionRecord.make(
mission_id=mission_id,
cluster_id=cluster_id,
motif=motif.name,
- sharing_condition=sharing_condition,
+ sharing_condition=cast(
+ "Literal['same_model', 'same_vendor', 'different_vendor']", sharing_condition
+ ),
route=route,
components=comps,
handoffs=handoffs,
diff --git a/src/agentassert_abc/experiments/providers.py b/src/agentassert_abc/experiments/providers.py
index dd14789..79249a1 100644
--- a/src/agentassert_abc/experiments/providers.py
+++ b/src/agentassert_abc/experiments/providers.py
@@ -500,12 +500,7 @@ def _parse_response(self, raw: dict, model: str) -> ModelResponse:
# PROVIDER_PRICES table (Meta/Grok do not report a cost field).
cost_usd: float
reported = usage.get("cost")
- reported_ok = (
- isinstance(reported, (int, float))
- and not isinstance(reported, bool)
- and reported >= 0
- )
- if reported_ok:
+ if isinstance(reported, (int, float)) and not isinstance(reported, bool) and reported >= 0:
cost_usd = float(reported)
else:
fallback_prices = (
diff --git a/src/agentassert_abc/mcp/interposer.py b/src/agentassert_abc/mcp/interposer.py
index 697bc4f..9e0fb00 100644
--- a/src/agentassert_abc/mcp/interposer.py
+++ b/src/agentassert_abc/mcp/interposer.py
@@ -28,7 +28,7 @@
import subprocess
import sys
import threading
-from typing import TYPE_CHECKING, TextIO
+from typing import TYPE_CHECKING, TextIO, cast
from agentassert_abc.mcp import jsonrpc
@@ -185,8 +185,8 @@ def run_guard(
guard,
client_in=client_in if client_in is not None else sys.stdin,
client_out=client_out if client_out is not None else sys.stdout,
- server_in=proc.stdin,
- server_out=proc.stdout,
+ server_in=cast("TextIO", proc.stdin),
+ server_out=cast("TextIO", proc.stdout),
)
try:
pump.run()
diff --git a/src/agentassert_abc/process/models.py b/src/agentassert_abc/process/models.py
index 5887760..222385b 100644
--- a/src/agentassert_abc/process/models.py
+++ b/src/agentassert_abc/process/models.py
@@ -260,8 +260,9 @@ class ContractSpecExtended(ContractSpec):
"""
dsl_version: str = "0.3"
- invariants: InvariantsExtended | None = None
- recovery: RecoveryConfigExtended | None = None
+ # Frozen Pydantic fields validate this narrowing; pyright treats them as mutable.
+ invariants: InvariantsExtended | None = None # pyright: ignore[reportIncompatibleVariableOverride]
+ recovery: RecoveryConfigExtended | None = None # pyright: ignore[reportIncompatibleVariableOverride]
upstream: UpstreamConfig | None = None