Python API Reference

Top-level exports

import rulegraph

rulegraph

rulegraph - Natural-language rulebook compiler for game arbitration.

EnforcementSession(incident, status='open', inventory=list(), clearance_token=None, blocked_actions=list(), notes='') dataclass

Mutable post-incident lockdown session (AgentWard runtime).

Attributes:

Name	Type	Description
status	IncidentStatus	open/lockdown/cleared/false_alarm.
inventory	list[str]	Recovered inventory of lost/affected resources.
clearance_token	str | None	Human/owner clearance id (required to clear).
blocked_actions	list[str]	High-risk actions attempted while locked.

IncidentEvent(incident_id, kind, paths=(), summary='', agent_id='', timestamp=0.0) dataclass

One destructive / loss event that opens enforcement.

ClosedLoopError

Bases: ValueError

Raised when policy gate refuses empty/conflicted/indeterminate graphs.

GateOutcome(ok, verdict, reason, exit_code, rule_count=0, edge_count=0, conflict_count=0, critical_conflict_count=0, tier=None, confidence=None, provenance=(), human_required=False, mean_logprob=None, min_logprob=None, token_count=0, action=None, brittle_spans=()) dataclass

Result of a closed-loop policy or logprob gate.

Attributes:

Name	Type	Description
ok	bool	True only when the pipeline may continue.
verdict	str	PASS, FAIL, or FAIL_LOUD.
reason	str	Always non-empty.
exit_code	int	0 PASS, 1 FAIL, 2 FAIL_LOUD.
rule_count	int	Nodes in the graph.
edge_count	int	Edges in the graph.
conflict_count	int	Conflicts detected.
critical_conflict_count	int	Severity=critical conflicts.
tier	str | None	Arbitration tier when a query was run.
confidence	float | None	Arbitration confidence or geometric mean token prob.
provenance	tuple[str, ...]	Rule ids used for the answer.
human_required	bool	True when policy needs human arbitration.
mean_logprob	float | None	Mean token logprob when a logprob gate ran.
min_logprob	float | None	Minimum token logprob when a logprob gate ran.
token_count	int	Number of tokens examined by a logprob gate.
action	str | None	Action / step name gated (logprob path).
brittle_spans	tuple[str, ...]	Span names that failed logprob thresholds.

LogprobSummary(token_count, mean_logprob, min_logprob, confidence) dataclass

Aggregate stats over a sequence of token logprobs (AgentUQ class).

CoverageTracker(arbiter)

Wraps a RuleArbiter to track which rules are invoked.

Source code in src/rulegraph/coverage.py

def __init__(self, arbiter: RuleArbiter) -> None:
    self._arbiter = arbiter
    self._query_counts: dict[str, int] = {}

arbitrate(query)

Delegate to arbiter and record which rules were used.

Source code in src/rulegraph/coverage.py

def arbitrate(self, query: str) -> ArbitrationResult:
    """Delegate to arbiter and record which rules were used."""
    result = self._arbiter.query(query)
    for rule_id in result.provenance:
        self._query_counts[rule_id] = self._query_counts.get(rule_id, 0) + 1
    return result

report()

Generate a coverage report.

Source code in src/rulegraph/coverage.py

def report(self) -> RuleCoverage:
    """Generate a coverage report."""
    graph: RuleGraph = self._arbiter.graph
    all_rule_ids = set(graph._nodes.keys())
    total = len(all_rule_ids)
    queried_ids = set(self._query_counts.keys())
    never_queried_ids = all_rule_ids - queried_ids
    rules_queried = len(queried_ids & all_rule_ids)
    coverage_pct = (rules_queried / total * 100.0) if total > 0 else 0.0

    sorted_used = sorted(
        [(rid, count) for rid, count in self._query_counts.items() if rid in all_rule_ids],
        key=lambda x: x[1],
        reverse=True,
    )

    return RuleCoverage(
        total_rules=total,
        rules_queried=rules_queried,
        rules_never_queried=len(never_queried_ids),
        coverage_pct=round(coverage_pct, 2),
        most_used_rules=sorted_used[:10],
        dead_rules=sorted(never_queried_ids),
    )

reset()

Reset query counts.

Source code in src/rulegraph/coverage.py

def reset(self) -> None:
    """Reset query counts."""
    self._query_counts.clear()

ArbitrationResult(query, answer, tier, provenance, confidence, contradictions) dataclass

The structured answer to a query against the rule graph.

Attributes:

Name	Type	Description
query	str	The original question posed.
answer	str	The synthesized answer.
tier	str	Classification - "determinate" | "indeterminate" | "unknown".
provenance	list[str]	List of rule_ids that were used to produce the answer.
confidence	float	Aggregate confidence in [0.0, 1.0].
contradictions	list[str]	rule_ids of rules that conflict with the answer.

to_dict()

Serialize to a JSON-compatible dict.

Source code in src/rulegraph/rule.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to a JSON-compatible dict."""
    return {
        "query": self.query,
        "answer": self.answer,
        "tier": self.tier,
        "provenance": self.provenance,
        "confidence": self.confidence,
        "contradictions": self.contradictions,
    }

from_dict(d) classmethod

Deserialize from a dict produced by to_dict().

Source code in src/rulegraph/rule.py

@classmethod
def from_dict(cls, d: dict[str, Any]) -> ArbitrationResult:
    """Deserialize from a dict produced by to_dict()."""
    return cls(
        query=d["query"],
        answer=d["answer"],
        tier=d["tier"],
        provenance=d.get("provenance", []),
        confidence=d.get("confidence", 0.0),
        contradictions=d.get("contradictions", []),
    )

RuleArbiter(graph)

Query engine that arbitrates questions against a RuleGraph.

Given a natural-language question, the arbiter: 1. Finds relevant rules by matching keywords against rule text, tags, and rule_id. 2. Detects contradictions among the found rules. 3. Classifies the query as determinate or indeterminate based on rule types. 4. Returns a structured ArbitrationResult with full provenance.

Attributes:

Name	Type	Description
graph		The RuleGraph to search.

Source code in src/rulegraph/rule.py

def __init__(self, graph: RuleGraph) -> None:
    self.graph = graph

query(question)

Arbitrate a natural-language question against the rule graph.

Finds relevant rules by keyword matching, detects contradictions, classifies as determinate/indeterminate, and returns a structured result with provenance.

Parameters:

Name	Type	Description	Default
question	str	A natural-language question about the rules.	required

Returns:

Type	Description
ArbitrationResult	An ArbitrationResult with the answer, tier, provenance, and
ArbitrationResult	any detected contradictions.

Source code in src/rulegraph/rule.py

def query(self, question: str) -> ArbitrationResult:
    """Arbitrate a natural-language question against the rule graph.

    Finds relevant rules by keyword matching, detects contradictions,
    classifies as determinate/indeterminate, and returns a structured
    result with provenance.

    Args:
        question: A natural-language question about the rules.

    Returns:
        An ArbitrationResult with the answer, tier, provenance, and
        any detected contradictions.
    """
    keywords = self._extract_keywords(question)
    relevant = self._find_relevant(keywords)

    if not relevant:
        return ArbitrationResult(
            query=question,
            answer="No matching rules found for this query.",
            tier="unknown",
            provenance=[],
            confidence=0.0,
            contradictions=[],
        )

    contradictions = self._detect_contradictions(relevant)
    tier = self._classify(relevant)
    confidence = self._aggregate_confidence(relevant, contradictions)
    answer = self._synthesize_answer(question, relevant, contradictions, tier)

    return ArbitrationResult(
        query=question,
        answer=answer,
        tier=tier,
        provenance=[n.rule_id for n in relevant],
        confidence=confidence,
        contradictions=[n.rule_id for n in contradictions],
    )

RuleEdge(source_id, target_id, relation, condition='', confidence=1.0) dataclass

A directed relationship between two rules.

Edges represent how rules interact: one rule may modify, supersede, require, or be an exception to another.

Attributes:

Name	Type	Description
source_id	str	rule_id of the source RuleNode.
target_id	str	rule_id of the target RuleNode.
relation	str	Type of relationship (e.g. "modifies", "supersedes", "requires", "exception-to").
condition	str	Optional condition under which the edge applies.
confidence	float	Certainty that this edge is correct, in [0.0, 1.0].
id	str	SHA-256[:16] of "{source_id}|{target_id}|{relation}", auto-set.

to_dict()

Serialize to a JSON-compatible dict.

Source code in src/rulegraph/rule.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to a JSON-compatible dict."""
    return {
        "id": self.id,
        "source_id": self.source_id,
        "target_id": self.target_id,
        "relation": self.relation,
        "condition": self.condition,
        "confidence": self.confidence,
    }

