diff --git a/lionagi/governance/__init__.py b/lionagi/governance/__init__.py new file mode 100644 index 000000000..6930cb0f8 --- /dev/null +++ b/lionagi/governance/__init__.py @@ -0,0 +1,51 @@ +# Copyright (c) 2023-2026, HaiyangLi +# SPDX-License-Identifier: Apache-2.0 + +"""lionagi.governance — minimal gates, evidence chain, and TaskCertificate. + +Public surface: + - GatePolicy / GateExecutor / GateVerdict / GateResult / Enforcement + - EvidenceChain / EvidenceNode / ChainVerification / ChainVerifier + - TaskCertificate / CertificateGrade + - GoverningContext (binds the above into one per-run handle) + - GovernanceViolationError +""" + +from lionagi.governance.certificate import CertificateGrade, TaskCertificate +from lionagi.governance.context import GoverningContext +from lionagi.governance.errors import GovernanceViolationError +from lionagi.governance.evidence import ( + GENESIS_HASH, + ChainVerification, + ChainVerifier, + EvidenceChain, + EvidenceNode, + LogTier, + compute_node_hash, +) +from lionagi.governance.gates import ( + Enforcement, + GateExecutor, + GatePolicy, + GateResult, + GateVerdict, +) + +__all__ = [ + "GENESIS_HASH", + "CertificateGrade", + "ChainVerification", + "ChainVerifier", + "Enforcement", + "EvidenceChain", + "EvidenceNode", + "GateExecutor", + "GatePolicy", + "GateResult", + "GateVerdict", + "GovernanceViolationError", + "GoverningContext", + "LogTier", + "TaskCertificate", + "compute_node_hash", +] diff --git a/lionagi/governance/certificate.py b/lionagi/governance/certificate.py new file mode 100644 index 000000000..4a8336c79 --- /dev/null +++ b/lionagi/governance/certificate.py @@ -0,0 +1,78 @@ +# Copyright (c) 2023-2026, HaiyangLi +# SPDX-License-Identifier: Apache-2.0 + +"""TaskCertificate: auditable outcome record produced when a governed task completes.""" + +from __future__ import annotations + +import enum +from datetime import datetime + +from pydantic import Field + +from lionagi.protocols.generic.element import Element + +__all__ = [ + "CertificateGrade", + "TaskCertificate", +] + + +class CertificateGrade(str, enum.Enum): + FULL = "full" # zero hard denials + PARTIAL = "partial" # advisory flags only + FAILED = "failed" # at least one hard denial + + +class TaskCertificate(Element): + """Immutable certificate minted when a governed task run finishes. + + Grades: FULL (clean), PARTIAL (advisories only), FAILED (hard denial hit). + """ + + task_id: str + grade: CertificateGrade + evidence_chain_head: str + started_at: datetime + completed_at: datetime + op_count: int = 0 + ops_allowed: int = 0 + ops_denied: int = 0 + ops_advisory: int = 0 + chain_verified: bool = True + gate_results_summary: dict[str, int] = Field(default_factory=dict) + + @classmethod + def mint( + cls, + *, + task_id: str, + evidence_chain_head: str, + started_at: datetime, + completed_at: datetime, + op_count: int, + ops_allowed: int, + ops_denied: int, + ops_advisory: int, + chain_verified: bool = True, + gate_results_summary: dict[str, int] | None = None, + ) -> TaskCertificate: + if not chain_verified or ops_denied > 0: + grade = CertificateGrade.FAILED + elif ops_advisory > 0: + grade = CertificateGrade.PARTIAL + else: + grade = CertificateGrade.FULL + return cls( + task_id=task_id, + grade=grade, + evidence_chain_head=evidence_chain_head, + started_at=started_at, + completed_at=completed_at, + op_count=op_count, + ops_allowed=ops_allowed, + ops_denied=ops_denied, + ops_advisory=ops_advisory, + chain_verified=chain_verified, + gate_results_summary=gate_results_summary or {}, + ) diff --git a/lionagi/governance/context.py b/lionagi/governance/context.py new file mode 100644 index 000000000..b62666f6d --- /dev/null +++ b/lionagi/governance/context.py @@ -0,0 +1,126 @@ +# Copyright (c) 2023-2026, HaiyangLi +# SPDX-License-Identifier: Apache-2.0 + +"""GoverningContext: runtime gate + evidence handle for one governed task run.""" + +from __future__ import annotations + +from datetime import datetime, timezone + +from lionagi.governance.certificate import TaskCertificate +from lionagi.governance.errors import GovernanceViolationError +from lionagi.governance.evidence import EvidenceChain, LogTier +from lionagi.governance.gates import GateExecutor, GatePolicy, GateResult, GateVerdict + +__all__ = [ + "GoverningContext", +] + + +class GoverningContext: + """Runtime governance handle: evaluate gates and record evidence for one task. + + Usage:: + + ctx = GoverningContext(task_id="run-42", policies=[...]) + result = ctx.check("my_tool") # raises GovernanceViolationError on hard DENY + ctx.record({"event": "tool_called"}) + cert = ctx.complete() + """ + + def __init__( + self, + task_id: str, + policies: list[GatePolicy] | None = None, + *, + raise_on_deny: bool = True, + ) -> None: + self.task_id = task_id + self.raise_on_deny = raise_on_deny + self._executor = GateExecutor(policies or []) + self._chain = EvidenceChain() + self._started_at: datetime = datetime.now(tz=timezone.utc) + self._op_count: int = 0 + self._ops_allowed: int = 0 + self._ops_denied: int = 0 + self._ops_advisory: int = 0 + self._gate_tally: dict[str, int] = {} + + def check(self, tool_name: str, ctx: object = None) -> GateResult: + """Evaluate all policies for *tool_name*. + + Appends the result to the evidence chain. Raises + GovernanceViolationError when the verdict is DENY and + *raise_on_deny* is True; otherwise returns the result. + + Also emits a GateDenied signal if a branch is provided via *ctx* + (duck-typed: ``ctx.emit`` must exist). + """ + result = self._executor.evaluate(tool_name, ctx) + self._op_count += 1 + verdict = result.verdict + + if verdict is GateVerdict.ALLOW: + self._ops_allowed += 1 + elif verdict is GateVerdict.ADVISORY: + self._ops_advisory += 1 + else: + self._ops_denied += 1 + + gate_id = result.gate_id or "_allow" + self._gate_tally[gate_id] = self._gate_tally.get(gate_id, 0) + 1 + + self._chain.append(result.to_evidence_dict(), tier=LogTier.IMMUTABLE) + + if verdict is GateVerdict.DENY: + self._emit_gate_denied(tool_name, result, ctx) + if self.raise_on_deny: + raise GovernanceViolationError(result) + + return result + + def record(self, content: dict, *, tier: LogTier = LogTier.IMMUTABLE) -> None: + """Append an arbitrary evidence entry to the chain.""" + self._chain.append(content, tier=tier) + + def complete(self) -> TaskCertificate: + """Mint a TaskCertificate from the accumulated run state.""" + return TaskCertificate.mint( + task_id=self.task_id, + evidence_chain_head=self._chain.head_hash(), + started_at=self._started_at, + completed_at=datetime.now(tz=timezone.utc), + op_count=self._op_count, + ops_allowed=self._ops_allowed, + ops_denied=self._ops_denied, + ops_advisory=self._ops_advisory, + chain_verified=self._chain.verify().valid, + gate_results_summary=dict(self._gate_tally), + ) + + @property + def evidence_chain(self) -> EvidenceChain: + return self._chain + + @staticmethod + def _emit_gate_denied(tool_name: str, result: GateResult, ctx: object) -> None: + if ctx is None or not hasattr(ctx, "emit"): + return + from lionagi.session.signal import GateDenied # noqa: PLC0415 + + try: + import asyncio # noqa: PLC0415 + + signal = GateDenied( + data={ + "tool_name": tool_name, + "gate_id": result.gate_id, + "justification": result.justification, + }, + emitter_role="governance", + ) + loop = asyncio.get_event_loop() + if loop.is_running(): + loop.create_task(ctx.emit(signal)) # type: ignore[attr-defined] + except Exception: # noqa: S110 — best-effort fire-and-forget; never block the caller + pass diff --git a/lionagi/governance/errors.py b/lionagi/governance/errors.py new file mode 100644 index 000000000..4379fbb7c --- /dev/null +++ b/lionagi/governance/errors.py @@ -0,0 +1,19 @@ +# Copyright (c) 2023-2026, HaiyangLi +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from lionagi.governance.gates import GateResult + +__all__ = [ + "GovernanceViolationError", +] + + +class GovernanceViolationError(Exception): + def __init__(self, result: GateResult) -> None: + self.result = result + super().__init__(f"Gate {result.gate_id!r} denied: {result.justification}") diff --git a/lionagi/governance/evidence.py b/lionagi/governance/evidence.py new file mode 100644 index 000000000..916f66ad0 --- /dev/null +++ b/lionagi/governance/evidence.py @@ -0,0 +1,180 @@ +# Copyright (c) 2023-2026, HaiyangLi +# SPDX-License-Identifier: Apache-2.0 + +"""Append-only hash-linked evidence chain for governance audit trails.""" + +from __future__ import annotations + +import hashlib +import json +import threading +from enum import Enum +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field + +from lionagi.protocols.generic.element import Element + +__all__ = [ + "GENESIS_HASH", + "ChainVerification", + "ChainVerifier", + "EvidenceChain", + "EvidenceNode", + "LogTier", + "compute_node_hash", +] + +GENESIS_HASH = "0" * 64 + + +class LogTier(str, Enum): + MUTABLE = "MUTABLE" + PROTECTED = "PROTECTED" + IMMUTABLE = "IMMUTABLE" + + +def _canonical_json(content: dict[str, Any]) -> str: + return json.dumps(content, sort_keys=True, separators=(",", ":")) + + +def _sha256_hex(payload: str) -> str: + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + +def compute_node_hash( + content: dict[str, Any], + previous_hash: str, + tier: LogTier | str = LogTier.IMMUTABLE, +) -> str: + tier_value = getattr(tier, "value", tier) + return _sha256_hex(_canonical_json(content) + "|" + previous_hash + "|" + str(tier_value)) + + +class EvidenceNode(Element): + model_config = ConfigDict( + arbitrary_types_allowed=True, + use_enum_values=True, + populate_by_name=True, + extra="forbid", + frozen=True, + ) + + content: dict[str, Any] = Field(default_factory=dict) + previous_hash: str = Field(default=GENESIS_HASH) + node_hash: str = Field(default="") + tier: LogTier = Field(default=LogTier.IMMUTABLE) + + def model_post_init(self, __context: Any) -> None: + super().model_post_init(__context) + if not self.node_hash: + object.__setattr__( + self, + "node_hash", + compute_node_hash(self.content, self.previous_hash, self.tier), + ) + + def verify_hash(self) -> bool: + return self.node_hash == compute_node_hash(self.content, self.previous_hash, self.tier) + + +class ChainVerification(BaseModel): + valid: bool + checked_count: int = 0 + expected_count: int = 0 + first_invalid_index: int | None = None + violation_type: str | None = None + message: str = "" + + +class EvidenceChain(Element): + """Append-only blockchain-style evidence log. Nodes are stored in insertion order.""" + + model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid") + + _nodes: list[EvidenceNode] = [] + tip_hash: str = Field(default=GENESIS_HASH) + node_count: int = Field(default=0) + + def model_post_init(self, __context: Any) -> None: + super().model_post_init(__context) + object.__setattr__(self, "_nodes", []) + object.__setattr__(self, "_lock", threading.Lock()) + + def append( + self, + content: dict[str, Any], + tier: LogTier = LogTier.IMMUTABLE, + ) -> EvidenceNode: + with self._lock: # type: ignore[attr-defined] + node = EvidenceNode(content=content, previous_hash=self.tip_hash, tier=tier) + self._nodes.append(node) # type: ignore[attr-defined] + object.__setattr__(self, "tip_hash", node.node_hash) + object.__setattr__(self, "node_count", self.node_count + 1) + return node + + def nodes(self) -> list[EvidenceNode]: + return list(self._nodes) # type: ignore[attr-defined] + + def verify(self) -> ChainVerification: + return ChainVerifier.verify(self) + + def head_hash(self) -> str: + return self.tip_hash + + +class ChainVerifier: + @staticmethod + def verify(chain: EvidenceChain) -> ChainVerification: + nodes = chain.nodes() + expected_count = chain.node_count + + if len(nodes) != expected_count: + return ChainVerification( + valid=False, + checked_count=len(nodes), + expected_count=expected_count, + first_invalid_index=min(len(nodes), expected_count), + violation_type="delete", + message="Node count does not match expected chain length.", + ) + + expected_prev = GENESIS_HASH + for idx, node in enumerate(nodes): + if node.previous_hash != expected_prev: + return ChainVerification( + valid=False, + checked_count=idx + 1, + expected_count=expected_count, + first_invalid_index=idx, + violation_type="reorder", + message="Previous hash does not match predecessor.", + ) + expected_node_hash = compute_node_hash(node.content, node.previous_hash, node.tier) + if node.node_hash != expected_node_hash: + return ChainVerification( + valid=False, + checked_count=idx + 1, + expected_count=expected_count, + first_invalid_index=idx, + violation_type="tamper", + message="Node hash does not match content.", + ) + expected_prev = node.node_hash + + if expected_prev != chain.tip_hash: + return ChainVerification( + valid=False, + checked_count=len(nodes), + expected_count=expected_count, + first_invalid_index=len(nodes) - 1 if nodes else None, + violation_type="head_mismatch", + message="Computed head does not match chain tip.", + ) + + return ChainVerification( + valid=True, + checked_count=len(nodes), + expected_count=expected_count, + message="Chain verified.", + ) diff --git a/lionagi/governance/gates.py b/lionagi/governance/gates.py new file mode 100644 index 000000000..d52f2004d --- /dev/null +++ b/lionagi/governance/gates.py @@ -0,0 +1,125 @@ +# Copyright (c) 2023-2026, HaiyangLi +# SPDX-License-Identifier: Apache-2.0 + +"""Tool gate evaluation: policy-driven ALLOW / ADVISORY / DENY verdicts.""" + +from __future__ import annotations + +import enum +import time +from collections.abc import Callable +from typing import Any + +from pydantic import BaseModel + +__all__ = [ + "Enforcement", + "GatePolicy", + "GateResult", + "GateVerdict", + "GateExecutor", +] + + +class Enforcement(str, enum.Enum): + HARD = "hard" + SOFT = "soft" + ADVISORY = "advisory" + + +class GatePolicy(BaseModel): + """A single policy rule binding a gate callable to a tool name.""" + + target_tool: str + enforcement: Enforcement = Enforcement.HARD + gate_id: str = "" + gate_fn: Callable[[str, Any], bool] | None = None + + model_config = {"arbitrary_types_allowed": True} + + def matches(self, tool_name: str) -> bool: + return self.target_tool == tool_name + + def evaluate(self, tool_name: str, ctx: Any) -> bool: + """Return True when the gate fires (i.e. wants to block).""" + if self.gate_fn is None: + return True + return bool(self.gate_fn(tool_name, ctx)) + + +class GateVerdict(str, enum.Enum): + ALLOW = "allow" + DENY = "deny" + ADVISORY = "advisory" + + +class GateResult(BaseModel): + verdict: GateVerdict + justification: str + gate_id: str + elapsed_ms: float = 0.0 + + def denied(self) -> bool: + return self.verdict is GateVerdict.DENY + + def to_evidence_dict(self) -> dict[str, Any]: + return { + "event": "gate_result", + "verdict": self.verdict.value, + "gate_id": self.gate_id, + "justification": self.justification, + "elapsed_ms": self.elapsed_ms, + } + + +class GateExecutor: + """Evaluates a list of GatePolicy rules for one tool invocation. + + Matching is by exact target_tool name. A HARD policy that fires short- + circuits with DENY. SOFT/ADVISORY policies are collected; if any fired + the final verdict is ADVISORY. No match → ALLOW. + """ + + def __init__(self, policies: list[GatePolicy]) -> None: + self.policies = policies + + def evaluate(self, tool_name: str, ctx: Any = None) -> GateResult: + start = time.perf_counter() + advisories: list[GateResult] = [] + + for policy in self.policies: + if not policy.matches(tool_name): + continue + if not policy.evaluate(tool_name, ctx): + continue + elapsed_ms = (time.perf_counter() - start) * 1000.0 + if policy.enforcement is Enforcement.HARD: + return GateResult( + verdict=GateVerdict.DENY, + justification=f"Hard gate {policy.gate_id!r} blocks {tool_name!r}", + gate_id=policy.gate_id, + elapsed_ms=elapsed_ms, + ) + advisories.append( + GateResult( + verdict=GateVerdict.ADVISORY, + justification=f"Advisory gate {policy.gate_id!r} flagged {tool_name!r}", + gate_id=policy.gate_id, + elapsed_ms=elapsed_ms, + ) + ) + + elapsed_ms = (time.perf_counter() - start) * 1000.0 + if advisories: + return GateResult( + verdict=GateVerdict.ADVISORY, + justification=f"{len(advisories)} advisory gate(s) flagged {tool_name!r}", + gate_id=advisories[-1].gate_id, + elapsed_ms=elapsed_ms, + ) + return GateResult( + verdict=GateVerdict.ALLOW, + justification=f"All gates passed for {tool_name!r}", + gate_id="", + elapsed_ms=elapsed_ms, + ) diff --git a/tests/governance/__init__.py b/tests/governance/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/governance/test_certificate.py b/tests/governance/test_certificate.py new file mode 100644 index 000000000..f556de173 --- /dev/null +++ b/tests/governance/test_certificate.py @@ -0,0 +1,84 @@ +# Copyright (c) 2023-2026, HaiyangLi +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +from datetime import datetime, timezone + +import pytest + +from lionagi.governance.certificate import CertificateGrade, TaskCertificate +from lionagi.governance.evidence import GENESIS_HASH + + +def _now(): + return datetime.now(tz=timezone.utc) + + +def _mint(**kwargs): + defaults = dict( + task_id="t1", + evidence_chain_head=GENESIS_HASH, + started_at=_now(), + completed_at=_now(), + op_count=0, + ops_allowed=0, + ops_denied=0, + ops_advisory=0, + ) + defaults.update(kwargs) + return TaskCertificate.mint(**defaults) + + +class TestCertificateGrade: + def test_full_when_no_denials_no_advisories(self): + cert = _mint(op_count=3, ops_allowed=3) + assert cert.grade == CertificateGrade.FULL + + def test_partial_when_only_advisories(self): + cert = _mint(op_count=2, ops_allowed=1, ops_advisory=1) + assert cert.grade == CertificateGrade.PARTIAL + + def test_failed_when_any_denial(self): + cert = _mint(op_count=2, ops_allowed=1, ops_denied=1) + assert cert.grade == CertificateGrade.FAILED + + def test_failed_takes_precedence_over_advisory(self): + cert = _mint(op_count=3, ops_allowed=1, ops_denied=1, ops_advisory=1) + assert cert.grade == CertificateGrade.FAILED + + def test_unverified_chain_forces_failed(self): + cert = _mint(op_count=3, ops_allowed=3, chain_verified=False) + assert cert.grade == CertificateGrade.FAILED + assert cert.chain_verified is False + + def test_verified_chain_defaults_true(self): + cert = _mint(op_count=1, ops_allowed=1) + assert cert.chain_verified is True + assert cert.grade == CertificateGrade.FULL + + def test_fields_stored(self): + t0 = _now() + t1 = _now() + cert = TaskCertificate.mint( + task_id="run-99", + evidence_chain_head="abc", + started_at=t0, + completed_at=t1, + op_count=5, + ops_allowed=4, + ops_denied=1, + ops_advisory=0, + gate_results_summary={"g1": 2}, + ) + assert cert.task_id == "run-99" + assert cert.evidence_chain_head == "abc" + assert cert.op_count == 5 + assert cert.gate_results_summary == {"g1": 2} + + def test_is_element(self): + from lionagi.protocols.generic.element import Element + + cert = _mint() + assert isinstance(cert, Element) + assert cert.id is not None diff --git a/tests/governance/test_evidence.py b/tests/governance/test_evidence.py new file mode 100644 index 000000000..ff50866ce --- /dev/null +++ b/tests/governance/test_evidence.py @@ -0,0 +1,155 @@ +# Copyright (c) 2023-2026, HaiyangLi +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import threading + +import pytest + +from lionagi.governance.evidence import ( + GENESIS_HASH, + ChainVerifier, + EvidenceChain, + EvidenceNode, + LogTier, + compute_node_hash, +) + + +class TestEvidenceNode: + def test_hash_computed_on_creation(self): + node = EvidenceNode(content={"k": "v"}, previous_hash=GENESIS_HASH) + assert node.node_hash != "" + assert len(node.node_hash) == 64 + + def test_verify_hash_passes(self): + node = EvidenceNode(content={"x": 1}, previous_hash=GENESIS_HASH) + assert node.verify_hash() + + def test_genesis_hash_is_zeros(self): + assert GENESIS_HASH == "0" * 64 + + def test_compute_node_hash_deterministic(self): + h1 = compute_node_hash({"a": 1}, GENESIS_HASH) + h2 = compute_node_hash({"a": 1}, GENESIS_HASH) + assert h1 == h2 + + def test_compute_node_hash_changes_with_content(self): + h1 = compute_node_hash({"a": 1}, GENESIS_HASH) + h2 = compute_node_hash({"a": 2}, GENESIS_HASH) + assert h1 != h2 + + def test_compute_node_hash_changes_with_prev(self): + h1 = compute_node_hash({"a": 1}, GENESIS_HASH) + h2 = compute_node_hash({"a": 1}, "1" * 64) + assert h1 != h2 + + +class TestEvidenceChain: + def test_empty_chain_verifies(self): + chain = EvidenceChain() + v = chain.verify() + assert v.valid + assert v.checked_count == 0 + + def test_append_single_node(self): + chain = EvidenceChain() + node = chain.append({"event": "test"}) + assert chain.node_count == 1 + assert chain.tip_hash == node.node_hash + assert chain.nodes()[0] is node + + def test_chain_links_correctly(self): + chain = EvidenceChain() + n1 = chain.append({"seq": 1}) + n2 = chain.append({"seq": 2}) + assert n2.previous_hash == n1.node_hash + + def test_verify_multi_node_chain(self): + chain = EvidenceChain() + for i in range(5): + chain.append({"i": i}) + v = chain.verify() + assert v.valid + assert v.checked_count == 5 + + def test_head_hash_matches_last_node(self): + chain = EvidenceChain() + nodes = [chain.append({"n": i}) for i in range(3)] + assert chain.head_hash() == nodes[-1].node_hash + + def test_nodes_returns_copy(self): + chain = EvidenceChain() + chain.append({"x": 1}) + snapshot = chain.nodes() + chain.append({"x": 2}) + assert len(snapshot) == 1 # snapshot is not affected by later appends + + def test_log_tier_stored(self): + chain = EvidenceChain() + node = chain.append({"e": "mutable"}, tier=LogTier.MUTABLE) + assert node.tier == LogTier.MUTABLE.value + + def test_concurrent_appends_are_safe(self): + chain = EvidenceChain() + errors: list[Exception] = [] + + def worker(): + try: + for i in range(20): + chain.append({"i": i}) + except Exception as e: + errors.append(e) + + threads = [threading.Thread(target=worker) for _ in range(5)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert not errors + assert chain.node_count == 100 + v = chain.verify() + assert v.valid + + +class TestChainVerifier: + def test_detects_count_mismatch(self): + chain = EvidenceChain() + chain.append({"x": 1}) + # Manually corrupt the count + object.__setattr__(chain, "node_count", 99) + v = ChainVerifier.verify(chain) + assert not v.valid + assert v.violation_type == "delete" + + def test_detects_tampered_node(self): + chain = EvidenceChain() + chain.append({"x": 1}) + nodes = chain.nodes() + # Tamper: rebuild a node with wrong hash stored + bad_node = EvidenceNode( + content={"x": 1}, + previous_hash=GENESIS_HASH, + node_hash="bad" + "0" * 61, # 64 chars, wrong + ) + chain._nodes[0] = bad_node # type: ignore[index] + v = chain.verify() + assert not v.valid + assert v.violation_type == "tamper" + + def test_detects_tier_downgrade(self): + chain = EvidenceChain() + chain.append({"x": 1}, tier=LogTier.IMMUTABLE) + node = chain.nodes()[0] + # Flip tier IMMUTABLE -> MUTABLE while keeping the stored node_hash. + object.__setattr__(node, "tier", LogTier.MUTABLE.value) + v = chain.verify() + assert not v.valid + assert v.violation_type == "tamper" + + def test_tier_is_part_of_hash(self): + immutable = compute_node_hash({"x": 1}, GENESIS_HASH, LogTier.IMMUTABLE) + mutable = compute_node_hash({"x": 1}, GENESIS_HASH, LogTier.MUTABLE) + assert immutable != mutable diff --git a/tests/governance/test_gates.py b/tests/governance/test_gates.py new file mode 100644 index 000000000..73a6176fe --- /dev/null +++ b/tests/governance/test_gates.py @@ -0,0 +1,98 @@ +# Copyright (c) 2023-2026, HaiyangLi +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import pytest + +from lionagi.governance.errors import GovernanceViolationError +from lionagi.governance.gates import Enforcement, GateExecutor, GatePolicy, GateVerdict + + +def _always_fires(tool_name, ctx): + return True + + +def _never_fires(tool_name, ctx): + return False + + +def make_policy(target, enforcement=Enforcement.HARD, gate_fn=_always_fires, gate_id="g1"): + return GatePolicy(target_tool=target, enforcement=enforcement, gate_id=gate_id, gate_fn=gate_fn) + + +class TestGateVerdict: + def test_allow_when_no_policies(self): + ex = GateExecutor([]) + r = ex.evaluate("my_tool") + assert r.verdict is GateVerdict.ALLOW + + def test_allow_when_no_matching_policy(self): + ex = GateExecutor([make_policy("other_tool")]) + r = ex.evaluate("my_tool") + assert r.verdict is GateVerdict.ALLOW + + def test_hard_deny_when_gate_fires(self): + ex = GateExecutor([make_policy("my_tool", Enforcement.HARD, _always_fires)]) + r = ex.evaluate("my_tool") + assert r.verdict is GateVerdict.DENY + assert r.denied() is not False # sanity + assert r.denied() + + def test_advisory_when_gate_fires_softly(self): + ex = GateExecutor([make_policy("my_tool", Enforcement.ADVISORY, _always_fires)]) + r = ex.evaluate("my_tool") + assert r.verdict is GateVerdict.ADVISORY + + def test_allow_when_gate_fn_returns_false(self): + ex = GateExecutor([make_policy("my_tool", Enforcement.HARD, _never_fires)]) + r = ex.evaluate("my_tool") + assert r.verdict is GateVerdict.ALLOW + + def test_hard_policy_short_circuits(self): + fired = [] + + def record(tool, ctx): + fired.append(tool) + return True + + policies = [ + make_policy("t", Enforcement.HARD, record, "g1"), + make_policy("t", Enforcement.HARD, record, "g2"), + ] + ex = GateExecutor(policies) + r = ex.evaluate("t") + assert r.verdict is GateVerdict.DENY + assert len(fired) == 1 # short-circuit after first hard deny + + def test_multiple_advisories_collapsed(self): + policies = [ + make_policy("t", Enforcement.ADVISORY, _always_fires, "a1"), + make_policy("t", Enforcement.ADVISORY, _always_fires, "a2"), + ] + ex = GateExecutor(policies) + r = ex.evaluate("t") + assert r.verdict is GateVerdict.ADVISORY + assert "2 advisory" in r.justification + + def test_elapsed_ms_is_nonnegative(self): + ex = GateExecutor([make_policy("t")]) + r = ex.evaluate("t") + assert r.elapsed_ms >= 0.0 + + def test_to_evidence_dict_shape(self): + ex = GateExecutor([]) + r = ex.evaluate("x") + d = r.to_evidence_dict() + assert d["event"] == "gate_result" + assert "verdict" in d and "gate_id" in d + + +class TestGovernanceViolationError: + def test_raises_with_gate_id(self): + from lionagi.governance.gates import GateResult + + result = GateResult(verdict=GateVerdict.DENY, justification="blocked", gate_id="g99") + err = GovernanceViolationError(result) + assert "g99" in str(err) + assert err.result is result diff --git a/tests/governance/test_governing_context.py b/tests/governance/test_governing_context.py new file mode 100644 index 000000000..a3a49e030 --- /dev/null +++ b/tests/governance/test_governing_context.py @@ -0,0 +1,122 @@ +# Copyright (c) 2023-2026, HaiyangLi +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import pytest + +from lionagi.governance.certificate import CertificateGrade +from lionagi.governance.context import GoverningContext +from lionagi.governance.errors import GovernanceViolationError +from lionagi.governance.gates import Enforcement, GatePolicy, GateVerdict + + +def _allow_policy(tool): + return GatePolicy( + target_tool=tool, enforcement=Enforcement.HARD, gate_id="block", gate_fn=lambda t, c: False + ) + + +def _deny_policy(tool): + return GatePolicy( + target_tool=tool, enforcement=Enforcement.HARD, gate_id="block", gate_fn=lambda t, c: True + ) + + +def _advisory_policy(tool): + return GatePolicy( + target_tool=tool, + enforcement=Enforcement.ADVISORY, + gate_id="warn", + gate_fn=lambda t, c: True, + ) + + +class TestGoverningContext: + def test_check_allow(self): + ctx = GoverningContext("t1", [_allow_policy("tool_a")]) + r = ctx.check("tool_a") + assert r.verdict is GateVerdict.ALLOW + + def test_check_deny_raises(self): + ctx = GoverningContext("t1", [_deny_policy("tool_a")]) + with pytest.raises(GovernanceViolationError) as exc: + ctx.check("tool_a") + assert "block" in str(exc.value) + + def test_check_deny_no_raise(self): + ctx = GoverningContext("t1", [_deny_policy("tool_a")], raise_on_deny=False) + r = ctx.check("tool_a") + assert r.denied() + + def test_advisory_does_not_raise(self): + ctx = GoverningContext("t1", [_advisory_policy("tool_a")]) + r = ctx.check("tool_a") + assert r.verdict is GateVerdict.ADVISORY + + def test_evidence_chain_grows(self): + ctx = GoverningContext("t1") + ctx.check("any_tool") + ctx.record({"custom": "event"}) + assert ctx.evidence_chain.node_count == 2 + + def test_complete_produces_full_cert(self): + ctx = GoverningContext("t1", [_allow_policy("t")]) + ctx.check("t") + cert = ctx.complete() + assert cert.grade == CertificateGrade.FULL + assert cert.op_count == 1 + assert cert.ops_allowed == 1 + + def test_complete_produces_failed_cert(self): + ctx = GoverningContext("t1", [_deny_policy("t")], raise_on_deny=False) + ctx.check("t") + cert = ctx.complete() + assert cert.grade == CertificateGrade.FAILED + assert cert.ops_denied == 1 + + def test_complete_produces_partial_cert(self): + ctx = GoverningContext("t1", [_advisory_policy("t")]) + ctx.check("t") + cert = ctx.complete() + assert cert.grade == CertificateGrade.PARTIAL + assert cert.ops_advisory == 1 + + def test_complete_evidence_chain_head_matches(self): + ctx = GoverningContext("t1") + ctx.record({"x": 1}) + cert = ctx.complete() + assert cert.evidence_chain_head == ctx.evidence_chain.head_hash() + + def test_no_policies_all_allowed(self): + ctx = GoverningContext("t1") + for i in range(5): + ctx.check(f"tool_{i}") + cert = ctx.complete() + assert cert.grade == CertificateGrade.FULL + assert cert.ops_allowed == 5 + + def test_complete_flags_tampered_chain(self): + ctx = GoverningContext("t1", [_allow_policy("t")]) + ctx.check("t") + # Tamper with the recorded evidence after the fact. + node = ctx.evidence_chain.nodes()[0] + object.__setattr__(node, "tier", "MUTABLE") + cert = ctx.complete() + assert cert.chain_verified is False + assert cert.grade == CertificateGrade.FAILED + + def test_gate_tally_in_cert(self): + policies = [ + GatePolicy( + target_tool="t", + enforcement=Enforcement.ADVISORY, + gate_id="warn", + gate_fn=lambda *_: True, + ), + ] + ctx = GoverningContext("t1", policies) + ctx.check("t") + ctx.check("t") + cert = ctx.complete() + assert cert.gate_results_summary.get("warn", 0) == 2