feat: implement event-driven governance across agent, CDK, and CLI (#230)

Add declarative event rules with sync gates in the agent runtime and async
evaluation in FanOut, plus contracts, API/CLI surfaces, and platform-default
rule pack resolution frozen on task creation.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: bgagent

agent/src/event_governance/__init__.py

@@ -0,0 +1,6 @@
+"""Event-driven governance (issue #230)."""
+
+from event_governance.coordinator import evaluate_sync_event
+from event_governance.evaluator import EventRule, match_rules
+
+__all__ = ["EventRule", "evaluate_sync_event", "match_rules"]

agent/src/event_governance/catalog.py

@@ -0,0 +1,37 @@
+"""Load the normative event catalog from contracts/."""
+
+from __future__ import annotations
+
+import json
+from functools import lru_cache
+from pathlib import Path
+from typing import Any
+
+
+def _catalog_path() -> Path:
+    here = Path(__file__).resolve()
+    for base in (here.parents[2], here.parents[1]):
+        candidate = base / "contracts" / "event-catalog" / "v1.json"
+        if candidate.is_file():
+            return candidate
+    deployed = Path("/app/contracts/event-catalog/v1.json")
+    if deployed.is_file():
+        return deployed
+    raise FileNotFoundError("event catalog v1.json not found")
+
+
+@lru_cache(maxsize=1)
+def load_catalog() -> dict[str, Any]:
+    """Return the parsed event catalog (cached)."""
+    with _catalog_path().open(encoding="utf-8") as fh:
+        return json.load(fh)
+
+
+def is_known_event(name: str) -> bool:
+    """Return True when ``name`` appears in the catalog."""
+    cat = load_catalog()
+    return (
+        name in cat.get("top_level_event_types", [])
+        or name in cat.get("agent_milestones", [])
+        or name in cat.get("checkpoints", [])
+    )

agent/src/event_governance/coordinator.py

@@ -0,0 +1,73 @@
+"""Sync event governance coordinator — evaluate rules at pipeline checkpoints."""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+from event_governance.engine import build_policy_engine
+from event_governance.evaluator import match_rules, parse_rules
+from event_governance.gate import EventGateResult, gate_on_event_async
+from event_governance.writer import policy_decision_metadata, write_policy_decision
+from shell import log
+
+
+def _run_async(coro: Any) -> Any:
+    try:
+        asyncio.get_running_loop()
+    except RuntimeError:
+        return asyncio.run(coro)
+    raise RuntimeError("evaluate_sync_event cannot run inside a running event loop")
+
+
+def evaluate_sync_event(
+    *,
+    rules_raw: list[dict[str, Any]] | None,
+    event_type: str,
+    metadata: dict[str, Any],
+    progress: Any,
+    rule_pack_id: str | None = None,
+    config: Any = None,
+    engine: Any = None,
+    task_id: str | None = None,
+    user_id: str | None = None,
+) -> EventGateResult | None:
+    """Evaluate sync rules for one event. Returns gate result when enforce blocks."""
+    rules = parse_rules(rules_raw, rule_pack_id=rule_pack_id)
+    matched = match_rules(rules, event_type=event_type, metadata=metadata, evaluation="sync")
+    if not matched:
+        return None
+
+    blocked: EventGateResult | None = None
+    for rule in matched:
+        enforce = rule.mode == "enforce"
+        meta = policy_decision_metadata(
+            rule=rule,
+            event_type=event_type,
+            metadata=metadata,
+            enforce=enforce,
+        )
+        write_policy_decision(progress, event_type="policy_decision", metadata=meta)
+
+        if rule.action == "require_approval" and enforce:
+            resolved_task_id = task_id or (config.task_id if config else None)
+            if engine is None and config is not None:
+                engine = build_policy_engine(config)
+            if engine is None or not resolved_task_id:
+                log("WARN", f"Event rule {rule.id} enforce blocked but engine/task_id missing")
+                return EventGateResult(allowed=False, reason="approval system unavailable")
+            result = _run_async(
+                gate_on_event_async(
+                    rule=rule,
+                    event_type=event_type,
+                    metadata=metadata,
+                    engine=engine,
+                    task_id=resolved_task_id,
+                    user_id=user_id or (config.user_id if config else None),
+                    progress=progress,
+                )
+            )
+            if not result.allowed:
+                return result
+            blocked = result
+    return blocked

agent/src/event_governance/engine.py

@@ -0,0 +1,32 @@
+"""Build PolicyEngine from TaskConfig for event governance gates."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from models import TaskConfig
+    from policy import PolicyEngine
+
+
+def build_policy_engine(config: TaskConfig) -> PolicyEngine:
+    """Mirror runner.py PolicyEngine construction for checkpoint gates."""
+    from policy import PolicyEngine
+
+    engine_kwargs: dict = {}
+    if config.initial_approvals:
+        engine_kwargs["initial_approvals"] = list(config.initial_approvals)
+    if config.approval_timeout_s is not None:
+        engine_kwargs["task_default_timeout_s"] = config.approval_timeout_s
+    if config.initial_approval_gate_count:
+        engine_kwargs["initial_approval_gate_count"] = config.initial_approval_gate_count
+    if config.approval_gate_cap is not None:
+        engine_kwargs["approval_gate_cap"] = config.approval_gate_cap
+    cedar = config.cedar_policies if config.cedar_policies else None
+    return PolicyEngine(
+        task_type=config.policy_principal,
+        repo=config.repo_url,
+        read_only=config.read_only,
+        extra_policies=cedar,
+        **engine_kwargs,
+    )

agent/src/event_governance/evaluator.py

@@ -0,0 +1,140 @@
+"""Declarative event rule matching (sync + async parity with CDK evaluator)."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass(frozen=True)
+class EventRule:
+    """One event governance rule from blueprint or registry pack."""
+
+    id: str
+    on: str
+    action: str
+    mode: str
+    evaluation: str
+    when: dict[str, Any] = field(default_factory=dict)
+    reason: str = ""
+    severity: str = "medium"
+    timeout_s: int | None = None
+    notify_channels: list[str] = field(default_factory=list)
+    rule_pack_id: str | None = None
+
+
+def _event_name(event_type: str, metadata: dict[str, Any]) -> str:
+    """Resolve the rule ``on`` key for an incoming event."""
+    if event_type == "agent_milestone":
+        milestone = metadata.get("milestone")
+        if isinstance(milestone, str):
+            return milestone
+    checkpoint = metadata.get("checkpoint")
+    if isinstance(checkpoint, str):
+        return checkpoint
+    return event_type
+
+
+def _fields_match(rule_when: dict[str, Any], metadata: dict[str, Any]) -> bool:
+    expected = rule_when.get("fields")
+    if not expected:
+        return True
+    if not isinstance(expected, dict):
+        return False
+    for key, want in expected.items():
+        got = metadata.get(key)
+        if got != want:
+            return False
+    return True
+
+
+def _aggregate_match(
+    rule_when: dict[str, Any],
+    metadata: dict[str, Any],
+    aggregate_state: dict[str, Any] | None,
+) -> bool:
+    agg = rule_when.get("aggregate")
+    if not agg:
+        return True
+    if not isinstance(agg, dict):
+        return False
+    cost_gte = agg.get("cost_usd_gte")
+    if cost_gte is not None:
+        cumulative = aggregate_state.get("cumulative_cost_usd") if aggregate_state else None
+        if cumulative is None:
+            cumulative = metadata.get("cumulative_cost_usd")
+        if cumulative is None:
+            return False
+        try:
+            if float(cumulative) < float(cost_gte):
+                return False
+        except (TypeError, ValueError):
+            return False
+    turn_gte = agg.get("turn_count_gte")
+    if turn_gte is not None:
+        turns = aggregate_state.get("turn_count") if aggregate_state else None
+        if turns is None:
+            turns = metadata.get("turn_count")
+        if turns is None:
+            return False
+        try:
+            if float(turns) < float(turn_gte):
+                return False
+        except (TypeError, ValueError):
+            return False
+    return True
+
+
+def match_rules(
+    rules: list[EventRule],
+    *,
+    event_type: str,
+    metadata: dict[str, Any],
+    evaluation: str | None = None,
+    aggregate_state: dict[str, Any] | None = None,
+) -> list[EventRule]:
+    """Return rules that match the given event."""
+    name = _event_name(event_type, metadata)
+    matched: list[EventRule] = []
+    for rule in rules:
+        if rule.on != name:
+            continue
+        if evaluation is not None and rule.evaluation != evaluation:
+            continue
+        when = rule.when or {}
+        if not _fields_match(when, metadata):
+            continue
+        if not _aggregate_match(when, metadata, aggregate_state):
+            continue
+        matched.append(rule)
+    return matched
+
+
+def parse_rules(
+    raw: list[dict[str, Any]] | None,
+    *,
+    rule_pack_id: str | None = None,
+) -> list[EventRule]:
+    """Parse blueprint/registry rule dicts into ``EventRule`` instances."""
+    if not raw:
+        return []
+    out: list[EventRule] = []
+    for item in raw:
+        if not isinstance(item, dict) or not item.get("id"):
+            continue
+        out.append(
+            EventRule(
+                id=str(item["id"]),
+                on=str(item["on"]),
+                action=str(item.get("action", "observe_only")),
+                mode=str(item.get("mode", "observe_only")),
+                evaluation=str(item.get("evaluation", "sync")),
+                when=dict(item.get("when") or {}),
+                reason=str(item.get("reason") or ""),
+                severity=str(item.get("severity") or "medium"),
+                timeout_s=item.get("timeout_s"),
+                notify_channels=list(item.get("notify_channels") or []),
+                rule_pack_id=rule_pack_id or item.get("rule_pack_id"),
+            )
+        )
+    return out

agent/src/event_governance/gate.py

@@ -0,0 +1,60 @@
+"""Event-sourced approval gates (Phase 2 enforce mode)."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from event_governance.evaluator import EventRule
+
+from hooks import _handle_require_approval
+from policy import PolicyDecision
+
+
+@dataclass
+class EventGateResult:
+    """Outcome of a sync event gate."""
+
+    allowed: bool
+    reason: str = ""
+
+
+async def gate_on_event_async(
+    *,
+    rule: EventRule,
+    event_type: str,
+    metadata: dict[str, Any],
+    engine: Any,
+    task_id: str | None,
+    user_id: str | None,
+    progress: Any,
+    ts_module: Any = None,
+) -> EventGateResult:
+    """Block on human approval for an event rule (enforce + require_approval)."""
+    checkpoint = metadata.get("checkpoint") or rule.on
+    tool_input = {"checkpoint": checkpoint, "event_type": event_type, **metadata}
+    decision = PolicyDecision.require_approval(
+        reason=rule.reason or f"Approval required at {checkpoint}",
+        severity=rule.severity,
+        timeout_s=rule.timeout_s or engine.task_default_timeout_s,
+        matching_rule_ids=(rule.id,),
+    )
+    tool_name = f"event:{checkpoint}"
+    response = await _handle_require_approval(
+        decision=decision,
+        tool_name=tool_name,
+        tool_input=tool_input,
+        engine=engine,
+        task_id=task_id,
+        user_id=user_id,
+        progress=progress,
+        ts=ts_module,
+        approval_source="event",
+        event_type=event_type,
+        event_checkpoint=str(checkpoint),
+        rule_id=rule.id,
+    )
+    allowed = response.get("hookSpecificOutput", {}).get("permissionDecision") == "allow"
+    reason = response.get("hookSpecificOutput", {}).get("permissionDecisionReason", "")
+    return EventGateResult(allowed=allowed, reason=reason or "")