from_dict(d) classmethod

Deserialize from a dict produced by to_dict().

Source code in src/rulegraph/rule.py

@classmethod
def from_dict(cls, d: dict[str, Any]) -> RuleEdge:
    """Deserialize from a dict produced by to_dict()."""
    return cls(
        source_id=d["source_id"],
        target_id=d["target_id"],
        relation=d["relation"],
        condition=d.get("condition", ""),
        confidence=d.get("confidence", 1.0),
    )

RuleGraph()

An in-memory directed graph of RuleNode objects connected by RuleEdge objects.

The graph supports efficient lookup by rule_id, tag, node_type, and keyword search in the text field.

Source code in src/rulegraph/rule.py

def __init__(self) -> None:
    self._nodes: dict[str, RuleNode] = {}  # keyed by rule_id
    self._edges: list[RuleEdge] = []

add_node(node)

Add a RuleNode to the graph (idempotent by rule_id).

Source code in src/rulegraph/rule.py

def add_node(self, node: RuleNode) -> None:
    """Add a RuleNode to the graph (idempotent by rule_id)."""
    self._nodes[node.rule_id] = node

add_edge(edge)

Add a RuleEdge to the graph.

Duplicate edges (same source_id, target_id, relation) are silently ignored to keep the graph idempotent.

Source code in src/rulegraph/rule.py

def add_edge(self, edge: RuleEdge) -> None:
    """Add a RuleEdge to the graph.

    Duplicate edges (same source_id, target_id, relation) are silently
    ignored to keep the graph idempotent.
    """
    for existing in self._edges:
        if existing.id == edge.id:
            return
    self._edges.append(edge)

get_node(rule_id)

Return the RuleNode with the given rule_id, or None.

Source code in src/rulegraph/rule.py

def get_node(self, rule_id: str) -> RuleNode | None:
    """Return the RuleNode with the given rule_id, or None."""
    return self._nodes.get(rule_id)

get_edges(source_id=None, relation=None)

Return edges, optionally filtered by source_id and/or relation.

Source code in src/rulegraph/rule.py

def get_edges(
    self,
    source_id: str | None = None,
    relation: str | None = None,
) -> list[RuleEdge]:
    """Return edges, optionally filtered by source_id and/or relation."""
    result = self._edges
    if source_id is not None:
        result = [e for e in result if e.source_id == source_id]
    if relation is not None:
        result = [e for e in result if e.relation == relation]
    return list(result)

find_rules(tag=None, node_type=None, text_contains=None)

Search for rules matching any combination of filters.

Parameters:

Name	Type	Description	Default
tag	str | None	If given, only nodes whose tags list contains this string.	None
node_type	str | None	If given, only nodes of this type.	None
text_contains	str | None	If given, only nodes whose text contains this substring (case-insensitive).	None

Returns:

Type	Description
list[RuleNode]	List of matching RuleNode objects in insertion order.

Source code in src/rulegraph/rule.py

def find_rules(
    self,
    tag: str | None = None,
    node_type: str | None = None,
    text_contains: str | None = None,
) -> list[RuleNode]:
    """Search for rules matching any combination of filters.

    Args:
        tag: If given, only nodes whose tags list contains this string.
        node_type: If given, only nodes of this type.
        text_contains: If given, only nodes whose text contains this
            substring (case-insensitive).

    Returns:
        List of matching RuleNode objects in insertion order.
    """
    result = list(self._nodes.values())
    if tag is not None:
        result = [n for n in result if tag in n.tags]
    if node_type is not None:
        result = [n for n in result if n.node_type == node_type]
    if text_contains is not None:
        lower = text_contains.lower()
        result = [n for n in result if lower in n.text.lower()]
    return result

node_ids()

Return a list of all rule_ids in the graph.

Source code in src/rulegraph/rule.py

def node_ids(self) -> list[str]:
    """Return a list of all rule_ids in the graph."""
    return list(self._nodes.keys())

nodes()

Return a list of all RuleNode objects in the graph.

Source code in src/rulegraph/rule.py

def nodes(self) -> list[RuleNode]:
    """Return a list of all RuleNode objects in the graph."""
    return list(self._nodes.values())

node_count()

Return the number of nodes in the graph.

Source code in src/rulegraph/rule.py

def node_count(self) -> int:
    """Return the number of nodes in the graph."""
    return len(self._nodes)

edge_count()

Return the number of edges in the graph.

Source code in src/rulegraph/rule.py

def edge_count(self) -> int:
    """Return the number of edges in the graph."""
    return len(self._edges)

RuleNode(rule_id, text, node_type, tags=list(), source='', confidence=1.0) dataclass

A single rule extracted from a rulebook, content-addressed by rule_id.

Two RuleNode objects with the same rule_id always share the same id, regardless of when they were created.

Attributes:

Name	Type	Description
rule_id	str	Human-readable identifier (e.g. "PHB.5e.attack_roll").
text	str	The full rule text.
node_type	str	Semantic category (e.g. "mechanic", "definition", "narrative").
tags	list[str]	Free-form labels for filtering (e.g. ["combat", "attack"]).
source	str	Source book or document (e.g. "D&D SRD 5.1").
confidence	float	Certainty that this is a deterministic rule, in [0.0, 1.0].
id	str	SHA-256[:16] of rule_id, set automatically in post_init.

to_dict()

Serialize to a JSON-compatible dict.

