PolicyEngine enforces named policies against recorded decisions, returning True if the decision satisfies all policy rules. Use it to gate AI decisions at runtime — attributions requiring dual-source confirmation, escalations requiring senior approval, or any decision category where compliance must be verified before the outcome is recorded. Policies are versioned graph nodes, so every check, exception, and approval chain is part of the permanent audit trail.
The Policy Engine sits above AgentContext and ContextGraph. Policies are stored as nodes in the same graph as decisions, giving them the same causal tracing, provenance, and temporal validity as any other knowledge graph entity. PolicyEngine and Policy import from semantica.context. Decision imports from semantica.context (it is a dataclass defined in semantica.context.decision_models). DecisionRecorder imports from semantica.context.decision_recorder.

​
Defining the policy

A Policy is a dataclass with a free-form rules dict — encode whatever your domain requires. 
from semantica.context import ContextGraph, PolicyEngine, Policy
from datetime import datetime

graph  = ContextGraph()
engine = PolicyEngine(graph_store=graph)

attribution_policy = Policy(
    policy_id   = "pol-attr-001",
    name        = "Nation-State Attribution — Dual Source + Senior Approval",
    description = (
        "Attributions to nation-state actors require corroboration from two independent "
        "intelligence sources and explicit senior analyst approval before being recorded "
        "in the authoritative graph."
    ),
    rules = {
        "min_independent_sources":  2,
        "required_approver_role":   "senior_analyst",
        "disallowed_outcomes":      ["nation_state_attributed_single_source"],
        "min_confidence":           0.85,
        "mandatory_fields":         ["source_a", "source_b", "approver"],
    },
    category   = "threat_attribution",
    version    = "1.0.0",
    created_at = datetime.utcnow(),
    updated_at = datetime.utcnow(),
)

policy_id = engine.add_policy(attribution_policy)
print(f"Policy registered: {policy_id}")
# Policy registered: pol-attr-001
The policy is now a node in the graph. It has a version string, a creation timestamp, and a rules dict that the compliance checker will read when evaluating decisions against it.

​
Checking a decision for compliance

check_compliance takes a Decision object and the policy ID and returns a boolean. 
from semantica.context import Decision

# The AI analyst's APT29 attribution — only one source cited, no senior approval yet
decision = Decision(
    decision_id   = "",                        # auto-generated if left empty
    category      = "threat_attribution",
    scenario      = "APT29 activity cluster in NATO network telemetry Q2 2025",
    reasoning     = (
        "Observed TTPs match APT29 historical patterns: HAMMERTOSS C2, "
        "spear-phishing via OneDrive lure, targeting foreign ministry staff. "
        "Single source: internal SIEM telemetry."
    ),
    outcome       = "nation_state_attributed_single_source",
    confidence    = 0.91,
    timestamp     = datetime.utcnow(),
    decision_maker= "ai_threat_analyst_v3",
)

is_compliant = engine.check_compliance(decision, policy_id)
print(f"Compliant: {is_compliant}")
# Compliant: False
#
# The outcome "nation_state_attributed_single_source" is in disallowed_outcomes.
# The policy requires min_independent_sources=2 — the decision only cited one.
The engine returns False. The decision has not been rejected — it has been flagged. What happens next depends on your workflow. In some organisations, a non-compliant result simply blocks the write to the authoritative graph. In others, it triggers an exception process where a human approver reviews the evidence and signs off.

​
Recording a policy exception

record_exception permanently links a decision, the policy it violated, the approver identity, and the justification. 
exception_id = engine.record_exception(
    decision_id  = decision.decision_id,
    policy_id    = policy_id,
    reason       = "Time-sensitive attribution — protective action required before second source available",
    approver     = "sr_analyst_chen",
    justification= (
        "Senior analyst reviewed SIEM telemetry and concurs with TTP matching. "
        "Exception approved under Emergency Attribution Procedure §3.2. "
        "Second source corroboration to be completed within 72 hours."
    ),
)

