From 4a8a1bd5a4562019413d5b0f2dab27e4f71f2fd4 Mon Sep 17 00:00:00 2001 From: OceanLi <122793010+ohdearquant@users.noreply.github.com> Date: Thu, 18 Jun 2026 16:42:23 -0400 Subject: [PATCH 1/2] feat(governance): minimal gates + evidence + TaskCertificate slice (#1393) Implements the minimal governance slice as directed (Ocean 2026-06-18): - GatePolicy / GateExecutor / GateVerdict / GateResult / Enforcement Declarative policy rules bound to tool names. HARD enforcement short- circuits with DENY; SOFT/ADVISORY policies accumulate and collapse to a single ADVISORY verdict. gate_fn is a plain callable so policies need no DSL dependency. - EvidenceChain / EvidenceNode / ChainVerifier / ChainVerification Blockchain-style append-only log with SHA-256 hash linkage. Thread- safe via threading.Lock. ChainVerifier detects tamper, reorder, and delete violations. Uses stdlib list instead of Pile (no append_only on current Pile). - TaskCertificate / CertificateGrade Immutable Element minted at task completion. Grade computed from op counts: FULL (zero denials), PARTIAL (advisory only), FAILED (hard denial). mint() factory encapsulates grading logic. - GoverningContext Per-run handle binding the three above: check(tool_name) evaluates policies, writes to the evidence chain, emits GateDenied signal to a branch if one is provided (duck-typed ctx.emit), raises GovernanceViolationError on DENY unless raise_on_deny=False. complete() mints the TaskCertificate. Out of scope: charter DSL, SoD, JIT grants, break-glass, all pile/branch/session rewiring. GateDenied from session/signal.py is the only integration point. 43 tests, all passing. ruff clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- lionagi/governance/__init__.py | 53 +++++++ lionagi/governance/certificate.py | 75 +++++++++ lionagi/governance/context.py | 125 +++++++++++++++ lionagi/governance/errors.py | 19 +++ lionagi/governance/evidence.py | 173 +++++++++++++++++++++ lionagi/governance/gates.py | 125 +++++++++++++++ tests/governance/__init__.py | 0 tests/governance/test_certificate.py | 74 +++++++++ tests/governance/test_evidence.py | 140 +++++++++++++++++ tests/governance/test_gates.py | 98 ++++++++++++ tests/governance/test_governing_context.py | 112 +++++++++++++ 11 files changed, 994 insertions(+) create mode 100644 lionagi/governance/__init__.py create mode 100644 lionagi/governance/certificate.py create mode 100644 lionagi/governance/context.py create mode 100644 lionagi/governance/errors.py create mode 100644 lionagi/governance/evidence.py create mode 100644 lionagi/governance/gates.py create mode 100644 tests/governance/__init__.py create mode 100644 tests/governance/test_certificate.py create mode 100644 tests/governance/test_evidence.py create mode 100644 tests/governance/test_gates.py create mode 100644 tests/governance/test_governing_context.py diff --git a/lionagi/governance/__init__.py b/lionagi/governance/__init__.py new file mode 100644 index 000000000..e8a03c152 --- /dev/null +++ b/lionagi/governance/__init__.py @@ -0,0 +1,53 @@ +# Copyright (c) 2023-2026, HaiyangLi +# SPDX-License-Identifier: Apache-2.0 + +"""lionagi.governance — minimal gates, evidence chain, and TaskCertificate. + +Public surface (in scope): + - GatePolicy / GateExecutor / GateVerdict / GateResult / Enforcement + - EvidenceChain / EvidenceNode / ChainVerification / ChainVerifier + - TaskCertificate / CertificateGrade + - GoverningContext (binds the above into one per-run handle) + - GovernanceViolationError + +Out of scope at library level: charter DSL, SoD, JIT grants, break-glass. +""" + +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..9e5a21421 --- /dev/null +++ b/lionagi/governance/certificate.py @@ -0,0 +1,75 @@ +# 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 + 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, + gate_results_summary: dict[str, int] | None = None, + ) -> TaskCertificate: + if 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, + 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..3e17ec16b --- /dev/null +++ b/lionagi/governance/context.py @@ -0,0 +1,125 @@ +# 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, + 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..520d854e3 --- /dev/null +++ b/lionagi/governance/evidence.py @@ -0,0 +1,173 @@ +# 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 + + +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) -> str: + return _sha256_hex(_canonical_json(content) + "|" + previous_hash) + + +class LogTier(str, Enum): + MUTABLE = "MUTABLE" + PROTECTED = "PROTECTED" + IMMUTABLE = "IMMUTABLE" + + +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) + ) + + def verify_hash(self) -> bool: + return self.node_hash == compute_node_hash(self.content, self.previous_hash) + + +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) + 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..0b7a44ef9 --- /dev/null +++ b/tests/governance/test_certificate.py @@ -0,0 +1,74 @@ +# 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_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..dd016b2f2 --- /dev/null +++ b/tests/governance/test_evidence.py @@ -0,0 +1,140 @@ +# 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" 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..8d7ab4647 --- /dev/null +++ b/tests/governance/test_governing_context.py @@ -0,0 +1,112 @@ +# 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_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 From dfc73c2216017f32882c3664b6c65f9a5c4fa2fa Mon Sep 17 00:00:00 2001 From: OceanLi <122793010+ohdearquant@users.noreply.github.com> Date: Thu, 18 Jun 2026 16:51:09 -0400 Subject: [PATCH 2/2] fix(governance): fold tier into evidence hash, gate cert on chain validity, drop scope leak - evidence: compute_node_hash now includes the node tier, so an IMMUTABLE->MUTABLE downgrade is caught as a tamper by ChainVerifier (previously tier was outside the digest and silently mutable). - certificate: mint() refuses FULL/PARTIAL when chain_verified is False; GoverningContext.complete() verifies the chain before minting. - __init__: remove the public docstring line enumerating out-of-scope features; describe only what the module does. Co-Authored-By: Claude Opus 4.8 (1M context) --- lionagi/governance/__init__.py | 4 +-- lionagi/governance/certificate.py | 5 +++- lionagi/governance/context.py | 1 + lionagi/governance/evidence.py | 29 ++++++++++++++-------- tests/governance/test_certificate.py | 10 ++++++++ tests/governance/test_evidence.py | 15 +++++++++++ tests/governance/test_governing_context.py | 10 ++++++++ 7 files changed, 59 insertions(+), 15 deletions(-) diff --git a/lionagi/governance/__init__.py b/lionagi/governance/__init__.py index e8a03c152..6930cb0f8 100644 --- a/lionagi/governance/__init__.py +++ b/lionagi/governance/__init__.py @@ -3,14 +3,12 @@ """lionagi.governance — minimal gates, evidence chain, and TaskCertificate. -Public surface (in scope): +Public surface: - GatePolicy / GateExecutor / GateVerdict / GateResult / Enforcement - EvidenceChain / EvidenceNode / ChainVerification / ChainVerifier - TaskCertificate / CertificateGrade - GoverningContext (binds the above into one per-run handle) - GovernanceViolationError - -Out of scope at library level: charter DSL, SoD, JIT grants, break-glass. """ from lionagi.governance.certificate import CertificateGrade, TaskCertificate diff --git a/lionagi/governance/certificate.py b/lionagi/governance/certificate.py index 9e5a21421..4a8336c79 100644 --- a/lionagi/governance/certificate.py +++ b/lionagi/governance/certificate.py @@ -39,6 +39,7 @@ class TaskCertificate(Element): 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 @@ -53,9 +54,10 @@ def mint( ops_allowed: int, ops_denied: int, ops_advisory: int, + chain_verified: bool = True, gate_results_summary: dict[str, int] | None = None, ) -> TaskCertificate: - if ops_denied > 0: + if not chain_verified or ops_denied > 0: grade = CertificateGrade.FAILED elif ops_advisory > 0: grade = CertificateGrade.PARTIAL @@ -71,5 +73,6 @@ def mint( 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 index 3e17ec16b..b62666f6d 100644 --- a/lionagi/governance/context.py +++ b/lionagi/governance/context.py @@ -94,6 +94,7 @@ def complete(self) -> TaskCertificate: 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), ) diff --git a/lionagi/governance/evidence.py b/lionagi/governance/evidence.py index 520d854e3..916f66ad0 100644 --- a/lionagi/governance/evidence.py +++ b/lionagi/governance/evidence.py @@ -28,6 +28,12 @@ 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=(",", ":")) @@ -36,14 +42,13 @@ 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) -> str: - return _sha256_hex(_canonical_json(content) + "|" + previous_hash) - - -class LogTier(str, Enum): - MUTABLE = "MUTABLE" - PROTECTED = "PROTECTED" - IMMUTABLE = "IMMUTABLE" +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): @@ -64,11 +69,13 @@ 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, + "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) + return self.node_hash == compute_node_hash(self.content, self.previous_hash, self.tier) class ChainVerification(BaseModel): @@ -143,7 +150,7 @@ def verify(chain: EvidenceChain) -> ChainVerification: violation_type="reorder", message="Previous hash does not match predecessor.", ) - expected_node_hash = compute_node_hash(node.content, node.previous_hash) + expected_node_hash = compute_node_hash(node.content, node.previous_hash, node.tier) if node.node_hash != expected_node_hash: return ChainVerification( valid=False, diff --git a/tests/governance/test_certificate.py b/tests/governance/test_certificate.py index 0b7a44ef9..f556de173 100644 --- a/tests/governance/test_certificate.py +++ b/tests/governance/test_certificate.py @@ -47,6 +47,16 @@ 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() diff --git a/tests/governance/test_evidence.py b/tests/governance/test_evidence.py index dd016b2f2..ff50866ce 100644 --- a/tests/governance/test_evidence.py +++ b/tests/governance/test_evidence.py @@ -138,3 +138,18 @@ def test_detects_tampered_node(self): 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_governing_context.py b/tests/governance/test_governing_context.py index 8d7ab4647..a3a49e030 100644 --- a/tests/governance/test_governing_context.py +++ b/tests/governance/test_governing_context.py @@ -96,6 +96,16 @@ def test_no_policies_all_allowed(self): 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(