feat: event-driven governance (sync + async event rules) (#230)

agent/event-rules-parity/README.md

@@ -0,0 +1,11 @@
+# Event rule evaluator parity fixtures
+
+Golden-file vectors shared by:
+
+- `agent/tests/test_event_rules_parity.py` (Python evaluator)
+- `cdk/test/handlers/shared/event-rules-parity.test.ts` (TypeScript evaluator)
+
+Fixtures live under `agent/event-rules/fixtures/` and are referenced by
+basename from this parity contract.
+
+Design reference: `docs/design/EVENT_GOVERNANCE.md` (issue #230).

agent/event-rules/README.md

@@ -0,0 +1,29 @@
+# Event governance rules
+
+Event rule packs, schema, and cross-language parity fixtures for event-driven
+governance (issue #230). Authored here alongside `agent/workflows/` — one asset
+tree, one authoring UX. The related event-catalog and policy-decision schemas
+stay under the repo-root `contracts/` tree.
+
+## Layout
+
+| Path | Purpose |
+|------|---------|
+| `schema.json` | JSON Schema for event rule packs |
+| `packs/*.json` | Versioned rule packs pinned by `Blueprint.security.eventRulePack` |
+| `fixtures/` | Golden fixtures for Python + TypeScript evaluator parity |
+
+## Consumers
+
+| Caller | Path | Reads |
+|--------|------|-------|
+| `cdk/src/handlers/shared/event-rule-pack-resolver.ts` | Bundles `packs/*.json` at build time (future `RegistryService.resolve`) | packs |
+| `cdk/test/.../event-rule-evaluator.test.ts`, `event-rules-parity.test.ts` | TypeScript evaluator + parity | fixtures |
+| `agent/tests/test_event_rules_parity.py`, `test_event_governance_evaluator.py` | Python evaluator + parity | fixtures |
+| `cli/src/commands/rules.ts` | `bgagent rules eval --fixture` | fixtures |
+
+The agent runtime does **not** read these files: it receives already-resolved
+`event_rules` frozen on the TaskRecord at submit time, so both evaluation planes
+consume one source (the CDK-resolved rules).
+
+Design reference: [EVENT_GOVERNANCE.md](../../docs/design/EVENT_GOVERNANCE.md).

agent/event-rules/fixtures/aggregate-cost-cancel.json

@@ -0,0 +1,28 @@
+{
+  "description": "Aggregate cost ceiling triggers cancel",
+  "event": {
+    "event_type": "agent_cost_update",
+    "metadata": {
+      "cumulative_cost_usd": 26.5
+    }
+  },
+  "rules": [
+    {
+      "id": "cost-ceiling",
+      "on": "agent_cost_update",
+      "when": {
+        "aggregate": {
+          "cost_usd_gte": 25
+        }
+      },
+      "action": "cancel_task",
+      "mode": "enforce",
+      "evaluation": "async",
+      "reason": "Cumulative cost exceeded $25"
+    }
+  ],
+  "expected_matches": ["cost-ceiling"],
+  "aggregate_state": {
+    "cumulative_cost_usd": 26.5
+  }
+}

agent/event-rules/fixtures/aggregate-turn-count-escalate.json

@@ -0,0 +1,28 @@
+{
+  "description": "Aggregate turn-count ceiling triggers escalate",
+  "event": {
+    "event_type": "agent_turn",
+    "metadata": {
+      "turn_count": 40
+    }
+  },
+  "rules": [
+    {
+      "id": "turn-ceiling",
+      "on": "agent_turn",
+      "when": {
+        "aggregate": {
+          "turn_count_gte": 30
+        }
+      },
+      "action": "escalate",
+      "mode": "enforce",
+      "evaluation": "async",
+      "reason": "Turn count exceeded 30"
+    }
+  ],
+  "expected_matches": ["turn-ceiling"],
+  "aggregate_state": {
+    "turn_count": 40
+  }
+}

agent/event-rules/fixtures/async-notify-pr.json

@@ -0,0 +1,21 @@
+{
+  "description": "Async notify on pr_created",
+  "event": {
+    "event_type": "agent_milestone",
+    "metadata": {
+      "milestone": "pr_created",
+      "pr_url": "https://github.com/org/repo/pull/1"
+    }
+  },
+  "rules": [
+    {
+      "id": "notify-on-pr",
+      "on": "pr_created",
+      "action": "notify",
+      "mode": "enforce",
+      "evaluation": "async",
+      "notify_channels": ["slack"]
+    }
+  ],
+  "expected_matches": ["notify-on-pr"]
+}

agent/event-rules/fixtures/observe-repo-setup.json

@@ -0,0 +1,27 @@
+{
+  "description": "Sync observe-only rule fires on repo_setup_complete",
+  "event": {
+    "event_type": "agent_milestone",
+    "metadata": {
+      "milestone": "repo_setup_complete",
+      "workflow_ref": "coding/new-task-v1"
+    }
+  },
+  "rules": [
+    {
+      "id": "plan-review-before-code",
+      "on": "repo_setup_complete",
+      "when": {
+        "fields": {
+          "workflow_ref": "coding/new-task-v1"
+        }
+      },
+      "action": "require_approval",
+      "mode": "observe_only",
+      "evaluation": "sync",
+      "reason": "Plan review before execution",
+      "severity": "medium"
+    }
+  ],
+  "expected_matches": ["plan-review-before-code"]
+}

agent/event-rules/packs/platform-default-v1.json

@@ -0,0 +1,23 @@
+{
+  "pack_id": "platform-default",
+  "pack_version": "1.0.0",
+  "rules": [
+    {
+      "id": "observe-repo-setup",
+      "on": "repo_setup_complete",
+      "action": "require_approval",
+      "mode": "observe_only",
+      "evaluation": "sync",
+      "reason": "Plan review before execution (platform default — observe)",
+      "severity": "medium"
+    },
+    {
+      "id": "notify-on-pr",
+      "on": "pr_created",
+      "action": "notify",
+      "mode": "enforce",
+      "evaluation": "async",
+      "notify_channels": ["slack"]
+    }
+  ]
+}

agent/event-rules/schema.json

@@ -0,0 +1,76 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$id": "https://abca.dev/agent/event-rules/schema.json",
+  "title": "EventRulePack",
+  "type": "object",
+  "required": ["rules"],
+  "properties": {
+    "pack_id": { "type": "string" },
+    "pack_version": { "type": "string" },
+    "rules": {
+      "type": "array",
+      "items": { "$ref": "#/$defs/EventRule" }
+    }
+  },
+  "$defs": {
+    "EventRule": {
+      "type": "object",
+      "required": ["id", "on", "action", "mode", "evaluation"],
+      "properties": {
+        "id": { "type": "string", "minLength": 1 },
+        "on": { "type": "string", "minLength": 1 },
+        "when": {
+          "type": "object",
+          "properties": {
+            "fields": {
+              "type": "object",
+              "additionalProperties": true
+            },
+            "aggregate": {
+              "type": "object",
+              "properties": {
+                "cost_usd_gte": { "type": "number", "minimum": 0 },
+                "turn_count_gte": { "type": "integer", "minimum": 1 }
+              },
+              "additionalProperties": false
+            }
+          },
+          "additionalProperties": false
+        },
+        "action": {
+          "type": "string",
+          "enum": [
+            "require_approval",
+            "notify",
+            "escalate",
+            "cancel_task",
+            "inject_nudge",
+            "observe_only"
+          ]
+        },
+        "mode": {
+          "type": "string",
+          "enum": ["observe_only", "enforce"]
+        },
+        "evaluation": {
+          "type": "string",
+          "enum": ["sync", "async"]
+        },
+        "reason": { "type": "string" },
+        "severity": {
+          "type": "string",
+          "enum": ["low", "medium", "high"]
+        },
+        "timeout_s": { "type": "integer", "minimum": 30, "maximum": 3600 },
+        "notify_channels": {
+          "type": "array",
+          "items": {
+            "type": "string",
+            "enum": ["slack", "email", "github", "linear", "webhook"]
+          }
+        }
+      },
+      "additionalProperties": false
+    }
+  }
+}

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