ExMaude.AI (ExMaude v0.4.0)

View Source

AI rule conflict detection using a Maude equational model.

Companion to ExMaude.IoT. Where ExMaude.IoT analyzes physical- IoT automation rules over Things, Properties, and Actions, this module analyzes AI-generated automation rules over Agents, Capabilities, ToolInvocations, and richer predicates (capability, budget, jurisdiction, authority, approval).

Targets the bundled priv/maude/ai-rules.maude template.

Conflict types

Seven conflict types are detected:

  1. :tool_call_conflict — same agent, same tool, conflicting required arguments
  2. :capability_shadowing — two rules grant the same capability at equal priority within a tenant
  3. :pack_tool_composition_mismatch — same capability name with mismatched type-shape signatures
  4. :sovereignty_violation — tool invocation routes through a jurisdiction outside the tenant's allowed set
  5. :authority_escalation — rule grants a capability that another rule requires at higher authority
  6. :approval_gate_bypass — high-impact invocation reachable without traversing an approval gate
  7. :agent_loop_cascade — one rule's capability grants another rule's required capability (cascade edge)

Usage

tenant = "acme"

rules = [
  %{
    id: "approve-then-dose",
    agent_id: {tenant, "ph-controller"},
    trigger: {:prop_lt, "ph", {:int, 6}},
    invocations: [
      {:require_approval, "dosing_high_delta"},
      {:invoke_tool, "dose_chemical",
       %{"chemical" => "ph_up", "ml" => 50}, "high_impact", :eu}
    ],
    capability_grants: [{:cap, "ph_dosing", "v1"}],
    authority_required: 2,
    priority: 1
  },
  %{
    id: "auto-dose",
    agent_id: {tenant, "ph-controller"},
    trigger: {:prop_lt, "ph", {:int, 5}},
    invocations: [
      {:invoke_tool, "dose_chemical",
       %{"chemical" => "ph_up", "ml" => 100}, "high_impact", :eu}
    ],
    capability_grants: [],
    authority_required: 0,
    priority: 1
  }
]

{:ok, conflicts} = ExMaude.AI.detect_conflicts(rules, jurisdictions: [:eu])
# => [%{type: :approval_gate_bypass, rule1: "auto-dose", rule2: nil, ...}]

Telemetry

Emits [:ex_maude, :ai, :detect_conflicts, :start | :stop] events with measurements :duration, :rule_count, :conflict_count and metadata including :result and :template (:ai_rules).

See ExMaude.Telemetry for full event documentation.

Summary

Types

agent_id()
ai_rule()
capability()
conflict()
predicate()
tenant_id()
tool_invocation()
value()

Functions

detect_conflicts(rules, opts \\ [])

Detects all AI-rule conflicts in a rule set.

detect_pair_conflicts(rules, opts \\ [])

Detects only pairwise conflicts (tool-call, capability shadowing, pack-tool composition, authority escalation, agent-loop cascade).

detect_single_rule_conflicts(rules, opts \\ [])

Detects only sovereignty + approval-gate conflicts (single-rule conflicts independent of pair analysis).

validate_rule(rule)

Validates an AI rule structure without sending it to Maude.

validate_rules(rules)

Validates a list of AI rules. Returns :ok if all are valid, or {:error, %{rule_id => [errors]}}.

Types

agent_id()

@type agent_id() :: {tenant_id(), String.t()}

ai_rule()

@type ai_rule() :: %{
  :id => String.t(),
  :agent_id => agent_id(),
  :trigger => predicate(),
  :invocations => [tool_invocation()],
  optional(:capability_grants) => [capability()],
  optional(:authority_required) => non_neg_integer(),
  optional(:priority) => non_neg_integer()
}

capability()

@type capability() ::
  {:cap, name :: String.t(), shape_hash :: String.t()} | String.t()

conflict()

@type conflict() :: ExMaude.AI.ConflictParser.conflict()

predicate()

@type predicate() ::
  {:always}
  | {:prop_eq, String.t(), term()}
  | {:prop_gt, String.t(), term()}
  | {:prop_lt, String.t(), term()}
  | {:prop_gte, String.t(), term()}
  | {:prop_lte, String.t(), term()}
  | {:capability_required, String.t()}
  | {:capability_granted, String.t()}
  | {:budget_within, String.t(),
     {:interval, non_neg_integer(), non_neg_integer()}}
  | {:authority_at_least, non_neg_integer()}
  | {:authority_required, non_neg_integer()}
  | {:jurisdiction_allowed, atom()}
  | {:jurisdiction_forbidden, atom()}
  | {:latency_at_most, non_neg_integer()}
  | {:and, predicate(), predicate()}
  | {:or, predicate(), predicate()}
  | {:not, predicate()}

tenant_id()

@type tenant_id() :: String.t()

tool_invocation()

@type tool_invocation() ::
  {:invoke_tool, name :: String.t(), args :: map(),
   capability_required :: String.t(), jurisdiction :: atom()}
  | {:require_approval, class :: String.t()}

value()

@type value() ::
  {:str, String.t()}
  | {:int, integer()}
  | {:bool, boolean()}
  | {:interval, non_neg_integer(), non_neg_integer()}
  | {:jurisdiction, atom()}
  | nil
  | String.t()
  | integer()
  | boolean()

Functions

detect_conflicts(rules, opts \\ [])

@spec detect_conflicts(
  [ai_rule()],
  keyword()
) :: {:ok, [conflict()]} | {:error, term()}

Detects all AI-rule conflicts in a rule set.

Options

  • :jurisdictions — list of allowed jurisdiction atoms for the tenant (default [] which disables sovereignty checking — pass explicit allowed set to enable)
  • :timeout — Maude command timeout in milliseconds (default 10_000)
  • :pool — registered caller-owned pool (default :ex_maude_pool)

Examples

{:ok, conflicts} = ExMaude.AI.detect_conflicts(rules, jurisdictions: [:eu, :ch])

detect_pair_conflicts(rules, opts \\ [])

@spec detect_pair_conflicts(
  [ai_rule()],
  keyword()
) :: {:ok, [conflict()]} | {:error, term()}

Detects only pairwise conflicts (tool-call, capability shadowing, pack-tool composition, authority escalation, agent-loop cascade).

Skips single-rule conflicts (sovereignty, approval-gate bypass).

detect_single_rule_conflicts(rules, opts \\ [])

@spec detect_single_rule_conflicts(
  [ai_rule()],
  keyword()
) :: {:ok, [conflict()]} | {:error, term()}

Detects only sovereignty + approval-gate conflicts (single-rule conflicts independent of pair analysis).

Requires :jurisdictions option for sovereignty checking. Approval- gate bypass detection runs regardless of jurisdiction set.

validate_rule(rule)

@spec validate_rule(ai_rule()) :: :ok | {:error, [String.t()]}

Validates an AI rule structure without sending it to Maude.

Returns :ok if the rule is valid, or {:error, errors} with a list of validation error messages.

validate_rules(rules)

@spec validate_rules([ai_rule()]) ::
  :ok | {:error, %{required(String.t()) => [String.t()]}}

Validates a list of AI rules. Returns :ok if all are valid, or {:error, %{rule_id => [errors]}}.