Policy Plugin Guide

August 10, 2026 · View on GitHub

A policy plugin is HiveMind's admission-control point: it sees every Mycroft Message (and every binary payload) about to be forwarded to the agent bus, and can allow it, deny it, or mutate it. Multiple policies form a chain, executed in an operator-declared order.

Policies are the mechanism for dynamic ACL checks that depend on current server state — per-API-key quotas, rate limits, parental controls, audit logs, sensitive-intent confirmation. Static ACLs that live on the Client record (allowed_types, blacklists) are themselves expressed as built-in policy plugins by hivemind-core.

Architecture context: see HiveMind-core#85. This document covers the plugin author surface defined in hivemind-plugin-manager. The chain runner that consumes these plugins lives in hivemind-core.

Base Class

@dataclass
class PolicyPlugin(_SubProtocol):
    """Base class for HiveMind policy plugins.

    Subclasses override review (and optionally review_binary / observe).
    They are loaded by hivemind-core via the hivemind.policy
    entry-point group and invoked in the order declared by the
    operator's policy.chain config.
    """
    config: Dict[str, Any] = dataclasses.field(default_factory=dict)
    hm_protocol: Optional['HiveMindListenerProtocol'] = None

    def review(self, message, client) -> Verdict: ...
    def review_binary(self, payload, client) -> Verdict: ...
    def observe(self, message, client) -> None: ...

Source: hivemind_plugin_manager/policy.py:115.

PolicyPlugin extends _SubProtocol (protocols.py:35), which provides .identity, .database, and .clients properties — see Concepts.

`client.is_admin`

Client.is_admin is informational only. The chain runner does not give it any special treatment — every policy is invoked for every client, admin or not.

Policies that care about admin status check client.is_admin themselves inside review and decide what to do with it. Concrete example (the only one in the first-party ecosystem): OVOSAgentPolicy refuses messages with session_id == "default" for non-admins but lets admins through.

There is no class-level BYPASS_ADMIN switch and no runner-level admin handling. If your policy should be a no-op for admins, branch on client.is_admin at the top of review.

Hooks

Method	Called when	Default impl
`review(message, client)`	A Mycroft `Message` is about to be forwarded to the agent bus.	Returns `Verdict.allow()`.
`review_binary(payload, client)`	A binary payload (e.g. raw audio) is about to be forwarded.	Returns `Verdict.allow()`.
`observe(message, client)`	A message was successfully emitted to the bus. Use for counters, audit logs, telemetry.	No-op.

All three are synchronous. The chain runner in hivemind-core is synchronous; introducing an async variant would require a coordinated change across hivemind-plugin-manager, hivemind-core, and every agent protocol — out of scope for this contract.

Failure semantics (enforced by the chain runner, not the plugin):

Unhandled exception in review / review_binary → treated as Verdict.deny("policy_error", reason="policy crashed"). Fail-closed.
Unhandled exception in observe → logged and swallowed. Observation never blocks delivery.
Unhandled exception in a Mutation.apply → logged and skipped; the message proceeds with whatever mutations did apply.

`Verdict`

The return value of review() and review_binary().

Source: hivemind_plugin_manager/policy.py:73

@dataclass(frozen=True)
class Verdict:
    denied: bool = False
    code: str = ""
    reason: str = ""
    data: Dict[str, Any] = field(default_factory=dict)
    mutations: List[Mutation] = field(default_factory=list)

    @classmethod
    def allow(cls, *mutations: Mutation) -> "Verdict": ...
    @classmethod
    def deny(cls, code: str, reason: str = "", **data: Any) -> "Verdict": ...

A verdict is either denying or allowing-with-mutations:

# Allow, no changes
return Verdict.allow()

# Allow, with mutations (concrete Mutation subclasses come from the agent plugin)
# from hivemind_ovos_agent_plugin.policy import AddBlacklistedSkill, SetSessionField
return Verdict.allow(
    AddBlacklistedSkill("adult.skill"),
    SetSessionField("filter_level", "family"),
)

# Deny — short-circuits the chain
return Verdict.deny(
    "quota_exceeded",
    "daily limit of 100 reached",
    limit=100, used=100, window="1d",
)

code is a stable, machine-readable string clients can switch on ("quota_exceeded", "intent_blacklisted", "policy_error"). reason is human-readable. Extra keyword arguments go into data, which the chain runner forwards to the client as part of the hive.policy.denied notification message.

Mutations attached to a denying verdict are ignored.

`Mutation`

Source: hivemind_plugin_manager/policy.py:47

Typed actions a policy can request on a message being allowed. hivemind-plugin-manager ships only the abstract base class (Mutation) — concrete mutations are agent-specific and live with the agent plugin that knows the message shape. The OVOS agent plugin, for example, ships:

