diff --git a/pyproject.toml b/pyproject.toml
index b85ca49..6c17f8d 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -5,7 +5,7 @@ description = "Runtime abstractions and interfaces for building agents and autom
 readme = { file = "README.md", content-type = "text/markdown" }
 requires-python = ">=3.11"
 dependencies = [
-  "uipath-core>=0.5.25, <0.6.0",
+  "uipath-core>=0.5.28, <0.6.0",
   "pyyaml>=6.0, <7.0",
   "vaderSentiment>=3.3.2, <4.0",
   "chardet>=5.2.0, <8.0",
diff --git a/src/uipath/runtime/governance/_audit/base.py b/src/uipath/runtime/governance/_audit/base.py
index a8a47b2..aa016e1 100644
--- a/src/uipath/runtime/governance/_audit/base.py
+++ b/src/uipath/runtime/governance/_audit/base.py
@@ -20,10 +20,12 @@
 from abc import ABC, abstractmethod
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
-from typing import Any
+from typing import Any, Callable
 
 from uipath.core.governance import EnforcementMode
 
+from .metadata import GovernanceRuntimeMetadata
+
 logger = logging.getLogger(__name__)
 
 
@@ -226,14 +228,31 @@ class AuditManager:
     # Trip a sink after this many consecutive emit failures (circuit-breaker).
     _SINK_FAILURE_THRESHOLD = 10
 
-    def __init__(self, register_default_sinks: bool = True) -> None:
+    def __init__(
+        self,
+        register_default_sinks: bool = True,
+        *,
+        track_event: Callable[..., None] | None = None,
+        runtime_metadata: GovernanceRuntimeMetadata | None = None,
+    ) -> None:
         """Initialize the audit manager.
 
         Args:
             register_default_sinks: If True (default), register the
-                always-on ``traces`` sink. Tests that want a bare
-                manager can pass ``False`` and register sinks
-                explicitly.
+                always-on ``traces`` and platform-mandated
+                ``track_events`` sinks. Tests that want a bare manager
+                can pass ``False`` and register sinks explicitly.
+            track_event: Platform-supplied telemetry callable wired by
+                the host. When ``None`` (or any sink-construction
+                error), the ``track_events`` sink is skipped and a
+                warning is logged — the runtime continues so a wiring
+                bug never breaks the agent run.
+            runtime_metadata: Constants stamped on every telemetry
+                event (execution engine, agent type, agent framework,
+                runtime version). Defaults to
+                :class:`GovernanceRuntimeMetadata` () — auto-resolved
+                version + ``unknown`` agent type / framework. The host
+                overrides with concrete values.
         """
         self._sinks: list[AuditSink] = []
         # Guards _sinks, _sink_failures, _tripped_sinks — all read +
@@ -246,6 +265,7 @@ def __init__(self, register_default_sinks: bool = True) -> None:
 
         if register_default_sinks:
             self._register_traces_sink()
+            self._register_track_event_sink(track_event, runtime_metadata)
 
     def _register_traces_sink(self) -> None:
         """Register the always-on ``traces`` sink.
@@ -262,6 +282,48 @@ def _register_traces_sink(self) -> None:
             self.register_sink(sink)
             logger.info("Governance audit sink registered: traces")
 
+    def _register_track_event_sink(
+        self,
+        track_event: Callable[..., None] | None,
+        runtime_metadata: GovernanceRuntimeMetadata | None,
+    ) -> None:
+        """Register the platform-mandated ``track_events`` sink.
+
+        Mirrors :meth:`_register_traces_sink`: deferred import,
+        construct, register, log. The sink is expected to be wired by
+        the host's platform layer.
+
+        Wrapped in a broad ``except`` so a misconfigured wiring layer
+        (missing callable, sink construction error) never crashes the
+        agent — the runtime logs and proceeds without the sink. The
+        sink-level circuit breaker handles per-emit failures
+        separately.
+        """
+        try:
+            from .track_events import TrackEventAuditSink
+
+            if track_event is None:
+                raise ValueError(
+                    "Platform-mandated track_event callable was not supplied; "
+                    "the host wiring layer must pass it to AuditManager(...)."
+                )
+            meta = (
+                runtime_metadata
+                if runtime_metadata is not None
+                else GovernanceRuntimeMetadata()
+            )
+            sink = TrackEventAuditSink(track_event, meta)
+            self.register_sink(sink)
+            logger.info("Governance audit sink registered: track_events")
+        except Exception as exc:  # noqa: BLE001 - registration must not crash the agent
+            # ``str(exc)`` instead of passing ``exc`` directly: the
+            # logging LogRecord retains its ``args`` tuple until the
+            # handler formats the record, and a raw exception there
+            # carries its ``__traceback__`` → frame chain → ``self``,
+            # which would keep the AuditManager alive in any
+            # log-capturing test. Stringifying breaks that ref.
+            logger.warning("Failed to register track_events sink: %s", str(exc))
+
     def register_sink(self, sink: AuditSink) -> None:
         """Register an audit sink.
 
@@ -385,6 +447,8 @@ def emit_rule_evaluation(
         detail: str = "",
         agent_name: str = "agent",
         description: str = "",
+        duration_ms: float = 0.0,
+        mapped_to_uipath: bool = False,
     ) -> None:
         """Convenience method to emit a rule evaluation event.
 
@@ -393,6 +457,10 @@ def emit_rule_evaluation(
         its own mode — parallel runtimes can run in different modes
         simultaneously, and a process-global wouldn't be authoritative
         for any of them.
+
+        ``duration_ms`` and ``mapped_to_uipath`` are stamped on the
+        event's data dict for telemetry sinks (e.g.
+        :class:`TrackEventAuditSink`); the OTel traces sink ignores them.
         """
         self.emit(
             AuditEvent(
@@ -409,6 +477,8 @@ def emit_rule_evaluation(
                     "detail": detail,
                     "description": description,
                     "status": "MATCHED" if matched else "PASS",
+                    "duration_ms": duration_ms,
+                    "mapped_to_uipath": mapped_to_uipath,
                 },
             )
         )
@@ -421,8 +491,41 @@ def emit_hook_summary(
         matched_rules: int,
         final_action: str,
         enforcement_mode: EnforcementMode,
+        duration_ms: float = 0.0,
+        skipped_policy_names: list[str] | None = None,
+        guardrail_dispatched_count: int = 0,
+        denied_count: int | None = None,
     ) -> None:
-        """Convenience method to emit a hook summary event."""
+        """Convenience method to emit a hook summary event.
+
+        ``matched_rules`` keeps its historical meaning — any rule whose
+        checks matched, regardless of the configured action — for
+        backward compatibility with existing sinks. The newer
+        ``denied_count`` captures only the rules the evaluator actually
+        wanted to act on (matched **and** configured action ≠
+        ``allow``). A matched rule whose action is ``allow`` is a
+        positive informational match and is folded into
+        ``passed_count``, not ``denied_count``.
+
+        Args:
+            duration_ms: Total wall time spent evaluating this hook.
+            skipped_policy_names: Rules in the pack that were not
+                evaluated (currently disabled). The summary carries
+                their ids so operators can spot which policies a tenant
+                turned off.
+            guardrail_dispatched_count: How many UiPath-mapped
+                guardrail-fallback rules fired for this hook and were
+                handed to the compensating path. Lets dashboards
+                compute the native-vs-dispatched ratio.
+            denied_count: Rules that matched **and** would have acted
+                (action ∈ {``deny``, ``escalate``, ``audit``}). When
+                ``None`` (legacy callers), falls back to
+                ``matched_rules`` so the old "matched == denial"
+                semantic is preserved.
+        """
+        skipped = list(skipped_policy_names or [])
+        actual_denied = denied_count if denied_count is not None else matched_rules
+        passed_count = max(total_rules - actual_denied, 0)
         self.emit(
             AuditEvent(
                 event_type=EventType.HOOK_END,
@@ -433,6 +536,12 @@ def emit_hook_summary(
                     "matched_rules": matched_rules,
                     "final_action": final_action,
                     "enforcement_mode": enforcement_mode,
+                    "duration_ms": duration_ms,
+                    "passed_count": passed_count,
+                    "denied_count": actual_denied,
+                    "skipped_count": len(skipped),
+                    "skipped_policy_names": skipped,
+                    "guardrail_dispatched_count": guardrail_dispatched_count,
                 },
             )
         )
diff --git a/src/uipath/runtime/governance/_audit/metadata.py b/src/uipath/runtime/governance/_audit/metadata.py
new file mode 100644
index 0000000..068a8a6
--- /dev/null
+++ b/src/uipath/runtime/governance/_audit/metadata.py
@@ -0,0 +1,69 @@
+"""Per-runtime metadata stamped on every governance telemetry event.
+
+The host constructs a :class:`GovernanceRuntimeMetadata` once per
+agent run and passes it to the telemetry sink. Every event produced
+by :class:`TrackEventAuditSink` carries these fields so downstream
+consumers can pivot on engine / agent type / framework / runtime
+version.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from importlib.metadata import PackageNotFoundError, version
+
+NATIVE_EXECUTION_ENGINE = "uipath_native_governance_checker"
+
+
+def _resolve_runtime_version() -> str:
+    """Read the ``uipath-runtime`` package version, or ``"unknown"``.
+
+    ``importlib.metadata.version`` fails when the package is imported
+    from a source checkout that was never installed (CI fixtures,
+    editable installs with stripped metadata). Telemetry must keep
+    flowing in those cases, so the fallback is a sentinel rather than
+    a raise.
+    """
+    try:
+        return version("uipath-runtime")
+    except PackageNotFoundError:
+        return "unknown"
+
+
+@dataclass(frozen=True)
+class GovernanceRuntimeMetadata:
+    """Constants stamped on every governance telemetry event.
+
+    Attributes:
+        execution_engine: Implementation behind the evaluator. Default
+            ``"uipath_native_governance_checker"``. When a future engine
+            (e.g. AGT) replaces the native checker, the host supplies
+            its own identifier here so the emitted event records which
+            engine produced the verdict.
+        agent_type: Category of agent under governance — e.g.
+            ``"uipath_coded"``, ``"uipath_lowcode"``, ``"servicenow"``,
+            or any other identifier the host wants to attach. External
+            agents (ServiceNow, etc.) join this taxonomy when they
+            land. ``"unknown"`` keeps telemetry flowing if the host
+            forgets to set it.
+        agent_framework: Framework that drives the agent — e.g.
+            ``"langchain"``, ``"openai_agents"``, ``"llamaindex"``,
+            ``"google_adk"``, ``"agent_framework"``, ``"mcp"``.
+        runtime_version: ``uipath-runtime`` package version. Resolved
+            from installed package metadata at construction; falls
+            back to ``"unknown"`` for source checkouts.
+    """
+
+    execution_engine: str = NATIVE_EXECUTION_ENGINE
+    agent_type: str = "unknown"
+    agent_framework: str = "unknown"
+    runtime_version: str = field(default_factory=_resolve_runtime_version)
+
+    def as_payload(self) -> dict[str, str]:
+        """Return the metadata as a dict ready to merge into an event payload."""
+        return {
+            "execution_engine": self.execution_engine,
+            "agent_type": self.agent_type,
+            "agent_framework": self.agent_framework,
+            "runtime_version": self.runtime_version,
+        }
diff --git a/src/uipath/runtime/governance/_audit/track_events.py b/src/uipath/runtime/governance/_audit/track_events.py
new file mode 100644
index 0000000..8a2b745
--- /dev/null
+++ b/src/uipath/runtime/governance/_audit/track_events.py
@@ -0,0 +1,329 @@
+"""Telemetry sink for governance evaluation events.
+
+Submits custom telemetry events through a caller-supplied **non-blocking**
+``track_event`` callable. The runtime layer does not own dispatch — the
+host is responsible for supplying a callable that returns immediately
+(typically a thread-pool-backed adapter around the real telemetry
+client). Tests pass a mock that captures payloads.
+
+Event-rate policy (volume control):
+
+- DENIED rules (``matched=True``) → one ``governance.rule.denied`` event
+  per matched rule.
+- PASSED + SKIPPED rules → one ``governance.hook.summary`` event per
+  hook with counts + the names of skipped rules + the count of
+  UiPath-mapped guardrails dispatched through the compensating path.
+
+Why: a 50-rule pack evaluated on every hook of every agent step would
+multiply per-step telemetry calls by 50. Bundling the "nothing
+happened" cases into a single hook summary cuts that to 1.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Callable
+
+from uipath.core.governance import EnforcementMode
+
+from .base import AuditEvent, AuditSink, EventType
+from .metadata import GovernanceRuntimeMetadata
+
+logger = logging.getLogger(__name__)
+
+
+TrackEventCallable = Callable[..., None]
+"""Host-supplied telemetry callable.
+
+Expected kwargs: ``event_name: str``, ``data: dict | None``,
+``operation_id: str | None``. The sink calls this synchronously from
+the caller's hook thread, so the callable **MUST** be non-blocking —
+the agent run shares that thread, and any wait propagates straight
+into latency the agent observes. Hosts that need to do network I/O
+are expected to wrap the underlying call in a thread-pool-backed
+adapter that submits the work and returns immediately.
+"""
+
+
+EVENT_RULE_DENIED = "governance.rule.denied"
+EVENT_HOOK_SUMMARY = "governance.hook.summary"
+
+
+def _resolve_operation_id() -> str | None:
+    """Read the current OTel span's trace id for downstream correlation.
+
+    Sink dispatch runs on the caller's thread (see
+    :meth:`AuditManager.emit`), so the live agent span is visible via
+    ``trace.get_current_span()`` directly. Rendered as a 32-char
+    lowercase hex string for use as a correlation id by the consumer.
+
+    Returns ``None`` when:
+
+    - OpenTelemetry isn't installed (sink no-ops gracefully).
+    - No span is recording (no trace context to correlate against).
+
+    The live span is the single source of truth for trace correlation;
+    the sink does not consult environment variables.
+    """
+    try:
+        from opentelemetry import trace
+    except ImportError:
+        return None
+
+    span = trace.get_current_span()
+    span_ctx = span.get_span_context()
+    if not span_ctx.is_valid:
+        return None
+    return f"{span_ctx.trace_id:032x}"
+
+
+def _mode_str(mode: Any) -> str:
+    """Coerce an enforcement-mode field to its canonical uppercase string."""
+    if isinstance(mode, EnforcementMode):
+        return mode.value.upper()
+    if isinstance(mode, str):
+        return mode.upper()
+    return "AUDIT"
+
+
+def _evaluator_result(action: str) -> str:
+    """Map a rule's configured action to the spec-vocabulary verdict.
+
+    Mirrors :func:`uipath.runtime.governance._audit.traces._derive_results`
+    so both sinks agree on the (matched, action) → verdict mapping.
+    """
+    action_lc = action.lower()
+    if action_lc == "deny":
+        return "DENY"
+    if action_lc == "escalate":
+        return "HITL"
+    if action_lc == "audit":
+        # The rule wanted to deny but was tagged audit-only at the
+        # check level — the evaluator's intent is still "deny".
+        return "DENY"
+    return "ALLOW"
+
+
+def _action_applied(evaluator_result: str, configured_action: str, mode: str) -> str:
+    """Mode-adjusted action: AUDIT mode collapses DENY/HITL into AUDIT."""
+    if mode == "AUDIT":
+        if evaluator_result in ("DENY", "HITL"):
+            return "AUDIT"
+        return "NONE"
+    # ENFORCE mode: per-rule ``audit`` override stays AUDIT.
+    if configured_action.lower() == "audit":
+        return "AUDIT"
+    return evaluator_result if evaluator_result != "ALLOW" else "NONE"
+
+
+class TrackEventAuditSink(AuditSink):
+    """Sink that emits governance telemetry via the injected ``track_event``.
+
+    Volume-controlled per the module docstring: individual events only
+    for denials, one aggregated event per hook for passed / skipped /
+    dispatched. Other event types are dropped at :meth:`accepts`.
+
+    The ``track_event`` callable is invoked synchronously on the
+    caller's hook thread, so the callable must be non-blocking — see
+    the :class:`TrackEventCallable` type alias for the contract the
+    host is expected to satisfy.
+    """
+
+    SINK_NAME = "track_events"
+
+    def __init__(
+        self,
+        track_event: TrackEventCallable,
+        runtime_metadata: GovernanceRuntimeMetadata,
+        *,
+        name: str = SINK_NAME,
+    ) -> None:
+        """Initialize the sink.
+
+        Args:
+            track_event: Host-supplied **non-blocking** telemetry
+                callable. Receives ``event_name``, ``data``,
+                ``operation_id`` kwargs. The sink invokes it
+                synchronously from the agent's hook thread, so any
+                network or otherwise blocking work in the underlying
+                implementation must be offloaded by the host (e.g.,
+                via a thread-pool-backed adapter). Tests pass a
+                capture callable. **Required** — ``None`` is a wiring
+                bug and is rejected here so the manager's registration
+                try/except surfaces it as a clear warning rather than
+                letting a ``TypeError`` fire on the first event.
+            runtime_metadata: Per-run constants stamped on every event
+                (execution engine, agent type, agent framework, runtime
+                version).
+            name: Sink name used for circuit-breaker accounting in the
+                manager. Override only if more than one telemetry sink
+                of this kind is registered against the same manager.
+        """
+        if track_event is None:
+            raise ValueError(
+                "TrackEventAuditSink requires a non-None track_event callable. "
+                "The host is expected to supply a non-blocking adapter — "
+                "wiring a blocking client (e.g., raw HTTP) on the agent's "
+                "hook thread will block the agent on every governance event."
+            )
+        self._track_event = track_event
+        self._meta = runtime_metadata
+        self._name = name
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    def accepts(self, event: AuditEvent) -> bool:
+        """Filter the audit stream down to denials + non-empty hook summaries.
+
+        Drops:
+
+        - **Any event whose mode is** :attr:`EnforcementMode.DISABLED`
+          — governance is off; emit nothing. The evaluator short-circuits
+          before ``_emit_audit`` in this case, so in practice the sink
+          shouldn't see these, but the guard here is belt-and-suspenders
+          against any future emitter that bypasses the short-circuit.
+        - Rules that didn't match (``matched=False``) — rolled into the
+          hook summary's ``passed_count``.
+        - Rules that matched but whose configured ``action`` is
+          ``allow`` — a positive informational match isn't a denial;
+          rolled into the hook summary too.
+        - **Empty hook summaries** — ``total_rules=0`` AND
+          ``skipped_count=0`` means nothing was evaluated and nothing
+          was deliberately skipped. Zero operator value; suppressed.
+          (A summary with ``skipped_count>0`` still fires because the
+          ``skipped_policy_names`` payload is operator-relevant — it
+          shows which policies a tenant turned off.)
+        - Other event types (``hook_start``, ``session_*``, generic
+          violations) — out of scope for this sink's telemetry surface.
+
+        The result is bounded volume: at most one ``rule.denied`` event
+        per matched-and-restrictive rule + at most one
+        ``hook.summary`` per hook end (and only when it carries data).
+        """
+        if _mode_str(event.data.get("enforcement_mode")) == "DISABLED":
+            return False
+
+        if event.event_type == EventType.RULE_EVALUATION:
+            if not event.data.get("matched"):
+                return False
+            action = str(event.data.get("action") or "allow").lower()
+            return action != "allow"
+        if event.event_type == EventType.HOOK_END:
+            data = event.data
+            total = int(data.get("total_rules") or 0)
+            skipped = int(data.get("skipped_count") or 0)
+            return not (total == 0 and skipped == 0)
+        return False
+
+    def emit(self, event: AuditEvent) -> None:
+        """Render the event and dispatch via the injected callable.
+
+        Re-checks ``matched`` for rule evaluations as a defense-in-depth
+        guard. :meth:`accepts` already drops passed rules at the
+        :class:`AuditManager` dispatch boundary, but a direct
+        ``sink.emit(event)`` call (tests, future alternate dispatch
+        paths) must not route a passed rule into the
+        ``governance.rule.denied`` telemetry stream.
+
+        Errors from ``track_event`` propagate to the audit manager,
+        which has its own circuit breaker (10 consecutive failures →
+        sink tripped for the rest of the process). The sink itself
+        doesn't catch — letting the manager track failure rate is the
+        whole point of the sink-failure protocol.
+        """
+        # DISABLED governance = no telemetry, full stop. Same
+        # defense-in-depth rationale as the matched/allow check below:
+        # ``accepts`` drops these at the manager boundary, but a
+        # direct ``sink.emit`` call must not leak a DISABLED-mode
+        # event downstream either.
+        if _mode_str(event.data.get("enforcement_mode")) == "DISABLED":
+            return
+
+        if event.event_type == EventType.RULE_EVALUATION:
+            if not event.data.get("matched"):
+                return
+            action = str(event.data.get("action") or "allow").lower()
+            if action == "allow":
+                # Positive informational match — not a denial. Same
+                # defense-in-depth rationale as the ``matched`` check
+                # above: ``accepts`` drops these at the manager
+                # boundary, but a direct ``sink.emit`` call must not
+                # leak them into the ``rule.denied`` stream.
+                return
+            self._emit_rule_denied(event)
+        elif event.event_type == EventType.HOOK_END:
+            data = event.data
+            total = int(data.get("total_rules") or 0)
+            skipped = int(data.get("skipped_count") or 0)
+            if total == 0 and skipped == 0:
+                # Empty hook — same defense-in-depth as the disabled
+                # mode and matched-allow filters.
+                return
+            self._emit_hook_summary(event)
+        # Other event types are filtered out at accepts(); reaching
+        # here for anything else would be a contract violation by the
+        # manager, not the sink.
+
+    def _common_payload(self, event: AuditEvent) -> dict[str, Any]:
+        """Per-runtime constants + event identifiers stamped on every payload."""
+        payload: dict[str, Any] = dict(self._meta.as_payload())
+        payload["agent_name"] = event.agent_name
+        payload["hook"] = event.hook
+        payload["timestamp"] = event.timestamp.isoformat()
+        return payload
+
+    def _emit_rule_denied(self, event: AuditEvent) -> None:
+        """Emit one ``governance.rule.denied`` per matched rule."""
+        data = event.data
+        mode = _mode_str(data.get("enforcement_mode"))
+        configured_action = str(data.get("action") or "allow")
+        evaluator_result = _evaluator_result(configured_action)
+        action_applied = _action_applied(evaluator_result, configured_action, mode)
+
+        payload = self._common_payload(event)
+        payload.update(
+            {
+                "pack": str(data.get("pack_name") or ""),
+                "clause": str(data.get("policy_id") or ""),
+                "rule_name": str(data.get("rule_name") or ""),
+                "mode": mode,
+                "evaluator_result": evaluator_result,
+                "action_applied": action_applied,
+                "duration_ms": float(data.get("duration_ms") or 0.0),
+                "mapped_to_uipath": bool(data.get("mapped_to_uipath", False)),
+                "detail": str(data.get("detail") or ""),
+            }
+        )
+        self._track_event(
+            event_name=EVENT_RULE_DENIED,
+            data=payload,
+            operation_id=_resolve_operation_id(),
+        )
+
+    def _emit_hook_summary(self, event: AuditEvent) -> None:
+        """Emit one ``governance.hook.summary`` per hook end."""
+        data = event.data
+        mode = _mode_str(data.get("enforcement_mode"))
+
+        payload = self._common_payload(event)
+        payload.update(
+            {
+                "mode": mode,
+                "passed_count": int(data.get("passed_count") or 0),
+                "skipped_count": int(data.get("skipped_count") or 0),
+                "skipped_policy_names": list(data.get("skipped_policy_names") or []),
+                "denied_count": int(data.get("denied_count") or 0),
+                "guardrail_dispatched_count": int(
+                    data.get("guardrail_dispatched_count") or 0
+                ),
+                "duration_ms": float(data.get("duration_ms") or 0.0),
+                "final_action": str(data.get("final_action") or "allow").upper(),
+            }
+        )
+        self._track_event(
+            event_name=EVENT_HOOK_SUMMARY,
+            data=payload,
+            operation_id=_resolve_operation_id(),
+        )
diff --git a/src/uipath/runtime/governance/native/evaluator.py b/src/uipath/runtime/governance/native/evaluator.py
index 71d41dd..769737a 100644
--- a/src/uipath/runtime/governance/native/evaluator.py
+++ b/src/uipath/runtime/governance/native/evaluator.py
@@ -12,6 +12,7 @@
 import logging
 import math
 import re
+import time
 from collections import Counter
 from datetime import datetime, timezone
 from functools import cache, lru_cache
@@ -361,14 +362,27 @@ def evaluate(self, context: CheckContext) -> AuditRecord:
         rules = self._policy_index.get_rules_for_hook(context.hook)
 
         evaluations: list[RuleEvaluation] = []
+        # Per-rule wall time (ms), keyed by rule_id. Forwarded to the
+        # telemetry sink via emit_rule_evaluation(duration_ms=...) so
+        # operators can see which rules are slow.
+        rule_durations: dict[str, float] = {}
+        # Disabled rules — never evaluated, surfaced in the hook
+        # summary's skipped_policy_names so dashboards can spot which
+        # policies a tenant turned off.
+        skipped_policy_ids: list[str] = []
         raw_action = Action.ALLOW  # The action before mode adjustment
         deny_would_fire = False  # Track if DENY would have fired
 
+        hook_start = time.monotonic()
+
         for rule in rules:
             if not rule.enabled:
+                skipped_policy_ids.append(rule.rule_id)
                 continue
 
+            rule_start = time.monotonic()
             evaluation = self._evaluate_rule(rule, context)
+            rule_durations[rule.rule_id] = (time.monotonic() - rule_start) * 1000.0
             evaluations.append(evaluation)
 
             if evaluation.matched:
@@ -384,6 +398,8 @@ def evaluate(self, context: CheckContext) -> AuditRecord:
                 elif eval_action == Action.AUDIT and raw_action == Action.ALLOW:
                     raw_action = Action.AUDIT
 
+        hook_duration_ms = (time.monotonic() - hook_start) * 1000.0
+
         # Apply enforcement mode
         final_action = self._apply_enforcement_mode(raw_action)
 
@@ -403,7 +419,13 @@ def evaluate(self, context: CheckContext) -> AuditRecord:
             metadata=record_metadata,
         )
 