Source code in src/rulegraph/rule.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to a JSON-compatible dict."""
    return {
        "id": self.id,
        "rule_id": self.rule_id,
        "text": self.text,
        "node_type": self.node_type,
        "tags": self.tags,
        "source": self.source,
        "confidence": self.confidence,
    }

from_dict(d) classmethod

Deserialize from a dict produced by to_dict().

Source code in src/rulegraph/rule.py

@classmethod
def from_dict(cls, d: dict[str, Any]) -> RuleNode:
    """Deserialize from a dict produced by to_dict()."""
    node = cls(
        rule_id=d["rule_id"],
        text=d["text"],
        node_type=d["node_type"],
        tags=d.get("tags", []),
        source=d.get("source", ""),
        confidence=d.get("confidence", 1.0),
    )
    return node

RuleStore(path)

SQLite-backed persistence layer for RuleNode, RuleEdge, and ArbitrationResult objects.

Attributes:

Name	Type	Description
path		Path to the SQLite database file.

Source code in src/rulegraph/rule.py

def __init__(self, path: str | Path) -> None:
    import time

    self._time = time
    self.path = Path(path)
    self.path.parent.mkdir(parents=True, exist_ok=True)
    self._conn = sqlite3.connect(str(self.path))
    self._conn.row_factory = sqlite3.Row
    self._conn.executescript(self._SCHEMA)
    self._conn.commit()

save_node(node)

Persist a RuleNode (upsert by rule_id).

Source code in src/rulegraph/rule.py

def save_node(self, node: RuleNode) -> None:
    """Persist a RuleNode (upsert by rule_id)."""
    self._conn.execute(
        """INSERT OR REPLACE INTO nodes
           (id, rule_id, text, node_type, tags, source, confidence)
           VALUES (?,?,?,?,?,?,?)""",
        (
            node.id,
            node.rule_id,
            node.text,
            node.node_type,
            json.dumps(node.tags),
            node.source,
            node.confidence,
        ),
    )
    self._conn.commit()

save_edge(edge)

Persist a RuleEdge (upsert by id).

Source code in src/rulegraph/rule.py

def save_edge(self, edge: RuleEdge) -> None:
    """Persist a RuleEdge (upsert by id)."""
    self._conn.execute(
        """INSERT OR REPLACE INTO edges
           (id, source_id, target_id, relation, condition, confidence)
           VALUES (?,?,?,?,?,?)""",
        (
            edge.id,
            edge.source_id,
            edge.target_id,
            edge.relation,
            edge.condition,
            edge.confidence,
        ),
    )
    self._conn.commit()

load_graph()

Load all persisted nodes and edges into a new RuleGraph.

Source code in src/rulegraph/rule.py

def load_graph(self) -> RuleGraph:
    """Load all persisted nodes and edges into a new RuleGraph."""
    graph = RuleGraph()
    for row in self._conn.execute("SELECT * FROM nodes").fetchall():
        d = dict(row)
        d["tags"] = json.loads(d["tags"])
        graph.add_node(RuleNode.from_dict(d))
    for row in self._conn.execute("SELECT * FROM edges").fetchall():
        graph.add_edge(RuleEdge.from_dict(dict(row)))
    return graph

save_result(result)

Persist an ArbitrationResult.

Source code in src/rulegraph/rule.py

def save_result(self, result: ArbitrationResult) -> None:
    """Persist an ArbitrationResult."""
    import time as _time

    self._conn.execute(
        """INSERT INTO results
           (query, answer, tier, provenance, confidence, contradictions, recorded_at)
           VALUES (?,?,?,?,?,?,?)""",
        (
            result.query,
            result.answer,
            result.tier,
            json.dumps(result.provenance),
            result.confidence,
            json.dumps(result.contradictions),
            _time.time(),
        ),
    )
    self._conn.commit()

list_results()

Return all stored ArbitrationResult objects, oldest first.

Source code in src/rulegraph/rule.py

def list_results(self) -> list[ArbitrationResult]:
    """Return all stored ArbitrationResult objects, oldest first."""
    rows = self._conn.execute("SELECT * FROM results ORDER BY recorded_at").fetchall()
    out: list[ArbitrationResult] = []
    for row in rows:
        d = dict(row)
        d["provenance"] = json.loads(d["provenance"])
        d["contradictions"] = json.loads(d["contradictions"])
        out.append(ArbitrationResult.from_dict(d))
    return out

close()

Close the underlying SQLite connection.

Source code in src/rulegraph/rule.py

def close(self) -> None:
    """Close the underlying SQLite connection."""
    self._conn.close()

assert_post_incident_ok(session, **kwargs)

Raise :class:ClosedLoopError unless :func:gate_post_incident is ok.

Source code in src/rulegraph/agentward.py

def assert_post_incident_ok(
    session: EnforcementSession | dict[str, Any] | None,
    **kwargs: Any,
) -> GateOutcome:
    """Raise :class:`ClosedLoopError` unless :func:`gate_post_incident` is ok."""
    outcome = gate_post_incident(session, **kwargs)
    if not outcome.ok:
        raise ClosedLoopError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

clear_incident(session, *, clearance_token, require_inventory=True)

Attempt to clear lockdown with a human/owner token.

Does not clear if inventory is empty when required - status stays locked.

Source code in src/rulegraph/agentward.py

def clear_incident(
    session: EnforcementSession,
    *,
    clearance_token: str,
    require_inventory: bool = True,
) -> EnforcementSession:
    """Attempt to clear lockdown with a human/owner token.

    Does **not** clear if inventory is empty when required - status stays locked.
    """
    token = (clearance_token or "").strip()
    if not token:
        session.notes = "clear refused: empty clearance_token"
        return session
    if require_inventory and not session.inventory:
        session.notes = "clear refused: empty inventory"
        return session
    session.clearance_token = token
    session.status = "cleared"
    session.notes = "cleared"
    return session

gate_post_incident(session, *, proposed_action=None, require_session_after_incident=False, incident_signaled=False, require_inventory=True, block_high_risk_while_locked=True, high_risk_extra=None)

Refuse high-risk continue under open AgentWard lockdown.

Rules:

• incident_signaled and no session when require_session_after_incident → FAIL_LOUD (enforcer missing after known destruction)
• No session and not required → PASS (nothing to enforce)
• Session open/lockdown + empty inventory when required → FAIL_LOUD
• Session locked + high-risk proposed_action → FAIL (record block)
• Session cleared with inventory + token → PASS
• Locked + low-risk / no proposed action → PASS with human_required note only when action is None (observe-only); if action is low-risk → PASS

Parameters:

Name	Type	Description	Default
session	EnforcementSession | dict[str, Any] | None	Open enforcement session or dict; None if no incident yet.	required
proposed_action	str | None	Next agent action name to gate.	None
require_session_after_incident	bool	When True and incident_signaled, missing session is FAIL_LOUD.	False
incident_signaled	bool	External signal that destruction already happened.	False
require_inventory	bool	Empty inventory while locked → FAIL_LOUD.	True
block_high_risk_while_locked	bool	High-risk under lockdown → FAIL.	True

Source code in src/rulegraph/agentward.py

def gate_post_incident(
    session: EnforcementSession | dict[str, Any] | None,
    *,
    proposed_action: str | None = None,
    require_session_after_incident: bool = False,
    incident_signaled: bool = False,
    require_inventory: bool = True,
    block_high_risk_while_locked: bool = True,
    high_risk_extra: Iterable[str] | None = None,
) -> GateOutcome:
    """Refuse high-risk continue under open AgentWard lockdown.

    Rules:

    * ``incident_signaled`` and no session when ``require_session_after_incident``
      → **FAIL_LOUD** (enforcer missing after known destruction)
    * No session and not required → **PASS** (nothing to enforce)
    * Session open/lockdown + empty inventory when required → **FAIL_LOUD**
    * Session locked + high-risk ``proposed_action`` → **FAIL** (record block)
    * Session cleared with inventory + token → **PASS**
    * Locked + low-risk / no proposed action → **PASS** with human_required note
      only when action is None (observe-only); if action is low-risk → **PASS**

    Args:
        session: Open enforcement session or dict; None if no incident yet.
        proposed_action: Next agent action name to gate.
        require_session_after_incident: When True and incident_signaled, missing
            session is FAIL_LOUD.
        incident_signaled: External signal that destruction already happened.
        require_inventory: Empty inventory while locked → FAIL_LOUD.
        block_high_risk_while_locked: High-risk under lockdown → FAIL.
    """
    if session is None:
        if require_session_after_incident and incident_signaled:
            return GateOutcome(
                ok=False,
                verdict="FAIL_LOUD",
                reason=(
                    "AGENTWARD: destructive incident signaled but no EnforcementSession "
                    "- runtime enforcer missing after agent deletion (HN AgentWard class)"
                ),
                exit_code=2,
                human_required=True,
                action=proposed_action,
            )
        return GateOutcome(
            ok=True,
            verdict="PASS",
            reason="AGENTWARD: no open incident session; nothing to enforce",
            exit_code=0,
            human_required=False,
            action=proposed_action,
        )

    if isinstance(session, dict):
        try:
            inc = _as_incident(session.get("incident") or session)
            status = str(session.get("status") or "open").strip().lower()
            if status not in {"open", "lockdown", "cleared", "false_alarm"}:
                status = "open"
            sess = EnforcementSession(
                incident=inc,
                status=status,  # type: ignore[arg-type]
                inventory=list(session.get("inventory") or []),
                clearance_token=session.get("clearance_token"),
                blocked_actions=list(session.get("blocked_actions") or []),
                notes=str(session.get("notes") or ""),
            )
        except (TypeError, ValueError) as exc:
            return GateOutcome(
                ok=False,
                verdict="FAIL_LOUD",
                reason=f"AGENTWARD: invalid session payload: {exc}",
                exit_code=2,
                human_required=True,
                action=proposed_action,
            )
    else:
        sess = session

    if sess.status == "false_alarm":
        return GateOutcome(
            ok=True,
            verdict="PASS",
            reason="AGENTWARD: session marked false_alarm",
            exit_code=0,
            action=proposed_action,
        )

    if sess.status == "cleared":
        if require_inventory and not sess.inventory:
            return GateOutcome(
                ok=False,
                verdict="FAIL_LOUD",
                reason=(
                    "AGENTWARD: status=cleared but inventory empty - clearance "
                    "without loss inventory is not load-bearing"
                ),
                exit_code=2,
                human_required=True,
                action=proposed_action,
            )
        if not (sess.clearance_token or "").strip():
            return GateOutcome(
                ok=False,
                verdict="FAIL_LOUD",
                reason=(
                    "AGENTWARD: status=cleared but clearance_token missing - "
                    "human/owner token required to lift lockdown"
                ),
                exit_code=2,
                human_required=True,
                action=proposed_action,
            )
        return GateOutcome(
            ok=True,
            verdict="PASS",
            reason=(
                f"AGENTWARD ok: incident={sess.incident.incident_id} cleared "
                f"inventory={len(sess.inventory)} token=present"
            ),
            exit_code=0,
            human_required=False,
            action=proposed_action,
        )

    # open / lockdown
    if require_inventory and not sess.inventory:
        return GateOutcome(
            ok=False,
            verdict="FAIL_LOUD",
            reason=(
                f"AGENTWARD: lockdown active for incident={sess.incident.incident_id} "
                f"kind={sess.incident.kind} but inventory is empty - record lost "
                "paths/resources before any further agent work"
            ),
            exit_code=2,
            human_required=True,
            action=proposed_action,
        )

    if (
        proposed_action
        and block_high_risk_while_locked
        and is_high_risk_action(proposed_action, extra=high_risk_extra)
    ):
        sess.blocked_actions.append(proposed_action)
        return GateOutcome(
            ok=False,
            verdict="FAIL",
            reason=(
                f"AGENTWARD: lockdown blocks high-risk action={proposed_action!r} "
                f"after incident={sess.incident.incident_id} "
                f"(inventory={len(sess.inventory)}) - obtain human clearance "
                "via clear_incident before continuing (post-deletion enforcer)"
            ),
            exit_code=1,
            human_required=True,
            action=proposed_action,
        )

    # Locked but only observing / low-risk
    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=(
            f"AGENTWARD: lockdown active incident={sess.incident.incident_id} "
            f"inventory={len(sess.inventory)}; proposed_action="
            f"{proposed_action!r} allowed (not high-risk) - human_required to clear"
        ),
        exit_code=0,
        human_required=True,
        action=proposed_action,
    )

is_high_risk_action(action, *, extra=None)

True if action name matches default or extra high-risk verbs.

Source code in src/rulegraph/agentward.py

def is_high_risk_action(
    action: str,
    *,
    extra: Iterable[str] | None = None,
) -> bool:
    """True if action name matches default or extra high-risk verbs."""
    a = (action or "").strip().lower().replace("-", "_").replace(" ", "_")
    if not a:
        return False
    banned = set(DEFAULT_HIGH_RISK_ACTIONS)
    if extra:
        banned |= {str(x).strip().lower().replace("-", "_") for x in extra}
    if a in banned:
        return True
    return any(a.startswith(b + "_") or a.endswith("_" + b) or b in a.split(".") for b in banned)

open_incident(incident, *, auto_lockdown=True)

Open an AgentWard session from a destructive incident signal.

Source code in src/rulegraph/agentward.py

def open_incident(
    incident: IncidentEvent | dict[str, Any],
    *,
    auto_lockdown: bool = True,
) -> EnforcementSession:
    """Open an AgentWard session from a destructive incident signal."""
    ev = _as_incident(incident)
    status: IncidentStatus = "lockdown" if auto_lockdown else "open"
    # Seed inventory from incident paths when present
    inv = list(ev.paths)
    return EnforcementSession(incident=ev, status=status, inventory=inv)

record_inventory(session, paths, *, replace=False)

Attach or merge post-incident inventory (what was lost / affected).

Source code in src/rulegraph/agentward.py

def record_inventory(
    session: EnforcementSession,
    paths: Iterable[str],
    *,
    replace: bool = False,
) -> EnforcementSession:
    """Attach or merge post-incident inventory (what was lost / affected)."""
    new_paths = [str(p).strip() for p in paths if str(p).strip()]
    if replace:
        session.inventory = new_paths
    else:
        seen = set(session.inventory)
        for p in new_paths:
            if p not in seen:
                session.inventory.append(p)
                seen.add(p)
    return session

assert_arbitration_ok(result, **kwargs)

Raise :class:ClosedLoopError unless arbitration result is ok.

Source code in src/rulegraph/closed_loop.py

def assert_arbitration_ok(result: ArbitrationResult, **kwargs: Any) -> GateOutcome:
    """Raise :class:`ClosedLoopError` unless arbitration result is ok."""
    outcome = gate_arbitration(result, **kwargs)
    if not outcome.ok:
        raise ClosedLoopError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

assert_logprob_ok(logprobs, **kwargs)

Raise :class:ClosedLoopError unless :func:gate_logprob is ok.

Source code in src/rulegraph/closed_loop.py

def assert_logprob_ok(
    logprobs: Sequence[float] | None,
    **kwargs: Any,
) -> GateOutcome:
    """Raise :class:`ClosedLoopError` unless :func:`gate_logprob` is ok."""
    outcome = gate_logprob(logprobs, **kwargs)
    if not outcome.ok:
        raise ClosedLoopError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

assert_policy_ok(graph, **kwargs)

Raise :class:ClosedLoopError unless the policy graph gates clean.

Source code in src/rulegraph/closed_loop.py

def assert_policy_ok(graph: RuleGraph, **kwargs: Any) -> GateOutcome:
    """Raise :class:`ClosedLoopError` unless the policy graph gates clean."""
    outcome = gate_policy_graph(graph, **kwargs)
    if not outcome.ok:
        raise ClosedLoopError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

compile_farm_policy_graph()

Compile Foundry/farm COI + endorse policy rules into a RuleGraph.

POLICY-ARBITRATION product artifact - not a README list. Rules:

• COI: no self-endorsement when financial interest exists
• COI: disclose material interest before public statements
• Endorse: require owner/review before public endorse
• Endorse: never auto-endorse without HITL
• Legal: never auto-fix legal gates (cross-ref worldoracle LEGAL-NO-AUTOFIX)

Edges encode requires/modifies relationships for arbitration provenance.

Source code in src/rulegraph/closed_loop.py

def compile_farm_policy_graph() -> RuleGraph:
    """Compile Foundry/farm COI + endorse policy rules into a RuleGraph.

    POLICY-ARBITRATION product artifact - not a README list. Rules:

    * COI: no self-endorsement when financial interest exists
    * COI: disclose material interest before public statements
    * Endorse: require owner/review before public endorse
    * Endorse: never auto-endorse without HITL
    * Legal: never auto-fix legal gates (cross-ref worldoracle LEGAL-NO-AUTOFIX)

    Edges encode requires/modifies relationships for arbitration provenance.
    """
    g = RuleGraph()

    nodes = [
        RuleNode(
            rule_id="FARM.coi.no_self_endorse",
            text=(
                "Conflict of interest: an agent or person with a material financial "
                "interest in a project MUST NOT publicly endorse that project as if "
                "independent. Self-endorsement under COI is prohibited."
            ),
            node_type="policy",
            tags=["coi", "endorse", "conflict", "farm"],
            source="farm_memory:POLICY-ARBITRATION",
            confidence=1.0,
        ),
        RuleNode(
            rule_id="FARM.coi.disclose_interest",
            text=(
                "Conflict of interest: before any public endorsement or recommendation, "
                "material interests MUST be disclosed. Hidden interest voids endorse."
            ),
            node_type="policy",
            tags=["coi", "disclose", "endorse", "farm"],
            source="farm_memory:POLICY-ARBITRATION",
            confidence=1.0,
        ),
        RuleNode(
            rule_id="FARM.endorse.require_review",
            text=(
                "Endorsement: public endorse / recommend / testimonial requires "
                "owner or designated reviewer approval before publication."
            ),
            node_type="policy",
            tags=["endorse", "review", "hitl", "farm"],
            source="farm_memory:POLICY-ARBITRATION",
            confidence=1.0,
        ),
        RuleNode(
            rule_id="FARM.endorse.no_auto",
            text=(
                "Endorsement: NEVER auto-endorse from agent pipelines without "
                "human-in-the-loop. Unattended endorse is denied."
            ),
            node_type="policy",
            tags=["endorse", "auto", "hitl", "farm"],
            source="farm_memory:POLICY-ARBITRATION",
            confidence=1.0,
        ),
        RuleNode(
            rule_id="FARM.legal.no_autofix",
            text=(
                "Legal gates (G-CONTRA and similar) ALWAYS stop for human review. "
                "NEVER auto-fix legal contradictions."
            ),
            node_type="policy",
            tags=["legal", "coi", "human_required", "farm"],
            source="farm_memory:LEGAL-NO-AUTOFIX",
            confidence=1.0,
        ),
        RuleNode(
            rule_id="FARM.action.require_provenance",
            text=(
                "Policy arbitration answers for allow/deny MUST include provenance "
                "(rule ids). Answers without provenance are indeterminate."
            ),
            node_type="policy",
            tags=["arbitration", "provenance", "farm"],
            source="farm_memory:POLICY-ARBITRATION",
            confidence=1.0,
        ),
    ]
    for n in nodes:
        g.add_node(n)

    edges = [
        RuleEdge(
            "FARM.coi.no_self_endorse",
            "FARM.endorse.require_review",
            "requires",
            condition="when endorsing under possible COI",
        ),
        RuleEdge(
            "FARM.coi.disclose_interest",
            "FARM.endorse.require_review",
            "requires",
            condition="before public endorse",
        ),
        RuleEdge(
            "FARM.endorse.no_auto",
            "FARM.endorse.require_review",
            "modifies",
            condition="blocks unattended path",
        ),
        RuleEdge(
            "FARM.legal.no_autofix",
            "FARM.action.require_provenance",
            "requires",
            condition="legal decisions need human + provenance",
        ),
        RuleEdge(
            "FARM.endorse.require_review",
            "FARM.action.require_provenance",
            "requires",
            condition="endorse decisions need provenance",
        ),
    ]
    for e in edges:
        g.add_edge(e)

    return g

gate_arbitration(result, *, require_determinate=False, require_provenance=True, min_confidence=0.0, refuse_contradictions=True)

Gate an :class:ArbitrationResult for load-bearing policy decisions.

• tier=unknown with require_provenance or require_determinate → FAIL
• empty provenance when required → FAIL (POLICY-ARBITRATION)
• contradictions when refuse_contradictions → FAIL
• confidence below min → FAIL

Source code in src/rulegraph/closed_loop.py

def gate_arbitration(
    result: ArbitrationResult,
    *,
    require_determinate: bool = False,
    require_provenance: bool = True,
    min_confidence: float = 0.0,
    refuse_contradictions: bool = True,
) -> GateOutcome:
    """Gate an :class:`ArbitrationResult` for load-bearing policy decisions.

    * ``tier=unknown`` with require_provenance or require_determinate → FAIL
    * empty provenance when required → FAIL (POLICY-ARBITRATION)
    * contradictions when refuse_contradictions → FAIL
    * confidence below min → FAIL
    """
    prov = tuple(result.provenance or [])
    conf = float(result.confidence)
    tier = result.tier

    if require_provenance and not prov:
        return _fail(
            f"POLICY-ARBITRATION: no provenance for query {result.query!r} "
            f"(tier={tier}) - refuse determinate allow/deny without rule ids",
            tier=tier,
            confidence=conf,
            provenance=prov,
        )

    if require_determinate and tier != "determinate":
        return _fail(
            f"POLICY-ARBITRATION: require_determinate but tier={tier!r} for query {result.query!r}",
            tier=tier,
            confidence=conf,
            provenance=prov,
            human_required=True,
        )

    if refuse_contradictions and result.contradictions:
        return _fail(
            f"POLICY-ARBITRATION: contradictions in provenance "
            f"{result.contradictions} for query {result.query!r}",
            tier=tier,
            confidence=conf,
            provenance=prov,
            conflict_count=len(result.contradictions),
        )

    if conf < min_confidence:
        return _fail(
            f"POLICY-ARBITRATION: confidence {conf:.3f} < min {min_confidence}",
            tier=tier,
            confidence=conf,
            provenance=prov,
        )

    if tier == "unknown" and not prov:
        return _fail_loud(
            f"no matching policy rules for {result.query!r}",
            tier=tier,
            confidence=conf,
        )

    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=f"arbitration ok tier={tier} confidence={conf:.3f} provenance={list(prov)}",
        exit_code=0,
        tier=tier,
        confidence=conf,
        provenance=prov,
        rule_count=len(prov),
        human_required=False,
    )

gate_logprob(logprobs, *, min_mean_logprob=DEFAULT_MIN_MEAN_LOGPROB, min_token_logprob=DEFAULT_MIN_TOKEN_LOGPROB, require_logprobs=True, high_risk=False, action=None, spans=None)

Block brittle LLM steps using token logprobs (AgentUQ / LOGPROB-GATE).

Public case: AgentUQ (HN Show HN) - single-pass runtime reliability gate from provider logprobs. Does not claim truth; refuses execution when the generation looks ambiguous/brittle, especially on high-risk tools (SQL, shell, paths, tool JSON).

Rules:

• require_logprobs and empty/missing logprobs → FAIL_LOUD (cannot gate a phantom confidence signal).
• mean logprob < min_mean_logprob → FAIL (human_required if high_risk).
• any token logprob < min_token_logprob → FAIL.
• optional spans: each named span (e.g. sql_clause, tool_args) is checked with the same thresholds; failing span names appear in brittle_spans.
• clean tokens → PASS with confidence = exp(mean_logprob).

Parameters:

Name	Type	Description	Default
logprobs	Sequence[float] | None	Full-step token logprobs (may be None if only spans given).	required
min_mean_logprob	float	Minimum allowed mean logprob (default -1.5).	DEFAULT_MIN_MEAN_LOGPROB
min_token_logprob	float	Minimum allowed single-token logprob (default -4.0).	DEFAULT_MIN_TOKEN_LOGPROB
require_logprobs	bool	If True, missing/empty tokens FAIL_LOUD.	True
high_risk	bool	If True, failures set human_required (tool exec class).	False
action	str | None	Optional step/tool name for the reason string.	None
spans	Mapping[str, Sequence[float]] | None	Optional map of span_name → token logprobs for localization.	None

Source code in src/rulegraph/closed_loop.py

def gate_logprob(
    logprobs: Sequence[float] | None,
    *,
    min_mean_logprob: float = DEFAULT_MIN_MEAN_LOGPROB,
    min_token_logprob: float = DEFAULT_MIN_TOKEN_LOGPROB,
    require_logprobs: bool = True,
    high_risk: bool = False,
    action: str | None = None,
    spans: Mapping[str, Sequence[float]] | None = None,
) -> GateOutcome:
    """Block brittle LLM steps using token logprobs (AgentUQ / LOGPROB-GATE).

    Public case: AgentUQ (HN Show HN) - single-pass runtime reliability gate
    from provider logprobs. Does **not** claim truth; refuses execution when
    the generation looks ambiguous/brittle, especially on high-risk tools
    (SQL, shell, paths, tool JSON).

    Rules:

    * ``require_logprobs`` and empty/missing logprobs → **FAIL_LOUD**
      (cannot gate a phantom confidence signal).
    * mean logprob < ``min_mean_logprob`` → **FAIL** (``human_required`` if
      high_risk).
    * any token logprob < ``min_token_logprob`` → **FAIL**.
    * optional ``spans``: each named span (e.g. ``sql_clause``, ``tool_args``)
      is checked with the same thresholds; failing span names appear in
      ``brittle_spans``.
    * clean tokens → **PASS** with ``confidence = exp(mean_logprob)``.

    Args:
        logprobs: Full-step token logprobs (may be None if only spans given).
        min_mean_logprob: Minimum allowed mean logprob (default -1.5).
        min_token_logprob: Minimum allowed single-token logprob (default -4.0).
        require_logprobs: If True, missing/empty tokens FAIL_LOUD.
        high_risk: If True, failures set ``human_required`` (tool exec class).
        action: Optional step/tool name for the reason string.
        spans: Optional map of span_name → token logprobs for localization.
    """
    act = (action or "").strip()
    brittle: list[str] = []

    # Collect all sequences to evaluate: main sequence + named spans.
    sequences: list[tuple[str | None, Sequence[float]]] = []
    if logprobs is not None:
        sequences.append((None, logprobs))
    if spans:
        for span_key, seq in spans.items():
            sequences.append((str(span_key), seq))

    if not sequences:
        if require_logprobs or high_risk:
            return _fail_loud(
                "LOGPROB-GATE/AgentUQ: no logprobs provided "
                f"(action={act!r} high_risk={high_risk}) - "
                "cannot gate brittle tool steps without provider logprobs",
                human_required=True,
                action=act,
                token_count=0,
            )
        return GateOutcome(
            ok=True,
            verdict="PASS",
            reason=f"LOGPROB-GATE: logprobs not required action={act!r}",
            exit_code=0,
            human_required=False,
            action=act,
        )

    # Empty any required sequence → FAIL_LOUD
    for seq_name, seq in sequences:
        if seq is None or len(list(seq)) == 0:
            label = seq_name or "step"
            if require_logprobs or high_risk:
                return _fail_loud(
                    f"LOGPROB-GATE/AgentUQ: empty logprobs for {label!r} "
                    f"(action={act!r}) - missing confidence signal",
                    human_required=True,
                    action=act,
                    token_count=0,
                    brittle_spans=(seq_name,) if seq_name else (),
                )

    # Evaluate main sequence (or first span if only spans).
    _primary_name, primary_seq = sequences[0]
    primary_vals = [float(x) for x in primary_seq]
    summary = summarize_logprobs(primary_vals)

    # Span checks (localize brittle tool args / SQL clauses).
    for seq_name, seq in sequences:
        if seq_name is None:
            continue
        vals = [float(x) for x in seq]
        if not vals:
            continue
        s = summarize_logprobs(vals)
        if s.mean_logprob < min_mean_logprob or s.min_logprob < min_token_logprob:
            brittle.append(seq_name)

    fail_mean = summary.mean_logprob < min_mean_logprob
    fail_min = summary.min_logprob < min_token_logprob

    if fail_mean or fail_min or brittle:
        parts: list[str] = []
        if fail_mean:
            parts.append(f"mean_logprob={summary.mean_logprob:.4f} < {min_mean_logprob}")
        if fail_min:
            parts.append(f"min_logprob={summary.min_logprob:.4f} < {min_token_logprob}")
        if brittle:
            parts.append(f"brittle_spans={brittle}")
        reason = (
            "LOGPROB-GATE/AgentUQ: brittle generation - "
            + "; ".join(parts)
            + f" action={act!r} tokens={summary.token_count}"
            + (" - refuse high-risk tool exec" if high_risk else "")
        )
        return _fail(
            reason,
            confidence=summary.confidence,
            human_required=bool(high_risk),
            mean_logprob=summary.mean_logprob,
            min_logprob=summary.min_logprob,
            token_count=summary.token_count,
            action=act,
            brittle_spans=tuple(brittle),
        )

    span_note = f" spans_ok={list(spans.keys())}" if spans else ""
    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=(
            f"LOGPROB-GATE ok mean={summary.mean_logprob:.4f} "
            f"min={summary.min_logprob:.4f} conf={summary.confidence:.4f} "
            f"tokens={summary.token_count} action={act!r}{span_note}"
        ),
        exit_code=0,
        confidence=summary.confidence,
        human_required=False,
        mean_logprob=summary.mean_logprob,
        min_logprob=summary.min_logprob,
        token_count=summary.token_count,
        action=act,
        brittle_spans=(),
    )

gate_policy_graph(graph, *, refuse_critical_conflicts=True, min_rules=1)

Gate a rule graph for POLICY-ARBITRATION readiness.

• Empty / below min_rules → FAIL_LOUD
• Critical conflicts (cycles, mutual supersede) → FAIL
• Otherwise PASS (warning-level overlaps do not fail by default)

Source code in src/rulegraph/closed_loop.py

def gate_policy_graph(
    graph: RuleGraph,
    *,
    refuse_critical_conflicts: bool = True,
    min_rules: int = 1,
) -> GateOutcome:
    """Gate a rule graph for POLICY-ARBITRATION readiness.

    * Empty / below min_rules → FAIL_LOUD
    * Critical conflicts (cycles, mutual supersede) → FAIL
    * Otherwise PASS (warning-level overlaps do not fail by default)
    """
    n = graph.node_count()
    e = graph.edge_count()
    if n < min_rules:
        return _fail_loud(
            f"empty policy graph - {n} rules (<{min_rules}); "
            f"write-only rulebook is ornament (POLICY-ARBITRATION)",
            rule_count=n,
            edge_count=e,
        )

    conflicts = detect_conflicts(graph)
    critical = [c for c in conflicts if c.severity == "critical"]
    if refuse_critical_conflicts and critical:
        kinds = sorted({c.conflict_type for c in critical})
        return _fail(
            f"POLICY-ARBITRATION: {len(critical)} critical conflict(s) "
            f"types={kinds} - refuse allow/deny until resolved",
            rule_count=n,
            edge_count=e,
            conflict_count=len(conflicts),
            critical_conflict_count=len(critical),
        )

    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=(
            f"policy graph ok: rules={n} edges={e} conflicts={len(conflicts)} "
            f"critical={len(critical)}"
        ),
        exit_code=0,
        rule_count=n,
        edge_count=e,
        conflict_count=len(conflicts),
        critical_conflict_count=len(critical),
        human_required=False,
    )

gate_policy_query(graph, query, *, require_determinate=False, require_provenance=True, min_confidence=0.0, refuse_critical_conflicts=True)

End-to-end: graph readiness + arbitrate query + gate the result.

This is the load-bearing closed-loop entry for POLICY-ARBITRATION.

Source code in src/rulegraph/closed_loop.py

def gate_policy_query(
    graph: RuleGraph,
    query: str,
    *,
    require_determinate: bool = False,
    require_provenance: bool = True,
    min_confidence: float = 0.0,
    refuse_critical_conflicts: bool = True,
) -> GateOutcome:
    """End-to-end: graph readiness + arbitrate *query* + gate the result.

    This is the load-bearing closed-loop entry for POLICY-ARBITRATION.
    """
    base = gate_policy_graph(
        graph,
        refuse_critical_conflicts=refuse_critical_conflicts,
    )
    if not base.ok:
        return base

    arbiter = RuleArbiter(graph)
    result = arbiter.query(query)
    out = gate_arbitration(
        result,
        require_determinate=require_determinate,
        require_provenance=require_provenance,
        min_confidence=min_confidence,
    )
    # Attach graph stats
    return GateOutcome(
        ok=out.ok,
        verdict=out.verdict,
        reason=out.reason,
        exit_code=out.exit_code,
        rule_count=base.rule_count,
        edge_count=base.edge_count,
        conflict_count=base.conflict_count,
        critical_conflict_count=base.critical_conflict_count,
        tier=out.tier,
        confidence=out.confidence,
        provenance=out.provenance,
        human_required=out.human_required,
    )

list_critical_conflicts(graph)

Return only critical conflicts (for CI reports).

Source code in src/rulegraph/closed_loop.py

def list_critical_conflicts(graph: RuleGraph) -> list[RuleConflict]:
    """Return only critical conflicts (for CI reports)."""
    return [c for c in detect_conflicts(graph) if c.severity == "critical"]

summarize_logprobs(logprobs)

Compute mean/min logprob and geometric-mean confidence from token logprobs.

Parameters:

Name	Type	Description	Default
logprobs	Sequence[float]	Per-token natural log-probabilities (typically ≤ 0).	required

Raises:

Type	Description
ValueError	if logprobs is empty.

Source code in src/rulegraph/closed_loop.py

def summarize_logprobs(logprobs: Sequence[float]) -> LogprobSummary:
    """Compute mean/min logprob and geometric-mean confidence from token logprobs.

    Args:
        logprobs: Per-token natural log-probabilities (typically ≤ 0).

    Raises:
        ValueError: if *logprobs* is empty.
    """
    if not logprobs:
        raise ValueError("empty logprobs - cannot summarize")
    vals = [float(x) for x in logprobs]
    mean_lp = sum(vals) / len(vals)
    min_lp = min(vals)
    # Clamp exp for numerical safety on extremely negative logprobs.
    conf = math.exp(max(mean_lp, -50.0))
    return LogprobSummary(
        token_count=len(vals),
        mean_logprob=mean_lp,
        min_logprob=min_lp,
        confidence=conf,
    )

detect_conflicts(graph)

Find rules that may contradict or circularly depend on each other.

Source code in src/rulegraph/conflicts.py

def detect_conflicts(graph: RuleGraph) -> list[RuleConflict]:
    """Find rules that may contradict or circularly depend on each other."""
    conflicts: list[RuleConflict] = []

    # 1. Circular dependencies
    cycles = find_cycles(graph)
    seen_cycle_pairs: set[frozenset[str]] = set()
    for cycle in cycles:
        for i in range(len(cycle)):
            a = cycle[i]
            b = cycle[(i + 1) % len(cycle)]
            pair = frozenset([a, b])
            if pair not in seen_cycle_pairs:
                seen_cycle_pairs.add(pair)
                conflicts.append(
                    RuleConflict(
                        rule_a_id=a,
                        rule_b_id=b,
                        conflict_type="circular_dependency",
                        description=(
                            "Circular dependency detected: " + " -> ".join([*cycle, cycle[0]])
                        ),
                        severity="critical",
                    )
                )

    # 2. Direct contradictions - rules that both supersede each other
    edges = graph.get_edges()
    supersedes_map: dict[str, set[str]] = {}
    for edge in edges:
        if edge.relation in ("supersedes", "exception-to"):
            supersedes_map.setdefault(edge.source_id, set()).add(edge.target_id)

    rule_ids = graph.node_ids()
    for i, a in enumerate(rule_ids):
        for b in rule_ids[i + 1 :]:
            a_supersedes_b = b in supersedes_map.get(a, set())
            b_supersedes_a = a in supersedes_map.get(b, set())
            if a_supersedes_b and b_supersedes_a:
                conflicts.append(
                    RuleConflict(
                        rule_a_id=a,
                        rule_b_id=b,
                        conflict_type="direct_contradiction",
                        description=f"Rules '{a}' and '{b}' mutually supersede each other.",
                        severity="critical",
                    )
                )

    # 3. Overlapping scope - same tags + different node_type
    nodes = graph.nodes()
    for i, a_node in enumerate(nodes):
        for b_node in nodes[i + 1 :]:
            if a_node.node_type != b_node.node_type:
                shared_tags = set(a_node.tags) & set(b_node.tags)
                if shared_tags:
                    conflicts.append(
                        RuleConflict(
                            rule_a_id=a_node.rule_id,
                            rule_b_id=b_node.rule_id,
                            conflict_type="overlapping_scope",
                            description=(
                                f"Rules '{a_node.rule_id}' ({a_node.node_type}) and "
                                f"'{b_node.rule_id}' ({b_node.node_type}) share tags: "
                                f"{', '.join(sorted(shared_tags))}"
                            ),
                            severity="warning",
                        )
                    )

    return conflicts

find_cycles(graph)

Find circular dependencies in the rule graph using DFS.

Source code in src/rulegraph/conflicts.py

def find_cycles(graph: RuleGraph) -> list[list[str]]:
    """Find circular dependencies in the rule graph using DFS."""
    # Build adjacency: only 'requires' edges form dependencies
    adj: dict[str, list[str]] = {rid: [] for rid in graph.node_ids()}
    for edge in graph.get_edges():
        if edge.relation == "requires":
            adj.setdefault(edge.source_id, []).append(edge.target_id)

    visited: set[str] = set()
    rec_stack: set[str] = set()
    cycles: list[list[str]] = []

    def dfs(node: str, path: list[str]) -> None:
        visited.add(node)
        rec_stack.add(node)
        path.append(node)
        for neighbor in adj.get(node, []):
            if neighbor not in visited:
                dfs(neighbor, path)
            elif neighbor in rec_stack:
                # Found a cycle
                cycle_start = path.index(neighbor)
                cycles.append(path[cycle_start:])
        path.pop()
        rec_stack.discard(node)

    for rule_id in graph.node_ids():
        if rule_id not in visited:
            dfs(rule_id, [])

    return cycles

import_from_file(path, source='')

Read a text file and parse it into RuleNodes.

Source code in src/rulegraph/importer.py

def import_from_file(path: Path, source: str = "") -> list[RuleNode]:
    """Read a text file and parse it into RuleNodes."""
    path = Path(path)
    text = path.read_text(encoding="utf-8")
    if not source:
        source = path.stem
    return import_from_text(text, source=source)

import_from_text(text, source='', default_type='mechanic')

Parse plain text into RuleNodes.

Each line starting with '- ', '* ', or a number+dot becomes a RuleNode. Tags are auto-extracted from [bracket] patterns. Rule ID is auto-generated as source + SHA hash of the line text.

Source code in src/rulegraph/importer.py

def import_from_text(text: str, source: str = "", default_type: str = "mechanic") -> list[RuleNode]:
    """Parse plain text into RuleNodes.

    Each line starting with '- ', '* ', or a number+dot becomes a RuleNode.
    Tags are auto-extracted from [bracket] patterns.
    Rule ID is auto-generated as source + SHA hash of the line text.
    """
    nodes: list[RuleNode] = []
    bullet_pattern = re.compile(r"^(?:[-*]|\d+[.)]) +(.+)$")

    for line in text.splitlines():
        line = line.strip()
        if not line:
            continue
        m = bullet_pattern.match(line)
        if not m:
            continue
        content = m.group(1).strip()

        # Extract tags from [tag] patterns
        tags = re.findall(r"\[([^\]]+)\]", content)
        # Remove tag annotations from the rule text
        clean_text = re.sub(r"\[[^\]]+\]", "", content).strip()
        if not clean_text:
            continue

        rule_hash = _sha16(content)
        prefix = (source + ".") if source else ""
        rule_id = f"{prefix}{rule_hash}"

        node = RuleNode(
            rule_id=rule_id,
            text=clean_text,
            node_type=default_type,
            tags=tags,
            source=source,
            confidence=1.0,
        )
        nodes.append(node)

    return nodes

infer_edges(rules)

Heuristically infer edges by looking for keyword references between rules.

For each rule, scan its text for keywords (modifies, supersedes, requires, exception) and cross-reference against other rules whose rule_id or tags appear in the text.

Source code in src/rulegraph/importer.py

def infer_edges(rules: list[RuleNode]) -> list[RuleEdge]:
    """Heuristically infer edges by looking for keyword references between rules.

    For each rule, scan its text for keywords (modifies, supersedes, requires, exception)
    and cross-reference against other rules whose rule_id or tags appear in the text.
    """
    edges: list[RuleEdge] = []
    # Also index by short hash suffix for easier matching
    short_map: dict[str, RuleNode] = {}
    for r in rules:
        parts = r.rule_id.split(".")
        for part in parts:
            short_map.setdefault(part.lower(), r)

    for source_rule in rules:
        text_lower = source_rule.text.lower()
        for pattern, relation in _KEYWORD_RELATIONS:
            if not re.search(pattern, text_lower):
                continue
            # Look for any other rule whose rule_id or tags are mentioned
            for target_rule in rules:
                if target_rule.rule_id == source_rule.rule_id:
                    continue
                # Check if target rule_id or any tag is mentioned in source text
                mentioned = target_rule.rule_id.lower() in text_lower or any(
                    tag.lower() in text_lower for tag in target_rule.tags
                )
                if mentioned:
                    edge = RuleEdge(
                        source_id=source_rule.rule_id,
                        target_id=target_rule.rule_id,
                        relation=relation,
                    )
                    edges.append(edge)

    return edges