Fix CI type gates and add a working contract acquisition path

README.md

@@ -1,7 +1,30 @@
-<p align="center">
-  <strong>AgentAssert</strong><br>
-  <em>Formal Behavioral Contracts for AI Agents</em>
-</p>
+# 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.
+
+
 
 <p align="center">
   <a href="https://pypi.org/project/agentassert-abc/"><img src="https://img.shields.io/pypi/v/agentassert-abc?style=flat-square&color=blue" alt="PyPI"></a>
@@ -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

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.

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()

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,20 +186,20 @@ 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)
+    )
 
 
 # ---------------------------------------------------------------------------
 # Thm B.6 — one-factor all-success reduction (Gauss–Hermite, O(mQ))
 # ---------------------------------------------------------------------------
 
 
-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:

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

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:

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,

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

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, ...],