Engine Internals¶

This document describes the internal structure of the HsrsEngine class (also referred to as MistaberEngine), the main interface for halachic reasoning in Mistaber.

Engine Class Structure¶

Location: mistaber/engine/engine.py

Class Overview¶

class HsrsEngine:
    """
    Main engine for halachic symbolic reasoning.

    Provides query interface for multi-world reasoning using
    Kripke semantics.
    """

Constructor¶

def __init__(
    self,
    ontology_path: Path,
    default_world: str = "base"
) -> None

Parameters: - ontology_path: Path to ontology directory (containing schema/, base/, worlds/) - default_world: Default world for queries (default: "base")

Initialization Process: 1. Store ontology path and default world 2. Create WorldManager instance for Kripke graph management 3. Initialize Clingo control (_ctl) as None 4. Set loaded flag to False 5. Initialize CompletenessChecker lazily 6. Create default EngineConfig 7. Call _load_ontology() to load all files

Instance Attributes¶

Attribute	Type	Description
`ontology_path`	`Path`	Path to ontology directory
`default_world`	`str`	Default world for queries
`_world_manager`	`WorldManager`	Kripke world DAG manager
`_ctl`	`clingo.Control`	Clingo solver control
`_is_loaded`	`bool`	Whether ontology is loaded
`_completeness_checker`	`CompletenessChecker`	OWA completeness checker (lazy)
`_config`	`EngineConfig`	Runtime configuration

Public Methods¶

is_loaded (property)¶

@property
def is_loaded(self) -> bool:
    """Return True if ontology is loaded."""

configure()¶

def configure(
    self,
    world: Optional[str] = None,
    interpretation_precedence: Optional[List[str]] = None,
    safek_resolution: Optional[str] = None,
    context: Optional[str] = None,
    policy: Optional[Dict[str, str]] = None
) -> "HsrsEngine"

Configure engine runtime settings. Returns self for method chaining.

Parameters: - world: Default world for queries - interpretation_precedence: Ordered list of commentators (first = lowest priority) - safek_resolution: How to resolve doubt ("chumra", "kula", "madrega") - context: Normative context ("lechatchila", "bediavad") - policy: Policy overrides dict

Example:

engine.configure(
    world="rema",
    interpretation_precedence=["taz", "shach"],
    safek_resolution="madrega",
    context="lechatchila",
    policy={"safek_issur": "strict", "shiur_system": "chazon_ish"}
)

ask()¶

def ask(self, atom: str, world: Optional[str] = None) -> bool

Boolean query - does this atom hold in any model?

Parameters: - atom: Ground atom like "world(base)" or pattern with variables - world: Optional world context (for future use)

Returns: True if atom holds, False otherwise

Logic: 1. Parse predicate name and arguments from atom string 2. If ground atom (no variables), check exact match in model 3. If pattern with variables, check if any results from query()

Example:

engine.ask("world(base)")  # True
engine.ask("forbidden(base, achiila, m1, ctx_normal)")  # depends on model

query()¶

def query(
    self,
    pattern: str,
    policy: Optional[Dict[str, str]] = None
) -> List[Dict[str, Any]]

Query for atoms matching a pattern, returning variable bindings.

Parameters: - pattern: ASP atom pattern like "world(X)" or "forbidden(W, A, F, C)" - policy: Optional policy overrides (merges with config policy)

Returns: List of dicts with variable bindings

Example:

results = engine.query("world(W)")
# Returns: [{"W": "base"}, {"W": "mechaber"}, {"W": "rema"}, ...]

results = engine.query("holds(issur(achiila, M, Madrega), W)")
# Returns: [{"M": "m1", "Madrega": "d_oraita", "W": "mechaber"}, ...]

compare()¶

def compare(
    self,
    pattern: str,
    worlds: Optional[List[str]] = None,
    policy: Optional[Dict[str, str]] = None
) -> Dict[str, List[Dict[str, Any]]]

Compare query results across multiple worlds.

Parameters: - pattern: ASP atom pattern - worlds: List of worlds to compare (default: all worlds) - policy: Optional policy overrides

Returns: Dict mapping world name to list of results

Example:

comparison = engine.compare(
    pattern="holds(issur(achiila, M, Madrega), W)",
    worlds=["mechaber", "rema"]
)
# Returns:
# {
#     "mechaber": [{"M": "fish_cream", "Madrega": "sakana", "_world": "mechaber"}],
#     "rema": []  # No issur in rema's world
# }

analyze()¶

def analyze(
    self,
    scenario: str,
    world: Optional[str] = None,
    policy: Optional[Dict[str, str]] = None
) -> Dict[str, Any]

Analyze a scenario in push mode - add facts and see what conclusions derive.