-        self._emit_audit(audit, mode)
+        self._emit_audit(
+            audit,
+            mode,
+            rule_durations=rule_durations,
+            skipped_policy_ids=skipped_policy_ids,
+            hook_duration_ms=hook_duration_ms,
+        )
 
         # For any guardrail mapped to UiPath but currently disabled,
         # dispatch a compensating-governance call via the injected
@@ -469,29 +491,59 @@ def _dispatch_compensation(
                 "Failed to dispatch compensating governance call: %s", exc
             )
 
-    def _emit_audit(self, audit: AuditRecord, mode: EnforcementMode) -> None:
+    def _emit_audit(
+        self,
+        audit: AuditRecord,
+        mode: EnforcementMode,
+        *,
+        rule_durations: dict[str, float] | None = None,
+        skipped_policy_ids: list[str] | None = None,
+        hook_duration_ms: float = 0.0,
+    ) -> None:
         """Emit per-rule and hook-summary events to the injected audit manager.
 
         No-op when no audit manager was supplied at construction. The
         per-runtime :class:`AuditManager` handles sink-level circuit
         breaking; emission errors stay there and never break evaluation.
+
+        Args:
+            audit: Resolved :class:`AuditRecord` produced by
+                :meth:`evaluate`.
+            mode: Active :class:`EnforcementMode`.
+            rule_durations: Per-rule wall time in ms, keyed by
+                ``rule_id``. Looked up when emitting individual
+                rule-evaluation events; unknown ids default to 0.
+            skipped_policy_ids: Ids of rules that were disabled and
+                therefore not evaluated. Folded into the hook
+                summary's ``skipped_policy_names``.
+            hook_duration_ms: Total wall time spent evaluating this
+                hook. Folded into the hook summary's ``duration_ms``.
         """
         manager = self._audit_manager
         if manager is None:
             return
 
         hook_name = audit.hook.name
