ExMaude.AI (ExMaude v0.2.0)

View Source

AI rule conflict detection using Maude formal verification.

Companion to ExMaude.IoT. Where ExMaude.IoT verifies physical- IoT automation rules over Things, Properties, and Actions, this module verifies 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

Ten 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. :budget_cascade — chained rule firings exceed tenant budget under bounded simulation (search-based, not yet wired up)
  5. :cost_ceiling_infeasibility — tenant policy intersection leaves an empty cost-acceptable provider set (search-based)
  6. :sovereignty_violation — tool invocation routes through a jurisdiction outside the tenant's allowed set
  7. :authority_escalation — rule grants a capability that another rule requires at higher authority
  8. :approval_gate_bypass — high-impact invocation reachable without traversing an approval gate
  9. :agent_loop_cascade — one rule's capability grants another rule's required capability (cascade edge)
  10. :provider_routing_infeasibility — empty action space under tenant policy intersection

Conflicts 1, 2, 3, 6, 7, 8, 9 are pure-equational and run on the hot path. Conflicts 4, 5, 10 require state-space search via ExMaude.search/4 and are not yet exposed on the public API.

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)

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]}}.