Parameters: - scenario: ASP code describing the scenario (facts and rules) - world: World context for analysis (default: default_world) - policy: Optional policy overrides

Returns: Dict with "world", "atoms", and "scenario" keys

Implementation: 1. Creates a fresh Clingo control (isolated from main engine) 2. Loads all ontology files 3. Adds scenario code 4. Grounds and solves 5. Returns all derived atoms

Example:

analysis = engine.analyze("""
    mixture(m1).
    contains(m1, salmon).
    contains(m1, cream).
    food_type(salmon, dag).
    food_type(cream, chalav).
""", world="rema")

# Returns:
# {
#     "world": "rema",
#     "atoms": ["mixture(m1)", "heter(achiila, m1)", ...],
#     "scenario": "mixture(m1). contains(m1, salmon). ..."
# }

query_preferred()¶

def query_preferred(
    self,
    pattern: str,
    world: Optional[str] = None
) -> List[Dict[str, Any]]

Query with asprin preference optimization, returning only optimal outcomes.

Preference Hierarchy (defined in preferences.lp): 1. Madrega strength: d_oraita > d_rabanan > minhag > chumra 2. Safek handling: Stringent for biblical, lenient for rabbinic 3. World priority: Explicit priority declarations 4. Specificity / quality tie-breaker: Prefer rules with more direct sources (makor)

Parameters: - pattern: ASP atom pattern - world: Optional world context

Returns: List of optimal results according to preference ordering