+        durations = rule_durations or {}
 
         # ``guardrail_fallback`` rules are traced by the compensating
         # path (see :meth:`_dispatch_compensation`), which carries the
         # actual validator verdict. Emitting a Python-side
         # ``rule_evaluation`` event here would produce a duplicate
         # trace carrying no verdict, so filter these rules out of every
-        # event the evaluator emits (per-rule AND the hook summary).
+        # per-rule event the evaluator emits AND out of the hook
+        # summary's passed/denied counts. The summary keeps a separate
+        # ``guardrail_dispatched_count`` so the mapped-vs-unmapped
+        # dimension is still queryable.
         emittable = [
             ev for ev in audit.evaluations
             if not self._is_guardrail_fallback_rule(ev.rule_id)
         ]
+        guardrail_dispatched_count = sum(
+            1
+            for ev in audit.evaluations
+            if ev.matched and self._is_guardrail_fallback_rule(ev.rule_id)
+        )
 
         # No emittable rules means every match this hook was a
         # guardrail-fallback already traced by the compensation path.
@@ -513,12 +565,30 @@ def _emit_audit(self, audit: AuditRecord, mode: EnforcementMode) -> None:
                 detail=evaluation.detail,
                 agent_name=audit.agent_name,
                 description=evaluation.description,
+                duration_ms=durations.get(evaluation.rule_id, 0.0),
+                # Per-rule events the evaluator emits are never for
+                # UiPath-mapped guardrails (those are filtered out of
+                # ``emittable`` above and traced by the compensating
+                # path). Stamping ``False`` keeps the schema stable
+                # for sinks.
+                mapped_to_uipath=False,
             )
 
-        # Derive the summary's final action from the emittable subset
-        # only. ``audit.final_action`` folded in fallback rules whose
-        # verdict travels on the compensation path; including them
-        # here would mix the two trace paths' verdicts.
+        # All summary metrics derive from ``emittable`` only —
+        # ``audit.final_action`` folded in fallback rules whose
+        # verdict travels on the compensation path, so using it here
+        # would mix the two trace paths' verdicts.
+        #
+        # ``matched_rules`` keeps the historical "any check matched"
+        # semantic (used by the legacy traces sink). ``denied_count``
+        # is the precise spec verdict — only rules whose configured
+        # action would actually act (not ``allow``). A matched rule
+        # configured as ``allow`` is an explicit positive match and
+        # rolls into ``passed_count`` via the manager's arithmetic.
+        matched_rules = sum(1 for ev in emittable if ev.matched)
+        denied_count = sum(
+            1 for ev in emittable if ev.matched and ev.action != Action.ALLOW
+        )
         summary_final_action = self._apply_enforcement_mode(
             self._most_restrictive_matched_action(emittable)
         )
@@ -527,9 +597,13 @@ def _emit_audit(self, audit: AuditRecord, mode: EnforcementMode) -> None:
             hook=hook_name,
             agent_name=audit.agent_name,
             total_rules=len(emittable),
-            matched_rules=sum(1 for ev in emittable if ev.matched),
+            matched_rules=matched_rules,
             final_action=summary_final_action.value,
             enforcement_mode=mode,
+            duration_ms=hook_duration_ms,
+            skipped_policy_names=list(skipped_policy_ids or []),
+            guardrail_dispatched_count=guardrail_dispatched_count,
+            denied_count=denied_count,
         )
 
     @staticmethod
diff --git a/src/uipath/runtime/governance/native/guardrail_compensation.py b/src/uipath/runtime/governance/native/guardrail_compensation.py
index f113e1a..096e12b 100644
--- a/src/uipath/runtime/governance/native/guardrail_compensation.py
+++ b/src/uipath/runtime/governance/native/guardrail_compensation.py
@@ -25,6 +25,8 @@
     GovernRequest,
 )
 
+from .._audit.metadata import GovernanceRuntimeMetadata
+
 logger = logging.getLogger(__name__)
 
 
@@ -109,15 +111,26 @@ class GuardrailCompensator:
     never breaks the agent hook.
     """
 
-    def __init__(self, provider: GovernanceCompensationProvider) -> None:
+    def __init__(
+        self,
+        provider: GovernanceCompensationProvider,
+        *,
+        runtime_metadata: GovernanceRuntimeMetadata | None = None,
+    ) -> None:
         """Construct a compensator bound to one provider.
 
         Args:
             provider: Host-supplied
                 :class:`GovernanceCompensationProvider`. ``compensate``
                 is invoked synchronously on the agent's hook thread.
+            runtime_metadata: Per-run identity (agent framework / type /
+                runtime version) stamped onto the ``/runtime/govern`` call
+                so the server can attach it to the rule-denied telemetry it
+                emits. ``None`` leaves the fields unset (server defaults
+                them to ``unknown``).
         """
         self._provider = provider
+        self._runtime_metadata = runtime_metadata
 
     def submit(
         self,
@@ -139,6 +152,7 @@ def submit(
         if not validators:
             return
 
+        meta = self._runtime_metadata
         request = GovernRequest(
             validators=validators,
             rules=rules,
@@ -147,6 +161,9 @@ def submit(
             src_timestamp=src_timestamp,
             agent_name=agent_name,
             runtime_id=runtime_id,
+            agent_framework=meta.agent_framework if meta else None,
+            agent_type=meta.agent_type if meta else None,
+            runtime_version=meta.runtime_version if meta else None,
         )
 
         try:
diff --git a/src/uipath/runtime/governance/runtime.py b/src/uipath/runtime/governance/runtime.py
index be8a6af..3b06543 100644
--- a/src/uipath/runtime/governance/runtime.py
+++ b/src/uipath/runtime/governance/runtime.py
@@ -24,17 +24,28 @@
 hooks (BEFORE_MODEL, AFTER_MODEL, TOOL_CALL, AFTER_TOOL) are fired by
 adapters that observe per-step events.
 
-Trace-id is intentionally **not** carried on this wrapper. The
-governance compensator dispatches synchronously on the caller's
-thread; the injected provider resolves the canonical trace id at
-call time. The runtime layer is fully env-free for this path.
+Trace correlation: :meth:`UiPathGovernedRuntime.execute` and
+:meth:`UiPathGovernedRuntime.stream` open a single OTel span that
+wraps BEFORE_AGENT, the delegate's run (including any framework
+adapter spans), and AFTER_AGENT. The span inherits the host's
+ambient context when one is set, or opens a new root trace
+otherwise. Every governance event emitted under that span shares
+its ``trace_id``, so runtime-layer telemetry and framework-layer
+OTel spans correlate end-to-end. Downstream consumers map that
+``trace_id`` to whatever correlation field they use.
+
+The governance compensator dispatches synchronously on the caller's
+thread; the injected compensation provider resolves the canonical
+trace id at call time. The runtime layer remains env-free for that
+path.
 """
 
 from __future__ import annotations
 
 import json
 import logging