print(f"Exception recorded: {exception_id}")
# Exception recorded: exc-pol-attr-001-20250621-001
#
# The exception node is linked to both the decision and the policy in the graph,
# creating a permanent three-way provenance link: decision → exception → policy.

​
Building a multi-level approval chain

For the highest-stakes decisions — formal attribution reports that will be shared with government partners — a single approver is not enough. Three people need to sign off: the team lead, the department head, and the CISO. DecisionRecorder.record_approval_chain captures all three in a single call, linking each approver to the communication method and context of their sign-off. 
from semantica.context.decision_recorder import DecisionRecorder

recorder = DecisionRecorder(graph_store=graph)

# approvers, methods, and contexts must be equal-length parallel lists
recorder.record_approval_chain(
    decision_id = decision.decision_id,
    approvers   = ["team_lead_okonkwo",  "dept_head_zhang",    "ciso_miller"],
    methods     = ["slack_dm",            "zoom_call",           "email"],
    contexts    = [
        "Team lead reviewed TTPs and SIEM evidence",
        "Dept head approved sharing with Five Eyes partners",
        "CISO authorised formal nation-state attribution report",
    ],
)

print("Three-level approval chain recorded")
# The graph now carries a directed approval chain:
#   decision → team_lead_okonkwo → dept_head_zhang → ciso_miller
# Each link is stamped with method and context for Inspector General review.

​
What-if impact analysis before changing a policy

Six months on, your threat intelligence lead wants to tighten the policy: raise the minimum confidence threshold from 0.85 to 0.92 to reduce false-positive attributions. Before she updates the policy, she wants to know how many past decisions would have been blocked under the stricter rule. analyze_policy_impact runs a what-if simulation over the historical decision record — no permanent changes are made. 
current_policy = engine.get_policy(policy_id)

impact = engine.analyze_policy_impact(
    policy_id      = policy_id,
    proposed_rules = {**current_policy.rules, "min_confidence": 0.92},
)

print(f"Decisions affected by raising confidence floor to 0.92: {impact.get('affected_decisions', 0)}")
# Decisions affected by raising confidence floor to 0.92: 4
#
# Four past attributions had confidence between 0.85 and 0.92.
# Under the new rule, all four would have required exceptions.
# The lead can now make an evidence-based choice: is that trade-off acceptable?
The impact dict contains per-decision detail, not just the count. You can inspect which specific attribution decisions would have been affected, review their reasoning, and decide whether tightening the threshold is worth it.

​
Updating the policy and finding affected decisions

The lead decides to proceed with the threshold increase. She updates the policy to version 1.1.0, recording her reason. The old version is preserved in the history. 
updated_policy_id = engine.update_policy(
    policy_id     = policy_id,
    rules         = {**current_policy.rules, "min_confidence": 0.92},
    change_reason = "Q3 attribution quality review — raise confidence floor from 0.85 to 0.92 "
                    "to reduce false-positive nation-state attributions",
    new_version   = "1.1.0",
)

print(f"Policy updated: {updated_policy_id} -> version 1.1.0")
# Policy updated: pol-attr-001 -> version 1.1.0

# Find all decisions that were evaluated under v1.0.0 —
# these need to be re-reviewed to confirm they still meet the new standard.
affected = engine.get_affected_decisions(
    policy_id    = policy_id,
    from_version = "1.0.0",
    to_version   = "1.1.0",
)

print(f"Decisions to re-audit: {len(affected)}")
for dec in affected:
    print(f"  {dec.get('decision_id')}  confidence={dec.get('confidence')}  outcome={dec.get('outcome')}")
# decision-a3f1  confidence=0.87  outcome=nation_state_attributed
# decision-b22c  confidence=0.89  outcome=nation_state_attributed
# ...
This is the re-audit workflow: every decision made under the old policy is surfaced, reviewed against the new standard, and either re-confirmed or flagged for correction. The graph preserves the full history of which policy version governed each decision.