Implementation: 1. Locate asprin via mistaber.engine.external_tools.find_tool("asprin") and verify it runs 2. If not available, fall back to regular query() 3. Build the file list using real file paths (preserves #include resolution) 4. Create a small temp .lp file containing runtime config facts plus #show. (so matching does not depend on existing #show directives) 5. Run asprin with -q 1 -n 1 to compute a single optimal model 6. Parse the final model's atoms and match pattern using the engine's query_eval.query_atoms()

explain()¶

def explain(
    self,
    atom: str,
    world: Optional[str] = None,
    additional_facts: str = ""
) -> Dict[str, Any]

Generate an explanation for why an atom holds using xclingo2.

Parameters: - atom: The atom to explain (e.g., "world(base)", "forbidden(W,A,S,C)") - world: World context (default: default_world) - additional_facts: Additional ASP facts to include

Returns: Dict containing: - atom: The queried atom - world: The world context - using_xclingo: Whether xclingo was used - derivation: Dict representation of derivation tree - tree: Human-readable text representation - raw_output: Raw xclingo output (if available)

Implementation: Delegates to XclingoExplainer class.

check_completeness()¶

def check_completeness(
    self,
    query: str,
    facts: Union[Set[str], List[str]]
) -> Tuple[bool, List[MissingFact]]

Check if all required OWA predicates have known values for the query.

Parameters: - query: ASP atom pattern (e.g., "is_kosher(chicken)") - facts: Set or list of fact strings

Returns: Tuple of (is_complete, missing_facts)

Example:

facts = {"is_food(chicken).", "food_cat(chicken, basar)."}
is_complete, missing = engine.check_completeness("is_kosher(chicken)", facts)
if not is_complete:
    for mf in missing:
        print(f"Need to know: {mf.question}")

query_with_completeness()¶

def query_with_completeness(
    self,
    pattern: str,
    world: Optional[str] = None,
    require_complete: bool = False,
    additional_facts: Optional[Union[Set[str], List[str]]] = None
) -> Dict[str, Any]

Query with completeness checking for OWA predicates.

Parameters: - pattern: ASP atom pattern - world: World context - require_complete: If True, return empty results when incomplete - additional_facts: Optional additional facts

Returns: Dict with complete, results, missing, world, query keys

Private Methods¶

_load_ontology()¶

def _load_ontology(self) -> None

Load all ontology files into Clingo control.

Loading Order: 1. Schema files (ontology/schema/*.lp) 2. Base ontology files (ontology/base/*.lp) 3. World definitions (ontology/worlds/*.lp) 4. Safek rules (engine/safek.lp) 5. Priority/override rules (engine/priorities.lp) 6. Interpretation layer (engine/interpretations.lp) 7. Individual commentator interpretations (ontology/interpretations/*.lp)

After loading, grounds the program with base part.

_check_asprin_available()¶

def _check_asprin_available(self) -> bool

Check if asprin is available on the system by running asprin --version.

_parse_asprin_output()¶

def _parse_asprin_output(
    self,
    output: str,
    pattern: str
) -> List[Dict[str, Any]]

Parse asprin stdout and extract matching atoms with variable bindings.

_parse_atom_args()¶

def _parse_atom_args(self, args_str: str) -> List[str]

Parse atom arguments, handling nested terms correctly.

_get_completeness_checker()¶

def _get_completeness_checker(self) -> CompletenessChecker

Get or create the completeness checker instance (lazy initialization).

_extract_pattern_matches()¶

def _extract_pattern_matches(
    self,
    pattern: str,
    atoms: List[str]
) -> List[Dict[str, Any]]

Extract variable bindings from atoms matching a pattern.

EngineConfig¶

Location: mistaber/engine/config.py

@dataclass
class EngineConfig:
    """Runtime configuration for HsrsEngine."""

    world: str = "base"
    interpretation_precedence: List[str] = field(
        default_factory=lambda: ["taz", "shach"]
    )
    safek_resolution: str = "madrega"
    context: str = "lechatchila"
    policy: Dict[str, str] = field(default_factory=lambda: {
        "safek_issur": "strict",
        "shiur_system": "chazon_ish",
        "minhag_region": "ashkenaz",
    })

Configuration Options¶

Option	Type	Default	Description
`world`	`str`	`"base"`	Default world for queries
`interpretation_precedence`	`List[str]`	`["taz", "shach"]`	Commentator priority (first = lowest)
`safek_resolution`	`str`	`"madrega"`	Doubt resolution strategy
`context`	`str`	`"lechatchila"`	Normative context
`policy`	`Dict[str, str]`	See above	Policy parameters

to_asp_facts()¶

def to_asp_facts(self) -> str

Convert configuration to ASP facts for solver injection.

Output Format:

config_world(base).
config_context(lechatchila).
config_safek_resolution(madrega).
config_policy(safek_issur, strict).
config_policy(shiur_system, chazon_ish).
config_policy(minhag_region, ashkenaz).
config_interp_precedence(taz, 1).
config_interp_precedence(shach, 2).

Query Processing Flow¶

flowchart TB
    query["User Query"]

    ask["ask(atom)"]
    qry["query(pattern)"]
    cmp["compare(pattern)"]
    anl["analyze(scenario)"]

    parse["Parse pattern"]
    solve["Solve with clingo.Control"]
    extract["Extract model symbols"]
    match["Match against predicate"]
    bind["Build variable bindings"]
    result["Return results"]

    query --> ask & qry & cmp & anl
    ask & qry & cmp & anl --> parse
    parse --> solve
    solve --> extract
    extract --> match
    match --> bind
    bind --> result

Error Handling¶

Exception Hierarchy¶

Exception	Module	Description
`RuntimeError`	engine.py	Ontology not loaded
`CyclicWorldError`	kripke/validation.py	Cycle detected in world DAG
`WorldNotFoundError`	kripke/worlds.py	Referenced non-existent world
`ImportError`	engine.py	Clingo not installed

Error Conditions¶

Ontology Not Loaded: Methods raise RuntimeError if _is_loaded is False
Invalid Pattern: Malformed patterns may return empty results
asprin Unavailable: Falls back to regular query
xclingo Unavailable: Falls back to basic verification

Performance Considerations¶

Grounding Size¶

The number of ground atoms grows with: - Number of declared entities (foods, vessels, mixtures) - Number of rules with variables - Depth of world inheritance hierarchy - Number of worlds loaded

Solving Complexity¶

Answer set computation is NP-complete. Factors affecting performance: - Number of choice rules - Depth of negation chains - Size of search space after constraints

Optimization Tips¶

Minimize variables: Use concrete values where possible
Add constraints early: Disjointness rules prune the search space
Use #show directives: Limit output to relevant predicates
Reuse engine instance: _load_ontology() is expensive
Use analyze() for isolated scenarios: Creates fresh control

Memory Usage¶

Main Clingo control holds all grounded rules
analyze() creates temporary control (garbage collected)
query_preferred() uses temp files (cleaned up in finally block)

Thread Safety¶

The engine is not thread-safe. Each thread should use its own engine instance. The Clingo control object maintains internal state that is not designed for concurrent access.

File Organization¶

mistaber/engine/
├── __init__.py           # Exports HsrsEngine
├── engine.py             # Main HsrsEngine class (875 lines)
├── config.py             # EngineConfig dataclass
├── completeness.py       # CompletenessChecker for OWA
├── xclingo_explain.py    # XclingoExplainer for derivations
├── kripke/               # Kripke semantics support
│   ├── __init__.py       # Exports validators and managers
│   ├── validation.py     # WorldDAGValidator, CyclicWorldError
│   └── worlds.py         # WorldManager, WorldNotFoundError
├── machloket/            # Dispute detection (future)
├── safek/                # Safek handling (future)
├── explain/              # Explanation formatting (future)
├── priorities.lp         # Rule activation and override logic
├── safek.lp              # Safek propagation rules
├── interpretations.lp    # Commentator interpretation layer
├── preferences.lp        # asprin preference specification
└── policy.lp             # Policy parameter rules