-from typing import Any, AsyncGenerator
+from contextlib import contextmanager
+from typing import Any, AsyncGenerator, Callable, Iterator
 
 from uipath.core.governance import EnforcementMode
 from uipath.core.governance.exceptions import GovernanceBlockException
@@ -54,6 +65,60 @@
 logger = logging.getLogger(__name__)
 
 
+@contextmanager
+def _governance_root_span(agent_name: str, runtime_id: str) -> Iterator[None]:
+    """Open the governance run's root OTel span if OTel is available.
+
+    Every governance event and every framework OTel span emitted
+    inside this block inherits the span's ``trace_id``. The result
+    is one trace per agent run that correlates BEFORE_AGENT,
+    BEFORE_MODEL / AFTER_MODEL, AFTER_AGENT, and any per-rule
+    governance events.
+
+    Behavior matrix:
+
+    - **OTel installed + host opened a parent span**: this becomes a
+      child of the host's span and inherits its ``trace_id`` — the
+      host's outer correlation context is preserved end-to-end.
+    - **OTel installed + no parent span**: this becomes the root
+      span of a fresh trace; everything below it shares the new
+      ``trace_id``.
+    - **OTel not installed**: no-op context manager — the wrapper
+      degrades gracefully (no spans, no correlation, but the
+      runtime still functions).
+
+    Lazy import mirrors
+    :func:`uipath.runtime.governance._audit.track_events._resolve_operation_id`:
+    OTel is a transitive runtime dependency, but the wrapper must
+    not crash if a downstream host strips it.
+    """
+    try:
+        from opentelemetry import trace
+    except ImportError:
+        yield
+        return
+
+    tracer = trace.get_tracer("uipath.runtime.governance")
+    # No explicit ``context=`` → OTel picks up the ambient context.
+    # If the host wrapped this call in its own span, we become its
+    # child (same trace_id). Otherwise we open a root span (new
+    # trace_id).
+    with tracer.start_as_current_span("uipath.governance.run") as span:
+        # Span attributes for downstream consumers. ``agent_name``
+        # and ``runtime_id`` are the primary keys an operator
+        # filters on; ``source`` identifies which producer emitted
+        # the span (mirrors the constant used by
+        # :class:`TracesAuditSink` so consumers stay consistent).
+        if agent_name:
+            span.set_attribute("uipath_governance.agent_name", agent_name)
+        if runtime_id:
+            span.set_attribute("uipath_governance.runtime_id", runtime_id)
+        span.set_attribute(
+            "uipath_governance.source", "governance-runtime-python"
+        )
+        yield
+
+
 def _serialize_payload(payload: Any) -> str:
     """Serialize an agent input / output to a string for evaluator checks.
 
@@ -98,6 +163,7 @@ def __init__(
         evaluator: GovernanceEvaluator | None = None,
         agent_name: str = "",
         runtime_id: str = "",
+        on_dispose: Callable[[], None] | None = None,
     ):
         """Initialize the governance runtime with a resolved policy snapshot.
 