​
Reviewing the full audit trail

At any point — for an Inspector General review, a board report, or an incident investigation — you can retrieve the complete version history of a policy. 
history = engine.get_policy_history(policy_id)

print(f"Policy versions on record: {len(history)}")
for version in history:
    print(f"  v{version.version}  updated={version.updated_at.date()}  rules_keys={list(version.rules.keys())}")
# v1.0.0  updated=2025-06-21  rules_keys=[min_independent_sources, required_approver_role, ...]
# v1.1.0  updated=2025-09-14  rules_keys=[min_independent_sources, required_approver_role, ...]
#
# The full rules dict for each version is preserved — you can replay exactly
# what the compliance check would have returned for any past decision against
# any past version of the policy.

​
Domain Examples

TLP:RED intelligence must never be shared outside the originating organisation without commander-level approval. The policy is enforced on every information-sharing decision that touches classified threat reporting. Violations are routed to the J2 officer for exception review rather than silently logged.
from semantica.context import ContextGraph, PolicyEngine, Policy, Decision
from semantica.context.decision_recorder import DecisionRecorder
from datetime import datetime

graph    = ContextGraph()
engine   = PolicyEngine(graph_store=graph)
recorder = DecisionRecorder(graph_store=graph)

opsec_policy = Policy(
    policy_id   = "pol-opsec-001",
    name        = "TLP:RED — Restricted Dissemination",
    description = "TLP:RED intelligence must not be shared outside the originating organisation",
    rules = {
        "classification":      "TLP:RED",
        "disallowed_outcomes": ["shared_with_partner", "published"],
        "min_confidence":      0.95,
        "mandatory_fields":    ["tlp", "classification", "authorised_recipients"],
    },
    category   = "information_sharing",
    version    = "2.1.0",
    created_at = datetime.utcnow(),
    updated_at = datetime.utcnow(),
)
engine.add_policy(opsec_policy)

decision = Decision(
    decision_id   = "",
    category      = "information_sharing",
    scenario      = "APT29 SIGINT report TLP:RED — share with Five Eyes partners?",
    reasoning     = "Tactical intelligence — partner request via UKIC liaison",
    outcome       = "shared_with_partner",   # violates TLP:RED policy
    confidence    = 0.88,
    timestamp     = datetime.utcnow(),
    decision_maker= "analyst_rodriguez",
)

is_compliant = engine.check_compliance(decision, "pol-opsec-001")
print(f"Compliant: {is_compliant}")
# Compliant: False — outcome 'shared_with_partner' is disallowed; confidence below 0.95

if not is_compliant:
    # Route to J2 for exception review — dual commander approval required
    exception_id = engine.record_exception(
        decision_id  = decision.decision_id,
        policy_id    = "pol-opsec-001",
        reason       = "Five Eyes partner urgent request — time-critical tactical intelligence",
        approver     = "j2_officer_hayes",
        justification= "Commander approved limited dissemination under UKUSA Article 4 emergency clause",
    )
    recorder.record_approval_chain(
        decision_id = decision.decision_id,
        approvers   = ["j2_officer_hayes", "unit_commander_brooks"],
        methods     = ["secure_phone",      "in_person"],
        contexts    = ["J2 tactical review", "Commander emergency approval"],
    )
    print(f"Exception recorded with dual-commander approval: {exception_id}")

# Version history for Inspector General review
history = engine.get_policy_history("pol-opsec-001")
print(f"Policy versions on record: {len(history)}")
  • Decision Intelligence — record_decision(), causal chains, and precedent search — the decisions that check_compliance() evaluates
  • Reasoning & Rules — complement policy rules with formal inference for logical conflict detection
  • SHACL Validation — enforce structural constraints on policy nodes themselves
  • Change Management — version-snapshot the policy graph alongside the knowledge graph
  • Provenance — W3C PROV-O lineage for every policy decision and exception
  • MCP Server — expose record_decision and find_precedents as MCP tools for AI agents