feat(governance): minimal gates + evidence + TaskCertificate slice (#1393)

lionagi/governance/__init__.py

@@ -0,0 +1,51 @@
+# Copyright (c) 2023-2026, HaiyangLi <quantocean.li at gmail dot com>
+# 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",
+]

lionagi/governance/certificate.py

@@ -0,0 +1,78 @@
+# Copyright (c) 2023-2026, HaiyangLi <quantocean.li at gmail dot com>
+# 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 {},
+        )

lionagi/governance/context.py

@@ -0,0 +1,126 @@
+# Copyright (c) 2023-2026, HaiyangLi <quantocean.li at gmail dot com>
+# 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

lionagi/governance/errors.py

@@ -0,0 +1,19 @@
+# Copyright (c) 2023-2026, HaiyangLi <quantocean.li at gmail dot com>
+# 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}")