Class	Effect
`AddBlacklistedSkill(skill_id)`	Appends to `message.context["session"]["blacklisted_skills"]`.
`AddBlacklistedIntent(intent_name)`	Appends to `message.context["session"]["blacklisted_intents"]`.
`SetSessionField(key, value)`	Sets one key in `message.context["session"]`.
`SetContextField(path, value)`	Sets a nested key in `message.context` (tuple-typed path).
`RewriteUtterance(text)`	Replaces `data["utterances"]` on a `recognizer_loop:utterance` message.

All of these live in hivemind_ovos_agent_plugin.policy — see that repo for current behaviour, signatures, and edge cases.

If you write an agent integration for something other than OVOS (e.g. a non-Mycroft skill engine), bring your own mutation set. The contract is just: subclass Mutation, implement apply(message, client).

from hivemind_plugin_manager import Mutation

class TagWithCorrelationID(Mutation):
    def __init__(self, correlation_id: str):
        self.correlation_id = correlation_id

    def apply(self, message, client) -> None:
        message.context["correlation_id"] = self.correlation_id

Why typed mutations and not a free-form dict-merge? So the chain runner (and reviewers) can see exactly what each plugin is allowed to change. Add new mutation kinds as named subclasses in the consumer that understands the field.

Registering a plugin

In your package's setup.py / pyproject.toml, register under hivemind.policy:

[project.entry-points."hivemind.policy"]
"hivemind-intent-quota-policy" = "hivemind_intent_quota:IntentQuotaPolicy"

Operators then enable your policy in hivemind-core's policy config block. Each chain entry is an object, not a bare string: PolicyChain.from_config reads module, config and optional off it, so a plain string raises AttributeError. hivemind-core catches that, logs the failure and installs DenyAllPolicy, so the server starts but denies every message until the config is fixed. hivemind-core policy list and policy test surface the error directly. Per-plugin settings go in that entry's config, not in a sibling key. optional: true means exceptions from that policy are logged and the chain continues past it — the policy is skipped, it does not contribute an allow verdict, and later policies still run. A plugin that fails to load aborts chain construction regardless of optional.

{
  "policy": {
    "chain": [
      {"module": "hivemind-client-acl-policy"},
      {
        "module": "hivemind-intent-quota-policy",
        "config": {
          "per_day": 1000,
          "redis_url": "redis://localhost:6379/0"
        },
        "optional": false
      }
    ]
  }
}

Example: intent quota plugin

A small, realistic policy that tracks per-account utterance counts in Redis and denies once a daily limit is reached. Uses Client.metadata (see Database) to group counters by account_id, so a deployment can have multiple HiveMind clients sharing one quota.

from hivemind_plugin_manager import PolicyPlugin, Verdict
# Mutations live with the agent plugin — import what you need from there.
# from hivemind_ovos_agent_plugin.policy import AddBlacklistedSkill


class IntentQuotaPolicy(PolicyPlugin):
    def __init__(self, config, hm_protocol=None):
        super().__init__(config=config, hm_protocol=hm_protocol)
        self.per_day = config.get("per_day", 1000)
        self.store = RedisCounter(config["redis_url"])

    def review(self, message, client):
        if message.msg_type != "recognizer_loop:utterance":
            return Verdict.allow()
        user = client.resolve_user(self.hm_protocol.db)
        account = (user.metadata.get("account_id") if user else None) or client.key
        used = self.store.get(account, window="1d")
        if used >= self.per_day:
            return Verdict.deny(
                "quota_exceeded",
                f"daily limit {self.per_day} reached",
                limit=self.per_day, used=used,
            )
        return Verdict.allow()

    def observe(self, message, client):
        # Only count messages that were actually emitted to the bus —
        # denials don't increment the counter.
        if message.msg_type != "recognizer_loop:utterance":
            return
        user = client.resolve_user(self.hm_protocol.db)
        account = (user.metadata.get("account_id") if user else None) or client.key
        self.store.incr(account, window="1d")

That's the whole plugin: ~20 lines. The chain runner handles the rest (emitting hive.policy.denied to the client, exception fail-closed, running multiple policies in the configured order, applying mutations).

Factory

from hivemind_plugin_manager import PolicyPluginFactory

cls = PolicyPluginFactory.get_class("hivemind-intent-quota-policy")
plugin = PolicyPluginFactory.create(
    "hivemind-intent-quota-policy",
    config={"per_day": 100, "redis_url": "redis://..."},
    hm_protocol=my_hm_protocol,
)

Source: hivemind_plugin_manager/__init__.py — PolicyPluginFactory.

hivemind-core uses this factory internally to assemble the configured chain at startup; most plugin authors will never call it directly.