@@ -123,6 +189,16 @@ def __init__(
             runtime_id: Runtime-instance id (conversation id, job id,
                 or a synthetic per-run id). Passed through so
                 per-runtime state routes cleanly.
+            on_dispose: Optional host-supplied cleanup callback invoked
+                from :meth:`dispose` after the delegate has been
+                disposed. Called in a ``finally`` so it runs even when
+                the delegate raises. Lets the host attach any
+                per-runtime teardown (flushing a telemetry dispatcher,
+                closing a session, etc.) to the same lifecycle path
+                the runtime already owns — so callers on every CLI
+                path don't have to remember to run it themselves. The
+                runtime treats it as an opaque ``Callable[[], None]``
+                and does not touch the host's underlying object.
         """
         self._delegate = delegate
         self._policy_index = policy_index
@@ -130,6 +206,7 @@ def __init__(
         self._evaluator = evaluator
         self._agent_name = agent_name
         self._runtime_id = runtime_id
+        self._on_dispose = on_dispose
 
     def _fire_before_agent(self, input: Any) -> None:
         """Fire BEFORE_AGENT when an evaluator is wired; otherwise no-op.
@@ -176,13 +253,20 @@ async def execute(
     ) -> UiPathRuntimeResult:
         """Execute the delegate, firing BEFORE_AGENT / AFTER_AGENT around it.
 
+        Wraps the entire invocation — BEFORE_AGENT, delegate execution
+        (which may itself open framework spans), and AFTER_AGENT — in
+        a single OTel span. The shared ``trace_id`` unifies the
+        runtime-side governance events with any framework-side OTel
+        spans the delegate emits.
+
         AFTER_AGENT fires only on successful return — if the delegate
         raises, there's no output to evaluate.
         """
-        self._fire_before_agent(input)
-        result = await self._delegate.execute(input, options=options)
-        self._fire_after_agent(result)
-        return result
+        with _governance_root_span(self._agent_name, self._runtime_id):
+            self._fire_before_agent(input)
+            result = await self._delegate.execute(input, options=options)
+            self._fire_after_agent(result)
+            return result
 
     async def stream(
         self,
@@ -191,21 +275,37 @@ async def stream(
     ) -> AsyncGenerator[UiPathRuntimeEvent, None]:
         """Stream events from the delegate, firing BEFORE_AGENT first.
 
+        Same root-span wrap as :meth:`execute`. Holding the span
+        across ``async for`` is safe — Python asyncio propagates
+        contextvars (including the OTel current-span pointer) across
+        ``await`` suspensions.
+
         AFTER_AGENT fires once a :class:`UiPathRuntimeResult` event is
         observed in the stream — that's the runtime's contract for
         signalling a completed invocation. Intermediate state events
         pass through untouched.
         """
-        self._fire_before_agent(input)
-        async for event in self._delegate.stream(input, options=options):
-            if isinstance(event, UiPathRuntimeResult):
-                self._fire_after_agent(event)
-            yield event
+        with _governance_root_span(self._agent_name, self._runtime_id):
+            self._fire_before_agent(input)
+            async for event in self._delegate.stream(input, options=options):
+                if isinstance(event, UiPathRuntimeResult):
+                    self._fire_after_agent(event)
+                yield event
 
     async def get_schema(self) -> UiPathRuntimeSchema:
         """Forward schema lookup to the delegate."""
         return await self._delegate.get_schema()
 
     async def dispose(self) -> None:
-        """Forward disposal to the delegate."""
-        await self._delegate.dispose()
+        """Dispose the delegate, then run the caller's cleanup hook.
+
+        Runs ``on_dispose`` (if supplied at construction) in a
+        ``finally`` so host-owned teardown — e.g. flushing a telemetry
+        dispatcher — happens even when the delegate raises. The runtime
+        does not inspect what the hook does; it's an opaque callable.
+        """
+        try:
+            await self._delegate.dispose()
+        finally:
+            if self._on_dispose is not None:
+                self._on_dispose()
diff --git a/tests/test_audit_manager_track_event_wiring.py b/tests/test_audit_manager_track_event_wiring.py
new file mode 100644
index 0000000..0afe9cb
--- /dev/null
+++ b/tests/test_audit_manager_track_event_wiring.py
@@ -0,0 +1,183 @@
+"""Tests for ``AuditManager`` auto-registration of the ``track_events`` sink.
+
+The sink is platform-mandated, like ``traces``. The host wires the
+``track_event`` callable + ``GovernanceRuntimeMetadata`` at
+construction. ``_register_track_event_sink`` mirrors the simple shape
+of ``_register_traces_sink`` — deferred import, construct, register,
+log — wrapped in a broad ``except`` so a misconfigured wiring layer
+(missing callable, sink-construction error) is surfaced as a warning
+instead of crashing the agent.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import pytest
+
+from uipath.runtime.governance._audit.base import AuditManager
+from uipath.runtime.governance._audit.metadata import GovernanceRuntimeMetadata
+from uipath.runtime.governance._audit.track_events import TrackEventAuditSink
+
+
+class _Capture:
+    """Stand-in for ``provider.track_event`` — records calls."""
+
+    def __init__(self) -> None:
+        self.calls: list[dict[str, Any]] = []
+
+    def __call__(self, **kwargs: Any) -> None:
+        self.calls.append(kwargs)
+
+
+# ---------------------------------------------------------------------------
+# Happy path — host wires the callable
+# ---------------------------------------------------------------------------
+
+
+def test_track_event_sink_registered_when_callable_supplied() -> None:
+    mgr = AuditManager(
+        track_event=_Capture(),
+        runtime_metadata=GovernanceRuntimeMetadata(agent_type="uipath_coded"),
+    )
+    try:
+        sinks = mgr.list_sinks()
+        assert TrackEventAuditSink.SINK_NAME in sinks
+        sink = mgr.get_sink(TrackEventAuditSink.SINK_NAME)
+        assert isinstance(sink, TrackEventAuditSink)
+    finally:
+        mgr.close()
+
+
+def test_traces_sink_is_still_registered_alongside() -> None:
+    """track_events doesn't replace traces — both are mandatory."""
+    mgr = AuditManager(
+        track_event=_Capture(),
+        runtime_metadata=GovernanceRuntimeMetadata(),
+    )
+    try:
+        sinks = mgr.list_sinks()
+        assert "traces" in sinks
+        assert TrackEventAuditSink.SINK_NAME in sinks
+    finally:
+        mgr.close()
+
+
+def test_supplied_runtime_metadata_reaches_sink() -> None:
+    mgr = AuditManager(
+        track_event=_Capture(),
+        runtime_metadata=GovernanceRuntimeMetadata(
+            agent_type="uipath_coded", agent_framework="langchain"
+        ),
+    )
+    try:
+        sink = mgr.get_sink(TrackEventAuditSink.SINK_NAME)
+        assert isinstance(sink, TrackEventAuditSink)
+        # Internal: confirm the sink carries the host-supplied metadata.
+        assert sink._meta.agent_type == "uipath_coded"
+        assert sink._meta.agent_framework == "langchain"
+    finally:
+        mgr.close()
+
+
+def test_metadata_defaults_when_not_supplied() -> None:
+    """Helper falls back to ``GovernanceRuntimeMetadata()`` defaults."""
+    mgr = AuditManager(track_event=_Capture())
+    try:
+        sink = mgr.get_sink(TrackEventAuditSink.SINK_NAME)
+        assert isinstance(sink, TrackEventAuditSink)
+        assert sink._meta.agent_type == "unknown"
+        assert sink._meta.agent_framework == "unknown"
+    finally:
+        mgr.close()
+
+
+# ---------------------------------------------------------------------------
+# Missing-callable path — caught by the helper's try/except, logged
+# ---------------------------------------------------------------------------
+
+
+def test_missing_callable_logs_warning_and_skips_sink(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """``track_event=None`` → sink __init__ raises → helper catches + warns."""
+    caplog.set_level(logging.WARNING, logger="uipath.runtime.governance._audit.base")
+    mgr = AuditManager()  # no track_event supplied
+    try:
+        # Traces sink still wired; track_events skipped because sink
+        # construction raised ValueError caught by the helper.
+        assert "traces" in mgr.list_sinks()
+        assert TrackEventAuditSink.SINK_NAME not in mgr.list_sinks()
+
+        warnings = [
+            r.message for r in caplog.records if r.levelno == logging.WARNING
+        ]
+        assert any(
+            "Failed to register track_events sink" in m for m in warnings
+        ), f"expected registration-failure warning, got: {warnings}"
+    finally:
+        mgr.close()
+
+
+def test_sink_construction_error_does_not_crash_manager(
+    caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """An exception inside ``TrackEventAuditSink.__init__`` is swallowed.
+
+    Forces a non-``ValueError`` exception (a raise inside ``__init__``
+    triggered via the runtime_metadata path) to confirm the helper's
+    ``except Exception`` covers the broad case the user asked for, not
+    just the validation error.
+    """
+    caplog.set_level(logging.WARNING, logger="uipath.runtime.governance._audit.base")
+
+    def _boom(*args: Any, **kwargs: Any) -> Any:
+        raise RuntimeError("synthetic registration failure")
+
+    from uipath.runtime.governance._audit import track_events as te
+    monkeypatch.setattr(te, "TrackEventAuditSink", _boom)
+
+    mgr = AuditManager(track_event=_Capture())
+    try:
+        # Helper swallowed the RuntimeError; manager kept the traces sink
+        # and reached a constructed state without crashing.
+        assert "traces" in mgr.list_sinks()
+        assert TrackEventAuditSink.SINK_NAME not in mgr.list_sinks()
+        warnings = [
+            r.message for r in caplog.records if r.levelno == logging.WARNING
+        ]
+        assert any(
+            "Failed to register track_events sink" in m
+            and "synthetic registration failure" in m
+            for m in warnings
+        ), f"expected wrapped registration failure, got: {warnings}"
+    finally:
+        mgr.close()
+
+
+# ---------------------------------------------------------------------------
+# Opt-out path — register_default_sinks=False
+# ---------------------------------------------------------------------------
+
+
+def test_opt_out_skips_both_sinks() -> None:
+    """register_default_sinks=False keeps the audit pipeline empty for tests."""
+    mgr = AuditManager(register_default_sinks=False)
+    try:
+        assert mgr.list_sinks() == []
+    finally:
+        mgr.close()
+
+
+def test_opt_out_does_not_warn(caplog: pytest.LogCaptureFixture) -> None:
+    """Opting out is an explicit signal — no missing-callable warning."""
+    caplog.set_level(logging.WARNING, logger="uipath.runtime.governance._audit.base")
+    mgr = AuditManager(register_default_sinks=False)
+    try:
+        warnings = [
+            r.message for r in caplog.records if r.levelno == logging.WARNING
+        ]
+        assert not any("track_events" in m for m in warnings)
+    finally:
+        mgr.close()
diff --git a/tests/test_evaluator_telemetry.py b/tests/test_evaluator_telemetry.py
new file mode 100644
index 0000000..a03d9a7
--- /dev/null
+++ b/tests/test_evaluator_telemetry.py
@@ -0,0 +1,289 @@
+"""Tests that the evaluator passes the new telemetry fields downstream.
+
+The evaluator must:
+
+- Measure per-rule and per-hook wall-clock duration.
+- Collect disabled rule ids into ``skipped_policy_names``.
+- Count matched UiPath-mapped guardrails (``guardrail_dispatched_count``).
+- Hand all of the above to ``AuditManager.emit_rule_evaluation`` and
+  ``emit_hook_summary`` so the ``TrackEventAuditSink`` payload carries
+  them.
+
+We capture events via a manager registered with no default sinks +
+one ``_CapturingSink`` so the assertions see exactly the per-rule and
+hook-summary events the evaluator emits.
+"""
+
+from __future__ import annotations
+
+import pytest
+from uipath.core.governance import EnforcementMode
+from uipath.core.governance.models import Action, LifecycleHook
+
+from uipath.runtime.governance._audit.base import (
+    AuditEvent,
+    AuditManager,
+    AuditSink,
+    EventType,
+)
+from uipath.runtime.governance.native.evaluator import GovernanceEvaluator
+from uipath.runtime.governance.native.models import (
+    Check,
+    CheckContext,
+    Condition,
+    PolicyIndex,
+    PolicyPack,
+    Rule,
+)
+
+
+class _CapturingSink(AuditSink):
+    def __init__(self) -> None:
+        self.events: list[AuditEvent] = []
+
+    @property
+    def name(self) -> str:
+        return "capturing"
+
+    def emit(self, event: AuditEvent) -> None:
+        self.events.append(event)
+
+
+def _rule(
+    rule_id: str,
+    *,
+    enabled: bool = True,
+    matches: bool = True,
+    action: Action = Action.DENY,
+) -> Rule:
+    """A rule whose ``contains`` check always (or never) matches the input."""
+    return Rule(
+        rule_id=rule_id,
+        name=f"rule-{rule_id}",
+        clause=rule_id,
+        hook=LifecycleHook.BEFORE_AGENT,
+        action=action,
+        enabled=enabled,
+        checks=[
+            Check(
+                conditions=[
+                    Condition(
+                        operator="contains",
+                        field="agent_input",
+                        value="needle" if matches else "absent-needle",
+                    )
+                ],
+                action=action,
+                message=f"matched {rule_id}",
+            )
+        ],
+    )
+
+
+def _pack(*rules: Rule) -> PolicyIndex:
+    idx = PolicyIndex()
+    idx.add_pack(
+        PolicyPack(name="test_pack", version="1.0", description="t", rules=list(rules))
+    )
+    return idx
+
+
+def _ctx() -> CheckContext:
+    return CheckContext(
+        hook=LifecycleHook.BEFORE_AGENT,
+        agent_name="agent-x",
+        runtime_id="run-1",
+        agent_input="needle",
+    )
+
+
+@pytest.fixture
+def sink_and_manager() -> tuple[_CapturingSink, AuditManager]:
+    """Sink-capturing manager with no default sinks (no traces, no track_events)."""
+    sink = _CapturingSink()
+    mgr = AuditManager(register_default_sinks=False)
+    mgr.register_sink(sink)
+    return sink, mgr
+
+
+def _rule_events(sink: _CapturingSink) -> list[AuditEvent]:
+    return [e for e in sink.events if e.event_type == EventType.RULE_EVALUATION]
+
+
+def _hook_summary(sink: _CapturingSink) -> AuditEvent:
+    summaries = [e for e in sink.events if e.event_type == EventType.HOOK_END]
+    assert len(summaries) == 1, f"expected 1 hook summary, got {len(summaries)}"
+    return summaries[0]
+
+
+# ---------------------------------------------------------------------------
+# Per-rule timing
+# ---------------------------------------------------------------------------
+
+
+def test_rule_evaluation_carries_duration_ms(
+    sink_and_manager: tuple[_CapturingSink, AuditManager],
+) -> None:
+    sink, mgr = sink_and_manager
+    try:
+        evaluator = GovernanceEvaluator(
+            _pack(_rule("A.1.1")),
+            enforcement_mode=EnforcementMode.AUDIT,
+            audit_manager=mgr,
+        )
+        evaluator.evaluate(_ctx())
+    finally:
+        mgr.close()
+
+    rule_evs = _rule_events(sink)
+    assert len(rule_evs) == 1
+    duration = rule_evs[0].data["duration_ms"]
+    assert isinstance(duration, float)
+    assert duration >= 0.0
+
+
+def test_rule_evaluation_mapped_to_uipath_is_false_for_native(
+    sink_and_manager: tuple[_CapturingSink, AuditManager],
+) -> None:
+    """Native rules (no guardrail_fallback) always emit ``mapped_to_uipath=False``."""
+    sink, mgr = sink_and_manager
+    try:
+        evaluator = GovernanceEvaluator(
+            _pack(_rule("A.1.1")),
+            enforcement_mode=EnforcementMode.AUDIT,
+            audit_manager=mgr,
+        )
+        evaluator.evaluate(_ctx())
+    finally:
+        mgr.close()
+
+    rule_evs = _rule_events(sink)
+    assert rule_evs[0].data["mapped_to_uipath"] is False
+
+
+# ---------------------------------------------------------------------------
+# Hook-summary aggregates
+# ---------------------------------------------------------------------------
+
+
+def test_hook_summary_carries_total_duration(
+    sink_and_manager: tuple[_CapturingSink, AuditManager],
+) -> None:
+    sink, mgr = sink_and_manager
+    try:
+        evaluator = GovernanceEvaluator(
+            _pack(_rule("A.1.1"), _rule("A.1.2", matches=False)),
+            enforcement_mode=EnforcementMode.AUDIT,
+            audit_manager=mgr,
+        )
+        evaluator.evaluate(_ctx())
+    finally:
+        mgr.close()
+
+    summary = _hook_summary(sink)
+    duration = summary.data["duration_ms"]
+    assert isinstance(duration, float)
+    assert duration >= 0.0
+
+
+def test_hook_summary_tracks_skipped_disabled_rules(
+    sink_and_manager: tuple[_CapturingSink, AuditManager],
+) -> None:
+    """Disabled rules appear in ``skipped_policy_names`` (and skipped_count)."""
+    sink, mgr = sink_and_manager
+    try:
+        evaluator = GovernanceEvaluator(
+            _pack(
+                _rule("A.1.1"),                      # enabled, matches
+                _rule("A.1.2", enabled=False),       # DISABLED
+                _rule("A.1.3", enabled=False),       # DISABLED
+            ),
+            enforcement_mode=EnforcementMode.AUDIT,
+            audit_manager=mgr,
+        )
+        evaluator.evaluate(_ctx())
+    finally:
+        mgr.close()
+
+    summary = _hook_summary(sink)
+    assert summary.data["skipped_count"] == 2
+    assert set(summary.data["skipped_policy_names"]) == {"A.1.2", "A.1.3"}
+
+
+def test_hook_summary_passed_and_denied_counts(
+    sink_and_manager: tuple[_CapturingSink, AuditManager],
+) -> None:
+    """One match + two non-matches → denied=1, passed=2, matched_rules=1."""
+    sink, mgr = sink_and_manager
+    try:
+        evaluator = GovernanceEvaluator(
+            _pack(
+                _rule("A.1.1"),                          # matches (DENY)
+                _rule("A.1.2", matches=False),           # passes
+                _rule("A.1.3", matches=False),           # passes
+            ),
+            enforcement_mode=EnforcementMode.AUDIT,
+            audit_manager=mgr,
+        )
+        evaluator.evaluate(_ctx())
+    finally:
+        mgr.close()
+
+    summary = _hook_summary(sink)
+    assert summary.data["denied_count"] == 1
+    assert summary.data["passed_count"] == 2
+    # ``matched_rules`` keeps its historical "any check matched" sense.
+    assert summary.data["matched_rules"] == 1
+
+
+def test_hook_summary_matched_allow_rule_counts_as_passed(
+    sink_and_manager: tuple[_CapturingSink, AuditManager],
+) -> None:
+    """A matched rule with action=ALLOW is a positive informational match.
+
+    It should NOT contribute to ``denied_count`` — it rolls into
+    ``passed_count`` instead. ``matched_rules`` (raw count) still
+    includes it for backward compatibility with the legacy traces sink.
+    """
+    sink, mgr = sink_and_manager
+    try:
+        evaluator = GovernanceEvaluator(
+            _pack(
+                _rule("A.1.1"),                                # matches (DENY)
+                _rule("A.1.2", action=Action.ALLOW),           # matches (ALLOW) — positive
+                _rule("A.1.3", matches=False),                 # passes (unmatched)
+            ),
+            enforcement_mode=EnforcementMode.AUDIT,
+            audit_manager=mgr,
+        )
+        evaluator.evaluate(_ctx())
+    finally:
+        mgr.close()
+
+    summary = _hook_summary(sink)
+    # 3 total rules; only the explicit DENY counts as a denial.
+    assert summary.data["denied_count"] == 1
+    # ``passed_count`` includes the unmatched rule AND the matched-allow rule.
+    assert summary.data["passed_count"] == 2
+    # Raw ``matched_rules`` still reflects "any check matched" — both
+    # the DENY and the ALLOW match contribute.
+    assert summary.data["matched_rules"] == 2
+
+
+def test_hook_summary_guardrail_dispatched_count_for_native_rules_is_zero(
+    sink_and_manager: tuple[_CapturingSink, AuditManager],
+) -> None:
+    """Without guardrail_fallback conditions, dispatched count is 0."""
+    sink, mgr = sink_and_manager
+    try:
+        evaluator = GovernanceEvaluator(
+            _pack(_rule("A.1.1")),
+            enforcement_mode=EnforcementMode.AUDIT,
+            audit_manager=mgr,
+        )
+        evaluator.evaluate(_ctx())
+    finally:
+        mgr.close()
+
+    summary = _hook_summary(sink)
+    assert summary.data["guardrail_dispatched_count"] == 0
diff --git a/tests/test_governance_metadata.py b/tests/test_governance_metadata.py
new file mode 100644
index 0000000..0feb264
--- /dev/null
+++ b/tests/test_governance_metadata.py
@@ -0,0 +1,81 @@
+"""Tests for :class:`GovernanceRuntimeMetadata`.
+
+The dataclass carries the per-runtime constants every governance
+telemetry event stamps. Defaults must keep telemetry flowing when the
+host hasn't populated agent type / framework yet, and the runtime
+version must resolve from installed package metadata when available.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+from uipath.runtime.governance._audit.metadata import (
+    NATIVE_EXECUTION_ENGINE,
+    GovernanceRuntimeMetadata,
+    _resolve_runtime_version,
+)
+
+
+def test_defaults() -> None:
+    """Default-constructed metadata uses native engine + ``unknown`` slots."""
+    meta = GovernanceRuntimeMetadata()
+    assert meta.execution_engine == NATIVE_EXECUTION_ENGINE
+    assert meta.agent_type == "unknown"
+    assert meta.agent_framework == "unknown"
+    # runtime_version is resolved at construction; either real or "unknown"
+    assert isinstance(meta.runtime_version, str)
+    assert meta.runtime_version != ""
+
+
+def test_explicit_overrides_persist() -> None:
+    """Host-supplied values override the defaults verbatim."""
+    meta = GovernanceRuntimeMetadata(
+        execution_engine="agt",
+        agent_type="uipath_coded",
+        agent_framework="langchain",
+        runtime_version="1.2.3",
+    )
+    assert meta.execution_engine == "agt"
+    assert meta.agent_type == "uipath_coded"
+    assert meta.agent_framework == "langchain"
+    assert meta.runtime_version == "1.2.3"
+
+
+def test_frozen() -> None:
+    """Dataclass is frozen — host can't mutate per-run constants mid-run."""
+    meta = GovernanceRuntimeMetadata()
+    try:
+        meta.agent_type = "other"  # type: ignore[misc]
+    except Exception as exc:
+        assert "frozen" in str(exc).lower() or "cannot assign" in str(exc).lower()
+    else:
+        raise AssertionError("frozen dataclass must reject attribute writes")
+
+
+def test_as_payload_contains_all_four_fields() -> None:
+    """``as_payload`` is the canonical merge-into-event-data dict shape."""
+    meta = GovernanceRuntimeMetadata(
+        execution_engine="agt",
+        agent_type="uipath_coded",
+        agent_framework="langchain",
+        runtime_version="1.2.3",
+    )
+    payload = meta.as_payload()
+    assert payload == {
+        "execution_engine": "agt",
+        "agent_type": "uipath_coded",
+        "agent_framework": "langchain",
+        "runtime_version": "1.2.3",
+    }
+
+
+def test_runtime_version_fallback_on_missing_package() -> None:
+    """A source checkout with no installed metadata falls back to ``unknown``."""
+    from importlib.metadata import PackageNotFoundError
+
+    with patch(
+        "uipath.runtime.governance._audit.metadata.version",
+        side_effect=PackageNotFoundError("uipath-runtime"),
+    ):
+        assert _resolve_runtime_version() == "unknown"
diff --git a/tests/test_governance_runtime.py b/tests/test_governance_runtime.py
index 324147b..ece0df0 100644
--- a/tests/test_governance_runtime.py
+++ b/tests/test_governance_runtime.py
@@ -11,6 +11,7 @@
 
 from typing import Any
 
+import pytest
 from uipath.core.governance import EnforcementMode
 
 from uipath.runtime.governance.native import (
@@ -154,3 +155,245 @@ async def test_governance_runtime_schema_and_dispose_delegate() -> None:
     await runtime.dispose()
     assert delegate.schema_called
     assert delegate.disposed
+
+
+# ---------------------------------------------------------------------------
+# on_dispose — host-owned cleanup hook
+# ---------------------------------------------------------------------------
+
+
+async def test_dispose_runs_on_dispose_after_delegate() -> None:
+    """Ordering — the delegate is disposed first, then the host's hook runs.
+
+    Rationale: teardown of the wrapped runtime is the primary concern; the
+    host's hook is auxiliary cleanup that gets to observe the post-dispose
+    state (or that has independent resources like a telemetry dispatcher's
+    thread pool to flush).
+    """
+    delegate = _StubDelegate()
+    order: list[str] = []
+
+    def _hook() -> None:
+        order.append("on_dispose")
+
+    # Wrap the delegate's dispose so we can observe ordering without
+    # depending on side-effect timing.
+    _orig_dispose = delegate.dispose
+
+    async def _tracked_dispose() -> None:
+        await _orig_dispose()
+        order.append("delegate.dispose")
+
+    delegate.dispose = _tracked_dispose  # type: ignore[method-assign]
+
+    runtime = UiPathGovernedRuntime(
+        delegate, PolicyIndex(), EnforcementMode.AUDIT, on_dispose=_hook
+    )
+
+    await runtime.dispose()
+
+    assert order == ["delegate.dispose", "on_dispose"]
+
+
+async def test_dispose_runs_on_dispose_even_when_delegate_raises() -> None:
+    """``finally`` semantics — a broken delegate must not skip host cleanup.
+
+    Regression guard: without ``try/finally``, a raise from the delegate
+    would leak past ``dispose`` and never call ``on_dispose``, leaving
+    the host's dispatcher (or any other resource) un-flushed.
+    """
+    hook_called = False
+
+    def _hook() -> None:
+        nonlocal hook_called
+        hook_called = True
+
+    class _RaisingDelegate(_StubDelegate):
+        async def dispose(self) -> None:  # type: ignore[override]
+            raise RuntimeError("delegate boom")
+
+    runtime = UiPathGovernedRuntime(
+        _RaisingDelegate(), PolicyIndex(), EnforcementMode.AUDIT, on_dispose=_hook
+    )
+
+    with pytest.raises(RuntimeError, match="delegate boom"):
+        await runtime.dispose()
+
+    assert hook_called, (
+        "on_dispose must run even when the delegate's dispose raises — "
+        "regression guard for the try/finally contract"
+    )
+
+
+async def test_dispose_is_a_noop_without_on_dispose() -> None:
+    """Default (``on_dispose=None``) leaves ``dispose`` semantically identical
+    to the passthrough case — no crash, no extra work.
+    """
+    delegate = _StubDelegate()
+    runtime = _make_runtime(delegate)  # on_dispose defaults to None
+
+    await runtime.dispose()
+
+    assert delegate.disposed
+    # No assertion on the hook — the point is that its absence
+    # doesn't affect anything.
+
+
+# ---------------------------------------------------------------------------
+# Root-span trace correlation — one trace_id per agent run
+# ---------------------------------------------------------------------------
+
+
+async def test_execute_opens_a_root_span_that_covers_the_whole_invocation() -> None:
+    """A single OTel span wraps BEFORE_AGENT, delegate, and AFTER_AGENT.
+
+    Verifies the unified-trace contract: every code path under
+    ``execute`` — including any framework adapter span the delegate
+    opens — sees the same ``trace_id`` and the same current span. The
+    ``track_events`` sink reads ``trace.get_current_span()`` at emit
+    time, so this guarantee is what makes its ``operation_id``
+    correlate across BEFORE_AGENT, BEFORE_MODEL, AFTER_MODEL, and
+    AFTER_AGENT events.
+    """
+    from opentelemetry import trace
+    from opentelemetry.sdk.trace import TracerProvider
+
+    # Replace the global tracer provider for this test so our span
+    # actually records (the default NoOp provider returns invalid
+    # contexts).
+    trace.set_tracer_provider(TracerProvider())
+
+    seen_trace_ids: list[int] = []
+
+    class _SpanProbeDelegate:
+        async def execute(self, input: Any = None, options: Any = None) -> Any:
+            seen_trace_ids.append(
+                trace.get_current_span().get_span_context().trace_id
+            )
+            return "result"
+
+        async def stream(self, input: Any = None, options: Any = None) -> Any:
+            if False:
+                yield  # pragma: no cover
+
+        async def get_schema(self) -> Any:
+            return "schema"
+
+        async def dispose(self) -> None:
+            return None
+
+    runtime = UiPathGovernedRuntime(
+        _SpanProbeDelegate(),
+        PolicyIndex(),
+        EnforcementMode.AUDIT,
+        agent_name="agent-x",
+        runtime_id="run-1",
+    )
+
+    # Outside execute(): no governance span is active.
+    pre_call_ctx = trace.get_current_span().get_span_context()
+    assert not pre_call_ctx.is_valid, (
+        "no span should be active before execute() opens the wrapper"
+    )
+
+    await runtime.execute({"x": 1})
+
+    # Inside the delegate: a valid span context was current.
+    assert len(seen_trace_ids) == 1
+    assert seen_trace_ids[0] != 0, (
+        "delegate must see a valid OTel trace_id from the wrapping span"
+    )
+
+    # After execute(): the span has closed.
+    post_call_ctx = trace.get_current_span().get_span_context()
+    assert not post_call_ctx.is_valid, "wrapper span must close on return"
+
+
+async def test_execute_span_inherits_host_trace_id_when_one_is_active() -> None:
+    """When the host opens a span first, our wrapper becomes a child.
+
+    Same ``trace_id`` flows through everything — governance events
+    emitted by runtime hooks correlate with the host's outer
+    operation. The host's OTel context is preserved end-to-end.
+    """
+    from opentelemetry import trace
+    from opentelemetry.sdk.trace import TracerProvider
+
+    trace.set_tracer_provider(TracerProvider())
+    tracer = trace.get_tracer("test-host")
+
+    inner_trace_id: list[int] = []
+
+    class _ProbeDelegate:
+        async def execute(self, input: Any = None, options: Any = None) -> Any:
+            inner_trace_id.append(
+                trace.get_current_span().get_span_context().trace_id
+            )
+            return "result"
+
+        async def stream(self, input: Any = None, options: Any = None) -> Any:
+            if False:
+                yield  # pragma: no cover
+
+        async def get_schema(self) -> Any:
+            return "schema"
+
+        async def dispose(self) -> None:
+            return None
+
+    runtime = UiPathGovernedRuntime(
+        _ProbeDelegate(), PolicyIndex(), EnforcementMode.AUDIT
+    )
+
+    with tracer.start_as_current_span("host-outer") as host_span:
+        host_trace_id = host_span.get_span_context().trace_id
+        await runtime.execute({"x": 1})
+
+    # The trace_id seen inside the delegate equals the host's
+    # trace_id — the wrapper became a child of the host's span.
+    assert inner_trace_id == [host_trace_id]
+
+
+async def test_stream_opens_a_root_span_that_covers_iteration() -> None:
+    """``stream``'s span stays open across the entire ``async for``.
+
+    Holding the span over ``await`` boundaries is safe — Python's
+    asyncio propagates contextvars (including OTel's current-span
+    pointer) across suspensions. The delegate (which yields events)
+    must see the same wrapper span on every iteration.
+    """
+    from opentelemetry import trace
+    from opentelemetry.sdk.trace import TracerProvider
+
+    trace.set_tracer_provider(TracerProvider())
+
+    per_event_trace_ids: list[int] = []
+
+    class _StreamProbeDelegate:
+        async def execute(self, input: Any = None, options: Any = None) -> Any:
+            return "result"
+
+        async def stream(self, input: Any = None, options: Any = None) -> Any:
+            for label in ("a", "b", "c"):
+                per_event_trace_ids.append(
+                    trace.get_current_span().get_span_context().trace_id
+                )
+                yield label
+
+        async def get_schema(self) -> Any:
+            return "schema"
+
+        async def dispose(self) -> None:
+            return None
+
+    runtime = UiPathGovernedRuntime(
+        _StreamProbeDelegate(), PolicyIndex(), EnforcementMode.AUDIT
+    )
+
+    events = [e async for e in runtime.stream({"x": 1})]
+
+    assert events == ["a", "b", "c"]
+    # Same valid trace_id on every yield — one span covers the whole
+    # stream.
+    assert len(set(per_event_trace_ids)) == 1
+    assert per_event_trace_ids[0] != 0
diff --git a/tests/test_guardrail_compensation.py b/tests/test_guardrail_compensation.py
index 5d8a674..7d52a03 100644
--- a/tests/test_guardrail_compensation.py
+++ b/tests/test_guardrail_compensation.py
@@ -215,6 +215,37 @@ def test_submit_invokes_provider_with_govern_request() -> None:
     assert request.process_key is None
     assert request.reference_id is None
     assert request.agent_version is None
+    # Runtime-identity fields are unset without metadata (server defaults them).
+    assert request.agent_framework is None
+    assert request.agent_type is None
+    assert request.runtime_version is None
+
+
+def test_submit_stamps_runtime_metadata() -> None:
+    """When runtime_metadata is supplied, its identity fields ride the request."""
+    from uipath.runtime.governance._audit.metadata import GovernanceRuntimeMetadata
+
+    meta = GovernanceRuntimeMetadata(
+        agent_type="uipath_coded",
+        agent_framework="langchain",
+        runtime_version="0.11.4",
+    )
+    provider = _provider()
+    compensator = GuardrailCompensator(provider, runtime_metadata=meta)
+
+    compensator.submit(
+        _rules("pii_detection"),
+        {"content": "x"},
+        "before_model",
+        "2026-06-06T00:00:00Z",
+        "langchain",
+        "patch-langchain",
+    )
+
+    (request,) = provider.compensate.call_args.args
+    assert request.agent_framework == "langchain"
+    assert request.agent_type == "uipath_coded"
+    assert request.runtime_version == "0.11.4"
 
 
 def test_submit_dedupes_validators() -> None:
diff --git a/tests/test_track_events_sink.py b/tests/test_track_events_sink.py
new file mode 100644
index 0000000..adb55a0
--- /dev/null
+++ b/tests/test_track_events_sink.py
@@ -0,0 +1,423 @@
+"""Tests for :class:`TrackEventAuditSink`.
+
+Verifies the volume-control filter (``accepts``), event-name vocabulary,
+payload shape (runtime metadata + per-event fields), and OTel-span →
+``operation_id`` correlation (sink resolves the correlation id from
+the live OTel context at emit time, not from the event itself). Uses
+a capture-callable in place of the host-supplied ``track_event`` so
+no I/O fires.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any
+
+import pytest
+from uipath.core.governance import EnforcementMode
+
+from uipath.runtime.governance._audit.base import AuditEvent, EventType
+from uipath.runtime.governance._audit.metadata import GovernanceRuntimeMetadata
+from uipath.runtime.governance._audit.track_events import (
+    EVENT_HOOK_SUMMARY,
+    EVENT_RULE_DENIED,
+    TrackEventAuditSink,
+)
+
+
+class _Capture:
+    """Stand-in for ``provider.track_event`` — records every call."""
+
+    def __init__(self) -> None:
+        self.calls: list[dict[str, Any]] = []
+
+    def __call__(
+        self,
+        *,
+        event_name: str,
+        data: dict[str, Any] | None = None,
+        operation_id: str | None = None,
+    ) -> None:
+        self.calls.append(
+            {"event_name": event_name, "data": data, "operation_id": operation_id}
+        )
+
+
+@pytest.fixture
+def capture() -> _Capture:
+    return _Capture()
+
+
+@pytest.fixture
+def metadata() -> GovernanceRuntimeMetadata:
+    return GovernanceRuntimeMetadata(
+        execution_engine="uipath_native_governance_checker",
+        agent_type="uipath_coded",
+        agent_framework="langchain",
+        runtime_version="9.9.9-test",
+    )
+
+
+@pytest.fixture
+def sink(
+    capture: _Capture, metadata: GovernanceRuntimeMetadata
+) -> TrackEventAuditSink:
+    return TrackEventAuditSink(capture, metadata)
+
+
+def _rule_event(
+    *,
+    matched: bool,
+    action: str = "deny",
+    mode: EnforcementMode = EnforcementMode.AUDIT,
+    duration_ms: float = 12.5,
+    mapped_to_uipath: bool = False,
+) -> AuditEvent:
+    return AuditEvent(
+        event_type=EventType.RULE_EVALUATION,
+        agent_name="agent-x",
+        hook="after_model",
+        data={
+            "policy_id": "A.6.1.4",
+            "rule_name": "commitment-language",
+            "pack_name": "iso42001",
+            "matched": matched,
+            "action": action,
+            "enforcement_mode": mode,
+            "detail": "Detected a commitment phrase.",
+            "duration_ms": duration_ms,
+            "mapped_to_uipath": mapped_to_uipath,
+        },
+        timestamp=datetime(2026, 6, 25, 12, 0, 0, tzinfo=timezone.utc),
+    )
+
+
+def _hook_event(
+    *,
+    mode: EnforcementMode = EnforcementMode.AUDIT,
+    duration_ms: float = 45.0,
+    passed_count: int = 4,
+    denied_count: int = 1,
+    skipped_policy_names: list[str] | None = None,
+    guardrail_dispatched_count: int = 0,
+) -> AuditEvent:
+    return AuditEvent(
+        event_type=EventType.HOOK_END,
+        agent_name="agent-x",
+        hook="after_model",
+        data={
+            "total_rules": passed_count + denied_count,
+            "matched_rules": denied_count,
+            "final_action": "audit",
+            "enforcement_mode": mode,
+            "duration_ms": duration_ms,
+            "passed_count": passed_count,
+            "denied_count": denied_count,
+            "skipped_count": len(skipped_policy_names or []),
+            "skipped_policy_names": list(skipped_policy_names or []),
+            "guardrail_dispatched_count": guardrail_dispatched_count,
+        },
+        timestamp=datetime(2026, 6, 25, 12, 0, 0, tzinfo=timezone.utc),
+    )
+
+
+# ---------------------------------------------------------------------------
+# accepts() — volume-control filter
+# ---------------------------------------------------------------------------
+
+
+def test_accepts_matched_rule_evaluation(sink: TrackEventAuditSink) -> None:
+    assert sink.accepts(_rule_event(matched=True)) is True
+
+
+def test_rejects_unmatched_rule_evaluation(sink: TrackEventAuditSink) -> None:
+    assert sink.accepts(_rule_event(matched=False)) is False
+
+
+def test_rejects_matched_allow_rule(sink: TrackEventAuditSink) -> None:
+    """A matched rule with action=allow is a positive informational
+    match — it should NOT trigger the ``rule.denied`` stream. It rolls
+    into the hook summary's ``passed_count`` instead.
+    """
+    assert sink.accepts(_rule_event(matched=True, action="allow")) is False
+
+
+def test_direct_emit_of_unmatched_rule_is_noop(
+    sink: TrackEventAuditSink, capture: _Capture
+) -> None:
+    """Defense-in-depth: even if a caller bypasses ``accepts``, passed
+    rules must not be routed to ``governance.rule.denied``.
+
+    The AuditManager dispatch path always consults ``accepts`` first,
+    but a direct ``sink.emit(...)`` call (tests, future alternate
+    dispatch) must still produce zero ``track_event`` calls for a
+    ``matched=False`` rule.
+    """
+    sink.emit(_rule_event(matched=False))
+    assert capture.calls == []
+
+
+def test_direct_emit_of_matched_allow_rule_is_noop(
+    sink: TrackEventAuditSink, capture: _Capture
+) -> None:
+    """Defense-in-depth equivalent of the matched-but-allow filter."""
+    sink.emit(_rule_event(matched=True, action="allow"))
+    assert capture.calls == []
+
+
+# ---------------------------------------------------------------------------
+# DISABLED mode — zero telemetry across every event type
+# ---------------------------------------------------------------------------
+
+
+def test_rejects_disabled_mode_rule_evaluation(sink: TrackEventAuditSink) -> None:
+    """Governance off → ``accepts`` returns False for rule events."""
+    assert (
+        sink.accepts(_rule_event(matched=True, mode=EnforcementMode.DISABLED))
+        is False
+    )
+
+
+def test_rejects_disabled_mode_hook_summary(sink: TrackEventAuditSink) -> None:
+    """Governance off → ``accepts`` returns False for hook summaries."""
+    assert sink.accepts(_hook_event(mode=EnforcementMode.DISABLED)) is False
+
+
+def test_direct_emit_of_disabled_rule_is_noop(
+    sink: TrackEventAuditSink, capture: _Capture
+) -> None:
+    """Defense-in-depth: ``emit`` drops DISABLED-mode events too."""
+    sink.emit(_rule_event(matched=True, mode=EnforcementMode.DISABLED))
+    assert capture.calls == []
+
+
+def test_direct_emit_of_disabled_hook_is_noop(
+    sink: TrackEventAuditSink, capture: _Capture
+) -> None:
+    """Defense-in-depth for hook summary in DISABLED mode."""
+    sink.emit(_hook_event(mode=EnforcementMode.DISABLED))
+    assert capture.calls == []
+
+
+# ---------------------------------------------------------------------------
+# Empty hook summary suppression
+# ---------------------------------------------------------------------------
+
+
+def test_rejects_empty_hook_summary(sink: TrackEventAuditSink) -> None:
+    """``total_rules=0`` AND ``skipped_count=0`` → nothing to report."""
+    ev = _hook_event(passed_count=0, denied_count=0, skipped_policy_names=[])
+    assert sink.accepts(ev) is False
+
+
+def test_accepts_hook_summary_with_only_skipped(sink: TrackEventAuditSink) -> None:
+    """Hook with ``total=0`` but ``skipped_count>0`` is still operator-useful."""
+    ev = _hook_event(
+        passed_count=0,
+        denied_count=0,
+        skipped_policy_names=["A.1.1", "A.1.2"],
+    )
+    assert sink.accepts(ev) is True
+
+
+def test_direct_emit_of_empty_hook_is_noop(
+    sink: TrackEventAuditSink, capture: _Capture
+) -> None:
+    """Defense-in-depth equivalent of the empty-hook filter."""
+    sink.emit(_hook_event(passed_count=0, denied_count=0, skipped_policy_names=[]))
+    assert capture.calls == []
+
+
+def test_accepts_hook_end(sink: TrackEventAuditSink) -> None:
+    assert sink.accepts(_hook_event()) is True
+
+
+def test_rejects_other_event_types(sink: TrackEventAuditSink) -> None:
+    for et in (
+        EventType.HOOK_START,
+        EventType.SESSION_START,
+        EventType.SESSION_END,
+        EventType.POLICY_VIOLATION,
+        EventType.POLICY_ALLOW,
+        EventType.PACKS_LOADED,
+    ):
+        ev = AuditEvent(event_type=et, agent_name="agent-x")
+        assert sink.accepts(ev) is False, f"sink must drop {et}"
+
+
+# ---------------------------------------------------------------------------
+# emit() — rule-denied event shape
+# ---------------------------------------------------------------------------
+
+
+def test_rule_denied_event_name_and_operation_id(
+    sink: TrackEventAuditSink, capture: _Capture
+) -> None:
+    """``operation_id`` is the live OTel span's trace id rendered as 32-hex.
+
+    The sink reads from :func:`opentelemetry.trace.get_current_span` at
+    emit time — the audit manager dispatches synchronously on the
+    caller's thread, so any span open on the calling thread is visible
+    to the sink directly.
+    """
+    from opentelemetry.sdk.trace import TracerProvider
+
+    tracer = TracerProvider().get_tracer("test")
+    with tracer.start_as_current_span("agent-run") as span:
+        expected = f"{span.get_span_context().trace_id:032x}"
+        sink.emit(_rule_event(matched=True))
+
+    assert len(capture.calls) == 1
+    call = capture.calls[0]
+    assert call["event_name"] == EVENT_RULE_DENIED
+    assert call["operation_id"] == expected
+
+
+def test_rule_denied_payload_carries_metadata(
+    sink: TrackEventAuditSink, capture: _Capture
+) -> None:
+    sink.emit(_rule_event(matched=True))
+    data = capture.calls[0]["data"]
+    assert data["execution_engine"] == "uipath_native_governance_checker"
+    assert data["agent_type"] == "uipath_coded"
+    assert data["agent_framework"] == "langchain"
+    assert data["runtime_version"] == "9.9.9-test"
+    assert data["agent_name"] == "agent-x"
+    assert data["hook"] == "after_model"
+
+
+def test_rule_denied_payload_splits_pack_and_clause(
+    sink: TrackEventAuditSink, capture: _Capture
+) -> None:
+    """``pack`` and ``clause`` are separate fields so consumers can aggregate."""
+    sink.emit(_rule_event(matched=True))
+    data = capture.calls[0]["data"]
+    assert data["pack"] == "iso42001"
+    assert data["clause"] == "A.6.1.4"
+    assert data["rule_name"] == "commitment-language"
+
+
+def test_rule_denied_audit_mode_collapses_to_audit(
+    sink: TrackEventAuditSink, capture: _Capture
+) -> None:
+    """AUDIT mode never escalates DENY past AUDIT, but evaluator_result is true."""
+    sink.emit(_rule_event(matched=True, action="deny", mode=EnforcementMode.AUDIT))
+    data = capture.calls[0]["data"]
+    assert data["mode"] == "AUDIT"
+    assert data["evaluator_result"] == "DENY"
+    assert data["action_applied"] == "AUDIT"
+
+
+def test_rule_denied_enforce_mode_deny(
+    sink: TrackEventAuditSink, capture: _Capture
+) -> None:
+    sink.emit(_rule_event(matched=True, action="deny", mode=EnforcementMode.ENFORCE))
+    data = capture.calls[0]["data"]
+    assert data["mode"] == "ENFORCE"
+    assert data["evaluator_result"] == "DENY"
+    assert data["action_applied"] == "DENY"
+
+
+def test_rule_denied_enforce_mode_escalate_is_hitl(
+    sink: TrackEventAuditSink, capture: _Capture
+) -> None:
+    sink.emit(
+        _rule_event(matched=True, action="escalate", mode=EnforcementMode.ENFORCE)
+    )
+    data = capture.calls[0]["data"]
+    assert data["evaluator_result"] == "HITL"
+    assert data["action_applied"] == "HITL"
+
+
+def test_rule_denied_carries_duration(
+    sink: TrackEventAuditSink, capture: _Capture
+) -> None:
+    sink.emit(_rule_event(matched=True, duration_ms=42.5))
+    assert capture.calls[0]["data"]["duration_ms"] == 42.5
+
+
+def test_rule_denied_mapped_to_uipath_passes_through(
+    sink: TrackEventAuditSink, capture: _Capture
+) -> None:
+    """Field defaults False for native events but the schema is stable."""
+    sink.emit(_rule_event(matched=True, mapped_to_uipath=False))
+    assert capture.calls[0]["data"]["mapped_to_uipath"] is False
+
+
+# ---------------------------------------------------------------------------
+# emit() — hook-summary event shape
+# ---------------------------------------------------------------------------
+
+
+def test_hook_summary_event_name_and_operation_id(
+    sink: TrackEventAuditSink, capture: _Capture
+) -> None:
+    """Same OTel-derived operation_id contract as the rule-denied event."""
+    from opentelemetry.sdk.trace import TracerProvider
+
+    tracer = TracerProvider().get_tracer("test")
+    with tracer.start_as_current_span("agent-run") as span:
+        expected = f"{span.get_span_context().trace_id:032x}"
+        sink.emit(_hook_event())
+
+    assert len(capture.calls) == 1
+    assert capture.calls[0]["event_name"] == EVENT_HOOK_SUMMARY
+    assert capture.calls[0]["operation_id"] == expected
+
+
+def test_hook_summary_carries_counts(
+    sink: TrackEventAuditSink, capture: _Capture
+) -> None:
+    sink.emit(
+        _hook_event(
+            passed_count=10,
+            denied_count=2,
+            skipped_policy_names=["A.1.2", "A.3.4"],
+            guardrail_dispatched_count=3,
+        )
+    )
+    data = capture.calls[0]["data"]
+    assert data["passed_count"] == 10
+    assert data["denied_count"] == 2
+    assert data["skipped_count"] == 2
+    assert data["skipped_policy_names"] == ["A.1.2", "A.3.4"]
+    assert data["guardrail_dispatched_count"] == 3
+
+
+def test_hook_summary_duration_ms(
+    sink: TrackEventAuditSink, capture: _Capture
+) -> None:
+    sink.emit(_hook_event(duration_ms=123.4))
+    assert capture.calls[0]["data"]["duration_ms"] == 123.4
+
+
+def test_hook_summary_carries_metadata(
+    sink: TrackEventAuditSink, capture: _Capture
+) -> None:
+    """Same per-runtime metadata stamping as rule-denied events."""
+    sink.emit(_hook_event())
+    data = capture.calls[0]["data"]
+    assert data["execution_engine"] == "uipath_native_governance_checker"
+    assert data["agent_type"] == "uipath_coded"
+    assert data["agent_framework"] == "langchain"
+    assert data["runtime_version"] == "9.9.9-test"
+
+
+def test_hook_summary_mode_string(
+    sink: TrackEventAuditSink, capture: _Capture
+) -> None:
+    sink.emit(_hook_event(mode=EnforcementMode.ENFORCE))
+    assert capture.calls[0]["data"]["mode"] == "ENFORCE"
+
+
+# ---------------------------------------------------------------------------
+# operation_id fallback when no OTel span is active
+# ---------------------------------------------------------------------------
+
+
+def test_operation_id_none_when_no_active_span(
+    sink: TrackEventAuditSink, capture: _Capture
+) -> None:
+    """No live OTel span → ``operation_id=None`` (consumer fills in its own)."""
+    sink.emit(_rule_event(matched=True))
+    assert capture.calls[0]["operation_id"] is None
diff --git a/uv.lock b/uv.lock
index 76fa1ec..d18b949 100644
--- a/uv.lock
+++ b/uv.lock
@@ -1148,16 +1148,16 @@ wheels = [
 
 [[package]]
 name = "uipath-core"
-version = "0.5.25"
+version = "0.5.28"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "opentelemetry-instrumentation" },
     { name = "opentelemetry-sdk" },
     { name = "pydantic" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/f6/f3/8c4220bcd6b5a85a0be48e5e48afc0a0182454526f76b4927be2b307b065/uipath_core-0.5.25.tar.gz", hash = "sha256:8645b65b7987b4fc7d686d07310a3a31b447c2e7dc5e567cff55612a96f24ab2", size = 130395, upload-time = "2026-06-29T10:01:34.602Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/4c/f9/8d2f1d98cbebbcf059cf4561f38f34ad4cd58423e4f15cad22bd297a2563/uipath_core-0.5.28.tar.gz", hash = "sha256:942987f6b612c64f93d612ad7b242276ed75f129fdd8f25bc71c24ec8887e388", size = 130578, upload-time = "2026-06-30T14:04:48.841Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/fe/56/c87da71a367fecd10a5f0f9cb4b05489ea214d80ba597ca636394cd5ebfa/uipath_core-0.5.25-py3-none-any.whl", hash = "sha256:5cf95a9ffa7bc2bd95d394aeb7f44dc4979a69e067c1ef92b41fd05ead802e7d", size = 54759, upload-time = "2026-06-29T10:01:33.189Z" },
+    { url = "https://files.pythonhosted.org/packages/e8/1e/385bb166232a57ebe938cc57ad2717f350bc922bb5d2ce31af84306b7569/uipath_core-0.5.28-py3-none-any.whl", hash = "sha256:b952a46a21710073cbc16d6d5684e9aa645c107f57a636b778cfb94aa81a1e48", size = 54980, upload-time = "2026-06-30T14:04:47.374Z" },
 ]
 
 [[package]]
@@ -1191,7 +1191,7 @@ dev = [
 requires-dist = [
     { name = "chardet", specifier = ">=5.2.0,<8.0" },
     { name = "pyyaml", specifier = ">=6.0,<7.0" },
-    { name = "uipath-core", specifier = ">=0.5.25,<0.6.0" },
+    { name = "uipath-core", specifier = ">=0.5.28,<0.6.0" },
     { name = "vadersentiment", specifier = ">=3.3.2,<4.0" },
 ]