LIBRARY GUIDE · 0.10.0a21

Reason from facts.
Keep the explanation.

Ting is a deterministic symbolic reasoning library. Write readable .ting modules or Python declarations, run inference, and inspect the evidence behind every conclusion.

STANDARDS & ASSURANCE

NASA standards alignment

Implemented controls: requirement traceability, configuration baselines and recorded verification.

• NPR 7150.2D — NASA Software Engineering Requirements
• NASA-STD-8739.8B — Software Assurance and Software Safety Standard

Scope: selected engineering controls. Full NASA compliance has not been established. No NASA approval or endorsement is implied. Read the assurance scope and verification procedure.

Choose a reading path

1. New to Python or Ting: install Ting, read the Python notation, then run Your first reasoner.
2. Understand a session: continue through facts, rules, running and changing inputs, and work limits.
3. Keep rules in editable files: follow Compile readable source, then the source language. Several source files explains imports.
4. Choose a task: evaluate options, explain a result, reuse selected knowledge, or choose how to run cases.

Complete examples include their imports and setup. Short continuations identify the example or saved file they require. Read a code block’s label before copying: python is application code, ting is rule source, and console is a terminal command. Use the API reference to look up signatures, and Availability and limits to check supported uses.

Install and verify

Install Python 3.11, 3.12 or 3.13 first. A terminal is the Command Prompt or PowerShell on Windows, or Terminal on Linux/macOS. Run the commands below in a folder containing the downloaded wheel. A virtual environment keeps this project’s Python packages separate. The commands use its Python directly, so activation is not required.

Windows — Command Prompt or PowerShell

py -3 -m venv .venv
.\.venv\Scripts\python.exe -m pip install .\ting_reasoner-0.10.0a21-py3-none-any.whl
.\.venv\Scripts\python.exe -m ting_reasoner.selftest

Linux or macOS — Terminal

python3 -m venv .venv
./.venv/bin/python -m pip install ./ting_reasoner-0.10.0a21-py3-none-any.whl
./.venv/bin/python -m ting_reasoner.selftest

The selftest prints a JSON report. Look for "ok": true. If it fails, keep the complete error and use Troubleshooting. Run this check from a terminal because it also starts worker processes. Pip also installs NumPy and Numba. Use NumPy 2.3.5 and Numba 0.63.1 with this package. Their compatible wheels include the numerical tools; a separate C compiler is not needed. For offline installation, first download Ting and all its dependencies into a wheel folder, then install from that complete folder.

Throughout this manual, python example.py means the Python where you installed Ting. With the environment above, use .\.venv\Scripts\python.exe example.py on Windows or ./.venv/bin/python example.py on Linux/macOS, from the same folder. Save Python examples as .py files; commands labelled console belong in the terminal, not in a Python file.

Understand the Python calls

You can use Ting from an ordinary Python script. Save Python code in a file such as my_reasoner.py and run python my_reasoner.py from its folder after installing Ting. Readable rules belong in a separate .ting file. The next sections provide complete examples of both approaches.

Names, objects and function calls

A variable is a name for a value or object. In engine = Engine(program), Python calls Engine, passes program as its argument, and assigns the returned session object to the name engine. You choose that variable name; Ting does not require it.

Indented lines belong to a block such as a loop, try, or with. Copy the indentation along with an example. Lines beginning with # are comments for the reader. An assert in an example checks an expected outcome and raises AssertionError if it is false; your application should use ordinary conditional checks to handle expected failures.

Read a result safely

The following fragment continues Your first reasoner after its input declarations, using that example’s engine. A for loop visits each conclusion and also works when there are none. Indexing engine.conclusions[0] requires at least one result.

result = engine.run()
if result.status.value == "quiescent":
    for conclusion in engine.conclusions:
        print(conclusion.name, conclusion.as_dict())
else:
    print("Reasoning stopped with:", result.status.value)

quiescent means the current rules have no remaining eligible work. A work limit can stop reasoning earlier. An empty result does not by itself mean “rejected” or “impossible”; your rules must explicitly define those decisions. See Work limits for controlled stopping.

Read a signature in the API reference

In compile_file(path, output=None, *, overwrite=False, ...), path is required, output has a default, and arguments after * must be named. An annotation such as str | Path lists accepted types. An arrow such as -> SourceBuild names the returned type. You call the function with values, without copying these annotations or the ... placeholder.

When a call fails, Python raises an exception and skips the remaining statements in its try block. An except SourceError as error block receives Ting’s source error object. An else block runs only when the call succeeded. The build error examples explain what to inspect and how to correct the input.

Your first reasoner

A schema describes data. A rule matches that data and derives facts or emits conclusions. A knowledge base combines schemas and rules; an Engine owns the working session.

from ting_reasoner import Engine, FactSchema, KnowledgeBase, Rule, bind, emit, use

Fruit = FactSchema("fruit", name=str, sweet=bool)
rule = Rule(
    "recognize_sweet_fruit",
    when=(Fruit.pattern(name=bind("name"), sweet=True),),
    then=(emit("sweet_fruit", name=use("name")),),
)
knowledge = KnowledgeBase((Fruit,), (rule,))
engine = Engine(knowledge)
engine.declare(Fruit(name="apple", sweet=True), Fruit(name="lemon", sweet=False))
result = engine.run()
assert result.firings == 1
assert engine.conclusions[0]["name"] == "apple"
print(engine.conclusions[0].as_dict())

bind("name") captures a fact value. use("name") refers to that earlier binding. The output is {"name": "apple"}. A conclusion is immutable; as_dict() gives you an ordinary mapping for integration.

What each call contributes

Facts and types

A fact is one record created from a schema, such as Fruit(name="apple", sweet=True). Its values cannot be edited in place. Pass several facts to one declare() call to add them together; declaring an identical input fact reuses its handle. Wrong field names or types are rejected before inputs change. A derived fact is produced by a rule and may match other rules. An emitted conclusion is an application result in engine.conclusions; it is not an input to further rule matching.

engine.facts returns a tuple of current facts. engine.fact_items returns (handle, fact) pairs. A handle identifies a fact within this session; do not transfer it to another Engine. Read Run and inspect before removing, replacing or resetting inputs.

Entities and strong references

Use an entity when several facts need to identify the same object while its other properties change. An EntitySchema declares an ordered key. EntityRef contains the schema’s qualified name and its key values. A ReferenceField requires an existing entity of its declared target schema when facts are published.

Declare, resolve and update a linked object

This complete example creates an asset and its sensor, rejects a broken link, then changes the asset identity and redirects the sensor atomically. The retained snapshot continues to resolve the original asset.

from ting_reasoner import (
    Engine, EntitySchema, FactSchema, Field, KnowledgeBase,
    MemoryPolicy, ReferenceField, SchemaError,
)

Asset = EntitySchema("equipment.Asset", {"serial": str, "location": str},
                     key=("serial",))
Sensor = FactSchema("equipment.Sensor", (
    Field("name", str), ReferenceField("asset", "equipment.Asset"),
))
engine = Engine(KnowledgeBase([Asset, Sensor]),
                memory=MemoryPolicy(max_entity_index_entries=8))

# The sensor may come first: the complete declaration is validated together.
sensor, asset = engine.declare(
    Sensor(name="temperature", asset=Asset.ref("pump-a")),
    Asset(serial="pump-a", location="north"),
)
saved = engine.snapshot()
assert engine.resolve(Asset.ref("pump-a"))["location"] == "north"
assert engine.referrers(Asset.ref("pump-a"))[0].handle == sensor

try:
    engine.retract(asset)
except SchemaError:
    pass  # The sensor still requires this target; neither fact changed.
else:
    raise AssertionError("a referenced entity must not be deleted")

# Rename explicitly, redirecting the incoming edge in the same transaction.
with engine.transaction() as transaction:
    transaction.retract(asset)
    transaction.declare(Asset(serial="pump-b", location="south"))
    transaction.replace(sensor, Sensor(name="temperature", asset=Asset.ref("pump-b")))

print(engine.resolve(Asset.ref("pump-b"))["location"])
print(saved.resolve(Asset.ref("pump-a"))["location"])
# Output: south, then north, on separate lines.
report = engine.entity_index_report
assert (report.entities, report.references) == (1, 1)
assert report.peak_reserved_entries == 4

Choose keys and reference fields

Keys contain one or more required fields, each with one exact int, str or bytes type. Key order matters. Booleans, floats, nulls, unions, arrays and other references cannot be key fields. For a composite key, write EntitySchema("Part", {"serial": str, "revision": int}, key=("serial", "revision")) and construct its reference with Part.ref("pump-a", 2). FactSchema retains ordinary scalar fields, including fields named key or entity_key.

Constructing a reference checks its portable value shape; EntitySchema.ref() also checks key arity and types. Publication checks the reference against the compiled schema and complete resulting fact set. Different live facts cannot share an entity key. Declaring the identical fact again reuses its existing handle. References contain no handles or physical row addresses, so compaction and process transfer preserve their identity.

Transactions, cycles and snapshots

A multi-fact declare(), transaction commit, reset or single rule firing validates its complete resulting graph before publication. Forward references and cycles are allowed when every target exists. Deleting a target requires removing or redirecting every surviving incoming reference in the same atomic operation. Replacing an entity within its schema preserves its key; rename with explicit retraction, declaration and redirection. A failed operation preserves committed facts, outputs and explanations. Earlier successful rule firings remain committed if a later firing fails.

engine.resolve(ref) returns the current immutable target fact. A well-typed key without a target raises KeyError; an unknown target schema or wrong key shape raises SchemaError. engine.referrers(ref) returns EntityLink records in source-fact publication order, then field order. Each link provides its source fact, field and explanation handle. saved.resolve(ref) and saved.referrers(ref) query only that saved snapshot. Reintroducing a key denotes the same logical identity in the new snapshot; no hidden generation is added.

Author, test and compile reference-aware rules

Save the following source as entity_equipment.ting. The explicit requires feature entity-references declaration enables key(...), reference Target required and ref Target [keys]. Names in source are local or explicitly imported; compiled reference targets use their qualified schema names. For a bytes key, use a lowercase hexadecimal object such as ref Part [{"bytes":"00ff"},2], in the declared key order.

The two fixed assets form a valid cycle. The rule joins a request to all matching alarms by exact reference identity. Both tests explicitly include the fixed knowledge using given seed.

ting source 1
requires semantics 1
module entity_equipment version "1.0.0" role program
requires feature entity-references
requires feature seed-facts

schemas:
  schema Asset key(serial):
    serial: text required
    backup: reference Asset required
    location: text required
  schema Request:
    asset: reference Asset required
  schema Alarm:
    asset: reference Asset required
    code: text required

conclusions:
  conclusion SelectedAlarm:
    asset: reference Asset required
    code: text required

knowledge:
  seed pump_a: Asset:
    serial: "pump-a"
    backup: ref Asset ["pump-b"]
    location: "north"
  seed pump_b: Asset:
    serial: "pump-b"
    backup: ref Asset ["pump-a"]
    location: "south"

rules:
  rule select_alarm:
    when:
      match Request:
        asset: bind selected
      match Alarm:
        asset: eq use selected
        code: bind code
    then:
      conclude SelectedAlarm:
        asset: use selected
        code: use code

tests:
  test selected_reference:
    given seed
    given:
      fact Request:
        asset: ref Asset ["pump-a"]
      fact Alarm:
        asset: ref Asset ["pump-a"]
        code: "temperature"
      fact Alarm:
        asset: ref Asset ["pump-b"]
        code: "pressure"
    expect:
      conclusion SelectedAlarm:
        asset: ref Asset ["pump-a"]
        code: "temperature"
    firings: 1
  test all_alternative_matches:
    given seed
    given:
      fact Request:
        asset: ref Asset ["pump-a"]
      fact Alarm:
        asset: ref Asset ["pump-a"]
        code: "temperature"
      fact Alarm:
        asset: ref Asset ["pump-a"]
        code: "pressure"
    expect:
      conclusion SelectedAlarm:
        asset: ref Asset ["pump-a"]
        code: "temperature"
      conclusion SelectedAlarm:
        asset: ref Asset ["pump-a"]
        code: "pressure"
    firings: 2

In the IDE, paste the complete example into Full source, save, then use Check, Tests and Compile. Data, Knowledge, Rules and Tests retain their usual categories. Editing entity declarations in a section enables the feature declaration automatically. The editor highlights key, reference and ref; Tests feedback reports entity and edge counts. From a terminal, run ting-build compile entity_equipment.ting -o entity_equipment.tingir. Use this script beside the compiled file; execution does not require the source.

from ting_reasoner import Engine, EntityRef, load_tingir

engine = Engine(load_tingir("entity_equipment.tingir"))
Request = engine.schema("entity_equipment.Request")
Alarm = engine.schema("entity_equipment.Alarm")
selected = EntityRef("entity_equipment.Asset", ("pump-a",))
engine.declare(Request(asset=selected), Alarm(asset=selected, code="temperature"))
engine.run()
result = engine.conclusions[0]
print(result["code"], engine.resolve(result["asset"])["location"])
# Output: temperature north
assert len(engine.explain(result)[0].fact_ids) == 2
assert engine.resolve(selected)["backup"] == EntityRef("entity_equipment.Asset", ("pump-b",))

Understand matching and evidence

Reference fields support binding, equality and inequality; the Python constraint API also supports membership with one_of(). Source reference literals use bind, eq or ne constraints. References have no less-than ordering. Binding references or using literal references in derived facts preserves their type; all targets must exist after that firing. Rule explanations record the facts explicitly matched by the rule. Reading a target with Python resolve() does not add a rule premise or an implicit target-property dependency. To reason from target properties, use an explicit follow condition or match the target and its key in the rule. Reference literals accept complete constant keys. For rule bindings, use the explicit action construction described below. Implicit dereference expressions are unsupported.

Follow a reference in a rule

Use EntitySchema.follow(use("binding"), ...) when a rule needs properties of an entity identified by an earlier fact. The reference binding must already be established by an earlier positive match and must target that exact schema. follow matches the complete ordered key, then applies the supplied non-key filters and bindings. Key fields are fixed by the reference; attempts to override or bind them in the same call are rejected. A Python EntityRef constant may also be passed directly.

This complete Python example explains an asset’s location using both the request and the asset itself. Changing the asset invalidates its previous conclusion; a retained snapshot continues to contain the earlier location.

from ting_reasoner import (
    Engine, EntitySchema, FactSchema, KnowledgeBase, ReferenceField,
    Rule, bind, emit, use,
)

Asset = EntitySchema("Asset", {"serial": str, "location": str}, key=("serial",))
Request = FactSchema("Request", [ReferenceField("asset", "Asset")])
locate = Rule("locate", (
    Request.pattern(asset=bind("selected")),
    Asset.follow(use("selected"), location=bind("place")),
), (emit("location", place=use("place")),))
engine = Engine(KnowledgeBase((Asset, Request), (locate,)))
asset, request = engine.declare(
    Asset(serial="pump-a", location="north"),
    Request(asset=Asset.ref("pump-a")),
)
engine.run()
result = engine.conclusions[0]
print(result["place"])
assert engine.explain(result)[0].fact_ids == (request.id, asset.id)
saved = engine.snapshot()

# The target is a real premise: changing it withdraws the old answer.
engine.replace(asset, Asset(serial="pump-a", location="south"))
engine.run()
assert len(engine.conclusions) == 1
print(engine.conclusions[0]["place"])
print(saved.resolve(Asset.ref("pump-a"))["location"])
# Output: north, south, north, on separate lines.

Save the next example as entity_traversal.ting. Declare both entity features. follow Asset via use selected: identifies the requested asset, and a second follow identifies its backup. The graph contains a cycle, but this rule performs exactly the two written matches. There is no implicit recursive walk. The two embedded tests exercise both directions.

ting source 1
requires semantics 1
module entity_traversal version "1.0.0" role program
requires feature entity-references
requires feature entity-traversal
requires feature seed-facts

schemas:
  schema Asset key(serial):
    serial: text required
    backup: reference Asset required
    location: text required
  schema Request:
    asset: reference Asset required

conclusions:
  conclusion BackupLocation:
    requested: reference Asset required
    location: text required

knowledge:
  seed pump_a: Asset:
    serial: "pump-a"
    backup: ref Asset ["pump-b"]
    location: "north"
  seed pump_b: Asset:
    serial: "pump-b"
    backup: ref Asset ["pump-a"]
    location: "south"

rules:
  rule locate_backup:
    when:
      match Request:
        asset: bind selected
      follow Asset via use selected:
        backup: bind backup
      follow Asset via use backup:
        location: bind location
    then:
      conclude BackupLocation:
        requested: use selected
        location: use location

tests:
  test first_backup:
    given seed
    given:
      fact Request:
        asset: ref Asset ["pump-a"]
    expect:
      conclusion BackupLocation:
        requested: ref Asset ["pump-a"]
        location: "south"
    firings: 1
  test second_backup:
    given seed
    given:
      fact Request:
        asset: ref Asset ["pump-b"]
    expect:
      conclusion BackupLocation:
        requested: ref Asset ["pump-b"]
        location: "north"
    firings: 1

In the IDE, use Full source for the complete example, then Check, Tests and Compile. Editing a follow statement in the Rules section adds both feature declarations automatically. The editor highlights follow and via. From a terminal run ting-build compile entity_traversal.ting -o entity_traversal.tingir. Place this script beside that compiled artifact; the readable source is no longer required.

from ting_reasoner import Engine, EntityRef, load_tingir

engine = Engine(load_tingir("entity_traversal.tingir"))
Request = engine.schema("entity_traversal.Request")
engine.declare(Request(asset=EntityRef("entity_traversal.Asset", ("pump-a",))))
engine.run()
result = engine.conclusions[0]
assert len(engine.explain(result)[0].fact_ids) == 3
print(result["location"])
# Output: south. The request, requested asset and backup asset are premises.

A positive follow contributes its target fact to the rule’s ordinary support and source explanation. Changing or retracting a supporting input uses the existing invalidation and maintenance rules. The ordinary equality indexes narrow candidate targets, including compound keys; all key components are checked with exact types. Candidate rows, constraints, activations and firings use the existing ExecutionBudget counters. An exhausted budget publishes no partially fired conclusion; earlier committed firings remain. Index construction, graph publication and total process memory are governed by their separate contracts, not by this matching budget.

Binding-based follow instructions use program format 10 / semantics 11. Older profiles retain their original meanings and bytes. The verifier checks the reference’s target type, prior initialization, complete key coverage and key order before execution. Deploy the updated compiler and runtime together for these programs. Indexed library format 4 can carry the new program profile and preserve selective linking and source evidence. Use action constructors to build references from scalar bindings; see reference construction.

Construct references from rule bindings

Use Asset.ref(use("serial")) in a Python rule action to construct an entity identity from an established scalar binding. Supply every key component in schema key order; constants and bindings may be mixed. The compiler requires each binding to have the exact key type in every branch. No conversion, nested constructor, array expression, optional key or partial key is accepted.

This complete example registers an asset and its referring assignment in one firing. The assignment may be written first because the complete candidate graph is checked before publication. The second rule follows it, so the target fact appears in the explanation. Updating the request refreshes the derived facts while the saved snapshot keeps its earlier values.

from ting_reasoner import (
    Engine, EntitySchema, FactSchema, KnowledgeBase, ReferenceField,
    Rule, bind, derive, emit, use,
)

Asset = EntitySchema("Asset", {"serial": str, "location": str}, key=("serial",))
NewAsset = FactSchema("NewAsset", serial=str, location=str)
Assignment = FactSchema("Assignment", (ReferenceField("asset", "Asset"),))
register = Rule("register", (
    NewAsset.pattern(serial=bind("serial"), location=bind("place")),
), (
    # Both derived facts are checked together before either is published.
    derive(Assignment, asset=Asset.ref(use("serial"))),
    derive(Asset, serial=use("serial"), location=use("place")),
))
locate = Rule("locate", (
    Assignment.pattern(asset=bind("selected")),
    Asset.follow(use("selected"), location=bind("place")),
), (emit("location", asset=use("selected"), place=use("place")),))
engine = Engine(KnowledgeBase((Asset, NewAsset, Assignment), (register, locate)))
handle = engine.declare(NewAsset(serial="pump-a", location="north"))
engine.run()
result = engine.conclusions[0]
print(result["place"])  # north
assert result["asset"] == Asset.ref("pump-a")
assert len(engine.explain(result)[0].fact_ids) == 2
saved = engine.snapshot()
engine.replace(handle, NewAsset(serial="pump-a", location="south"))
engine.run()
print(engine.conclusions[0]["place"])  # south
print(saved.resolve(Asset.ref("pump-a"))["location"])  # north

In readable source, add requires feature entity-construction alongside entity-references. Inside an action’s ordered ref list, {"use": "serial"} reads the named binding; ordinary JSON strings/integers and {"bytes": "00ff"} are constant keys. These binding objects are valid only in then fields, not in seeds, conditions or expected test values. Add entity-traversal when also using follow.

Save this complete example as entity_construction.ting. Its embedded tests each create an asset, follow the resulting assignment and verify the expected location.

ting source 1
requires semantics 1
module entity_construction version "1.0.0" role program
requires feature entity-references
requires feature entity-construction
requires feature entity-traversal

schemas:
  schema NewAsset:
    serial: text required
    location: text required
  schema Asset key(serial):
    serial: text required
    location: text required
  schema Assignment:
    asset: reference Asset required

conclusions:
  conclusion Location:
    asset: reference Asset required
    location: text required

rules:
  rule register:
    when:
      match NewAsset:
        serial: bind serial
        location: bind place
    then:
      derive Assignment:
        asset: ref Asset [{"use": "serial"}]
      derive Asset:
        serial: use serial
        location: use place
  rule locate:
    when:
      match Assignment:
        asset: bind selected
      follow Asset via use selected:
        location: bind place
    then:
      conclude Location:
        asset: use selected
        location: use place

tests:
  test north:
    given:
      fact NewAsset:
        serial: "pump-a"
        location: "north"
    expect:
      conclusion Location:
        asset: ref Asset ["pump-a"]
        location: "north"
    firings: 2
  test south:
    given:
      fact NewAsset:
        serial: "pump-b"
        location: "south"
    expect:
      conclusion Location:
        asset: ref Asset ["pump-b"]
        location: "south"
    firings: 2

In the IDE, edit the Rules section to add construction; the required feature declarations are inserted automatically. Check and Tests report source diagnostics before Compile publishes an artifact. A failed compile preserves the previous output. Compile with ting-build compile entity_construction.ting -o entity_construction.tingir, then run this script beside the artifact. The original source file is no longer required.

from ting_reasoner import Engine, EntityRef, load_tingir

engine = Engine(load_tingir("entity_construction.tingir"))
NewAsset = engine.schema("entity_construction.NewAsset")
engine.declare(NewAsset(serial="pump-a", location="north"))
engine.run()
result = engine.conclusions[0]
print(result["location"])  # north
assert result["asset"] == EntityRef("entity_construction.Asset", ("pump-a",))
assert len(engine.explain(result)[0].fact_ids) == 2

Identity and evidence: construction does not read a target, add a premise or guarantee existence. An emitted reference is an identity value and may not resolve; resolve() then raises KeyError. A reference stored in a fact’s ReferenceField must resolve in the complete resulting graph. A missing target or conflicting entity rejects that firing with SchemaError and preserves the previous committed state. Earlier successful firings remain committed. Use an explicit target match or follow whenever target properties must support a conclusion.

Dynamic construction uses program format 11 / semantics 12. The verifier bounds the flat key list by the declared schema and charges each key against instruction fuel. Ordinary firing budgets and graph-entry admission still apply; they do not guarantee elapsed-time deadlines or total process memory. Literal-only references retain their earlier profiles. Use matching compiler/runtime versions; source-free indexed libraries and worker transfer preserve exact keys and evidence.

Bound indexes and inspect their cost

MemoryPolicy(max_entity_index_entries=N) bounds the current Engine’s entity and edge entries plus a staged replacement index. One entity and one edge each count as one entry. For a replacement with two current and two candidate entries, allow four. The count is checked before target and incoming-edge maps are allocated. A resource error publishes no partial graph. The limit is independent of numerical byte and store-buffer limits; it does not estimate Python bytes, key sizes, snapshot indexes or process RAM. Use the IDE’s General → Entity index entries control; Unlimited leaves this entry bound unset. Each worker applies its own job policy.

entity_index_report exposes current entities, references, peak_reserved_entries and limit_entries. The peak counts accepted old/new overlap for the Engine’s lifetime; reset preserves that peak. Current and saved-snapshot lookup uses a hash index with cost proportional to key width on average. Graph validation counts and scans candidate rows and edges without recursively expanding targets; a graph mutation rebuilds its indexes. Programs with no entity or reference fields bypass this graph path. Engine-created snapshots share the validated immutable index for their captured generation. An update publishes a new index; older snapshots retain the earlier one until released.

Snapshot ownership and index storage

Use engine.snapshot() to retain one consistent set of facts, conclusions and explanations. The snapshot shares its immutable entity index with the Engine and other snapshots of that generation. It keeps no Engine, packed store, lock or registry alive. Updating the Engine publishes a new graph without changing earlier lookups. Fact records, ordered fact tuples and proof collections still have their own snapshot costs; snapshot creation is not constant-time or a zero-allocation operation.

Read engine.entity_storage_report when you need to inspect current and retained index storage. Inspection scans each previously unmeasured live generation once and caches its shallow total. Subsequent inspections enumerate live generations and reuse those totals. The weak registry does not keep obsolete indexes alive; the returned EntityStorageReport contains counts and scope text, not graph references.

import gc
from ting_reasoner import Engine, EntitySchema, FactSchema, KnowledgeBase, ReferenceField

Asset = EntitySchema("Asset", {"id": str, "location": str}, key=("id",))
Sensor = FactSchema("Sensor", (ReferenceField("asset", "Asset"),))
engine = Engine(KnowledgeBase((Asset, Sensor)))
asset, sensor = engine.declare(
    Asset(id="pump-a", location="north"), Sensor(asset=Asset.ref("pump-a")),
)
first = engine.snapshot()
second = engine.snapshot()
report = engine.entity_storage_report
assert report.live_index_generations == 1  # Both snapshots share this index.
assert report.snapshot_index_reuses == 2
assert report.current_entries == 2  # One entity and one reference.

engine.replace(asset, Asset(id="pump-a", location="south"))
report = engine.entity_storage_report
assert report.live_index_generations == 2
assert report.retained_index_generations == 1  # The old index is counted once.
assert report.retained_entries == 2
assert report.live_index_shallow_bytes == (
    report.current_index_shallow_bytes + report.retained_index_shallow_bytes
)
print(first.resolve(Asset.ref("pump-a"))["location"])  # north
print(engine.resolve(Asset.ref("pump-a"))["location"])  # south

del first, second
gc.collect()  # Explicit collection makes this demonstration portable.
report = engine.entity_storage_report
assert report.retained_index_generations == 0
assert report.retained_index_shallow_bytes == 0
assert report.current_index_shallow_bytes > 0
print(report.as_dict())  # Byte counts depend on the Python implementation/build.

Measurement scope: shallow sizes include index objects and maps, generated identities/key tuples, fact wrappers/value tuples, incoming-link tuples, and link/handle wrappers. A fact used by several links is counted once per index. Borrowed schemas, key payloads and referenced values are not recursively sized. Snapshot records and their fact tuples, unpublished staging/candidate tracebacks, registry/inspection metadata, other indexes, proofs, allocator overhead and process RAM are excluded. These Python-build-dependent totals are diagnostic estimates, not an admission limit, an exclusive heap allocation total or a prediction of RSS released by deleting snapshots.

The existing max_entity_index_entries limit still covers current plus candidate entries only; retained old generations do not consume that limit. Release unneeded snapshots deliberately. Pickle/deepcopy and the public EngineSnapshot(...) constructor produce detached indexes outside the originating Engine’s report. dataclasses.replace(snapshot, ...) validates the supplied records and rebuilds an independent index. Programs without entity/reference fields report zero graph storage. No source feature or portable-format change is required to use this Python inspection API.

Portable files and explicit errors

Entity programs use Compact-IR format 9 with semantics 10; binding-based follow uses format 10 with semantics 11; dynamic action construction uses format 11 with semantics 12. Existing programs retain their established profiles. Verification checks keys, target schemas, reference constants and complete seed graphs before execution. Use program artifacts, explicit source imports or compiled entity libraries. RuleSpec version 1 serialization of references is rejected explicitly. A SchemaError identifies duplicate keys, an unresolved strong reference or a key-changing replacement; use one transaction to make the intended graph consistent. A ResourceLimitError reports the entry count required by the chosen operation.

Rules and conditions

Rules have an explicit name, conditions and consequences. Bindings are established left to right. Reusing a bound name compares exact values. Use Rule(..., salience=10) to request a higher priority within the same dependency level; the default is 0. In readable source the corresponding entry is priority: 10. Dependencies needed for negation take precedence over priority.

from ting_reasoner import Engine, FactSchema, KnowledgeBase, Rule, bind, emit, eq, use

Request = FactSchema("request", orchard=int, variety=str)
Stock = FactSchema("stock", orchard=int, variety=str, boxes=int)
rule = Rule("find_stock", (
    Request.pattern(orchard=bind("orchard"), variety=bind("variety")),
    Stock.pattern(orchard=eq(use("orchard")), variety=eq(use("variety")), boxes=bind("boxes")),
), (emit("available", variety=use("variety"), boxes=use("boxes")),))
engine = Engine(KnowledgeBase((Request, Stock), (rule,)))
engine.declare(Request(orchard=3, variety="apple"), Stock(orchard=3, variety="apple", boxes=12))
engine.run()
assert engine.conclusions[0]["boxes"] == 12

This example joins Request and Stock by requiring both orchard and variety to agree. The expected number of boxes is 12. Rules define comparisons and outputs; your Python application acts on the returned conclusions. A rule cannot use the absence of a fact to derive the very fact whose absence it tests, directly or through a dependency cycle. Such recursion through negation fails compilation. See Write Ting rules for the equivalent source notation.

Run and inspect

engine.run() applies eligible rules until no work remains or a limit stops it. A rule application to a particular combination of facts is called an activation; executing its actions is a firing. Running again without changing inputs continues remaining work and does not repeat already completed firings.

What happens when you change inputs?

By default, an input change immediately clears derived facts, conclusions, proofs and trace. Call run() afterward to rebuild results. An unchanged replacement, duplicate input declaration or missing-handle retraction makes no change. Optional support maintenance can retain still-supported results; it also requires run() to find new consequences.

Read result.status.value: "quiescent" means no eligible work remains; "firing_limit" or "budget_exhausted" means the run stopped at a limit. Results can be incomplete in those two cases. Counts such as result.firings describe this call. See Work limits for resuming a stopped run. Each firing either commits all its actions or none; an exception later in the run does not undo previous successful firings.

Work limits

Use logical work budgets when predictable exhaustion boundaries matter. Candidate rows, constraint evaluations, activations and firings are charged before the corresponding work. A call accepts either max_firings or budget.

from ting_reasoner import Engine, ExecutionBudget, FactSchema, KnowledgeBase, Rule, bind, emit, use

Item = FactSchema("item", value=int)
rule = Rule("record", (Item.pattern(value=bind("v")),), (emit("value", value=use("v")),))
engine = Engine(KnowledgeBase((Item,), (rule,)))
engine.declare(*(Item(value=i) for i in range(5)))
result = engine.run(budget=ExecutionBudget(max_candidate_rows=2, max_firings=10))
assert result.status.value == "budget_exhausted"
assert result.exhausted_budget == "candidate_rows"
assert result.budget_usage.candidate_rows == 2

Inspect exhausted_budget and budget_usage before deciding how to resume. Repeating an insufficient matching budget can repeat the same scan; matching exhaustion is not a saved cursor through arbitrary partial work. Firing exhaustion is resumable. A budget bounds counted work, not elapsed time or total memory. Supplying one can affect speed; compare the same budget settings when measuring performance.

Transactions

A transaction stages base-fact declarations, retractions and replacements. It commits all edits together. The expected epoch is the session’s change counter, used to detect an intervening update. A stale transaction raises StaleTransactionError; an exception inside the context rolls it back.

from ting_reasoner import Engine, FactSchema, KnowledgeBase

Reading = FactSchema("reading", sensor=str, value=int)
engine = Engine(KnowledgeBase((Reading,)))
with engine.transaction(expected_epoch=engine.epoch) as tx:
    staged = tx.declare(Reading(sensor="north", value=21))
handle = tx.receipt.resolve(staged)
assert handle is not None
assert engine.facts == (Reading(sensor="north", value=21),)
snapshot = engine.snapshot()
assert snapshot.epoch == engine.epoch

A staged handle belongs to its creating transaction. Resolve it through the commit receipt after success. Run inference after the transaction closes. Snapshots are immutable views, not mutable clones of an Engine.

Bound transaction publication buffers

Use engine.transaction(max_buffer_bytes=...) to check the store-buffer payload required by a changed transaction. The limit is a non-negative integer in bytes; None applies no payload limit. The check happens when the transaction commits. An unchanged commit needs no replacement buffers and returns no buffer report, even when the limit is zero.

The bound includes the current store, the complete replacement base buffers, and a conservative allowance for the old derived-buffer payload when support maintenance is enabled. Maintenance may retain fewer rows or select recomputation, so the reserved bound can exceed the payload actually copied. Buffer lengths include tombstones in the old store. The committed replacement base store is dense.

This limit covers initialized numeric/byte columns and row-buffer payload lengths. It excludes container headers and spare capacity, values held in object columns (including array payloads), the transaction's staged projection, indexes, proofs, saved snapshots, JIT and process RAM. MemoryPolicy continues to govern numeric operations separately. The transaction limit does not control memory used while staging edits.

If the payload bound exceeds the limit, ResourceLimitError exposes resource, requested and limit. The transaction closes without a receipt and leaves the engine's committed facts, results, handles and epoch unchanged. Retry in a new transaction with a suitable limit. Allocation failure, a stale epoch or another commit error also requires a new transaction.

from ting_reasoner import (
    Engine, FactSchema, KnowledgeBase, ResourceLimitError, Rule,
    between, bind, emit, use,
)

Reading = FactSchema("Reading", number=int, value=float)
rule = Rule(
    "report", Reading.pattern(number=between(0, 2), value=bind("value")),
    [emit("Reported", value=use("value"))],
)
engine = Engine(KnowledgeBase([Reading], [rule]))
handles = engine.declare(*(Reading(number=i, value=float(i)) for i in range(300)))
engine.run()
saved = engine.snapshot()

try:
    with engine.transaction(max_buffer_bytes=0) as rejected:
        rejected.replace(handles[0], Reading(number=0, value=99.0))
except ResourceLimitError as error:
    assert error.resource == "transaction store buffer bytes"
    assert rejected.closed and rejected.receipt is None
else:
    raise AssertionError("A changed transaction needs publication buffers")
assert engine.snapshot() == saved

# This replacement keeps the same schemas and row count. Twice the current
# payload covers both stores; enabled support maintenance also fits this bound.
# Adding rows can require a larger bound.
maximum = 2 * engine.store_memory_report.current_buffer_payload_bytes
with engine.transaction(expected_epoch=engine.epoch, max_buffer_bytes=maximum) as update:
    staged = update.replace(handles[0], Reading(number=0, value=99.0))
receipt = update.receipt
assert receipt.resolve(staged) == handles[0]
report = receipt.buffer_report
assert report is not None
assert report.peak_buffer_payload_bound_bytes <= maximum
assert report.buffer_limit_bytes == maximum
engine.run()
assert engine.conclusions[0]["value"] == 99.0
assert saved.fact_items[0][1]["value"] == 0.0
print(report.peak_buffer_payload_bound_bytes)  # 17400
print(report.as_dict())

Read receipt.buffer_report after a changed commit. It is an immutable TransactionBufferReport; as_dict() returns a detached dictionary. The report belongs to that receipt and remains available after subsequent updates. It describes publication buffers, not the next inference run.

Compile readable source

Before this section: Understand the Python calls and Your first reasoner.

You choose where your files live. Save editable knowledge as a UTF-8 .ting file in any folder you can write to. Pass its path to compile_file(), which validates the source, runs its embedded tests, and writes a portable .tingir program. You can choose a different folder for that compiled file.

1. Choose the source folder and filename

Run this Python code to create an example folder under your home directory and print the full source path. orchard.ting is an example filename; you can use your own name. Creating the folder does not yet create the source file.

from pathlib import Path

project_dir = Path.home() / "ting_examples"
project_dir.mkdir(parents=True, exist_ok=True)
source_path = project_dir / "orchard.ting"
print("Save the Ting source here:", source_path)

Path.home() supplies your actual home directory on Windows, Linux or macOS. To choose another folder, replace the project_dir assignment above with a path for your system, then create the folder in the same way:

These are illustrative paths: replace the drive, username and folder with your own. The / operator joins folders and filenames for the operating system running Python; spaces and Unicode names are supported. The r before a Windows string preserves its backslashes. Windows drive paths do not become Linux paths. If a user supplies a path beginning with ~, call Path(value).expanduser() to expand the home directory.

2. Save the readable source at that path

In your text editor, use Save As and save the following text to the full path printed as source_path. Choose UTF-8 encoding and the exact filename orchard.ting, without an added .txt extension. The schema describes input facts, the conclusion describes an output, and the test checks that an apple is reported while a lemon is not.

# Edit this file, then call compile_file("orchard.ting").
ting source 1
requires semantics 1
module orchard version "1.0.0" role program

schemas:
  schema Fruit:
    name: text required
    sweet: boolean required

conclusions:
  conclusion SweetFruit:
    name: text required

rules:
  rule recognize_sweet_fruit:
    when:
      match Fruit:
        name: bind name
        sweet: true
    then:
      conclude SweetFruit:
        name: use name

tests:
  test apple_is_sweet:
    given:
      fact Fruit:
        name: "apple"
        sweet: true
      fact Fruit:
        name: "lemon"
        sweet: false
    expect:
      conclusion SweetFruit:
        name: "apple"
    firings: 1

If your application already holds that text in a Python string called source_text, save it through the same path:

with source_path.open("x", encoding="utf-8") as stream:
    stream.write(source_text)

Use the editor or the Python save operation to create the source file. Mode "x" raises FileExistsError if the file already exists; use "w" only when you intend to replace its contents. Folder creation and source writing use ordinary Python filesystem exceptions such as PermissionError. Ting's compile_file() reads an existing source file; it does not create that editable source for you.

3. Compile the saved file, then load and run it

This code uses source_path from step 1. Keep it in the same Python session or script. If you run it separately, repeat the folder and source-path assignments first.

from ting_reasoner import Engine, compile_file, load_tingir

built = compile_file(source_path)
print("Compiled program:", built.output)
# Default: orchard.tingir in the same folder as orchard.ting.
engine = Engine(load_tingir(built.output))
Fruit = engine.schema("orchard.Fruit")
engine.declare(Fruit(name="pear", sweet=True))
engine.run()
print(engine.conclusions[0].as_dict())  # {"name": "pear"}

built contains the verified program, test report, source analysis, output path and digest. Use Engine(built.program) to run immediately without reloading the file. built.schema("Fruit") accepts a name visible in the root module; after loading an artifact alone, engine.schema("orchard.Fruit") uses its exact qualified name.

4. Choose a different compiled-output folder

Pass a second argument to select the full compiled filename. Create its parent directory first. This example keeps compiled programs in a compiled subfolder; output_dir can instead be any other writable absolute folder path.

output_dir = project_dir / "compiled"
output_dir.mkdir(parents=True, exist_ok=True)
output_path = output_dir / "orchard.tingir"

built = compile_file(source_path, output_path)
print(built.output)  # The full path to compiled/orchard.tingir.
engine = Engine(load_tingir(built.output))

The compiler requires existing parent directories and protects existing outputs. To deliberately rebuild the same compiled file, call compile_file(source_path, output_path, overwrite=True). Compilation and embedded tests must succeed before the previous compiled file is replaced. Catch SourceError for compilation or publication failures; see Handle build errors.

What does a filename without a folder mean?

compile_file("orchard.ting") looks in the Python process's current working directory. Print Path.cwd() to see that directory. Relative paths for explicit outputs and imported module files also start there. Use the explicit source_path and output_path variables above when you want the same locations regardless of where your application is launched.

from pathlib import Path
from tempfile import TemporaryDirectory
from ting_reasoner import Engine, SourceError, compile_file, load_tingir

source_text = """# Edit this file, then call compile_file("orchard.ting").
ting source 1
requires semantics 1
module orchard version "1.0.0" role program

schemas:
  schema Fruit:
    name: text required
    sweet: boolean required

conclusions:
  conclusion SweetFruit:
    name: text required

rules:
  rule recognize_sweet_fruit:
    when:
      match Fruit:
        name: bind name
        sweet: true
    then:
      conclude SweetFruit:
        name: use name

tests:
  test apple_is_sweet:
    given:
      fact Fruit:
        name: "apple"
        sweet: true
      fact Fruit:
        name: "lemon"
        sweet: false
    expect:
      conclusion SweetFruit:
        name: "apple"
    firings: 1
"""

def save_and_run(project_dir):
    project_dir = Path(project_dir).expanduser().absolute()
    project_dir.mkdir(parents=True, exist_ok=True)
    source_path = project_dir / "orchard.ting"
    # "x" creates a new source file and protects an existing one.
    with source_path.open("x", encoding="utf-8") as stream:
        stream.write(source_text)

    beside_source = compile_file(source_path)
    assert beside_source.output == source_path.with_suffix(".tingir")

    output_dir = project_dir / "compiled"
    output_dir.mkdir(parents=True, exist_ok=True)
    output_path = output_dir / "orchard.tingir"
    built = compile_file(source_path, output_path)
    assert built.output == output_path and output_path.is_file()
    assert source_path.read_text(encoding="utf-8") == source_text

    # An existing compiled file also requires an explicit overwrite choice.
    try:
        compile_file(source_path, output_path)
    except SourceError as error:
        assert error.diagnostic.code == "TING-IO-003"
    else:
        raise AssertionError("Expected existing output to be protected")
    built = compile_file(source_path, output_path, overwrite=True)

    engine = Engine(load_tingir(built.output))
    Fruit = engine.schema("orchard.Fruit")
    engine.declare(Fruit(name="pear", sweet=True))
    engine.run()
    assert engine.conclusions[0]["name"] == "pear"
    print("Source:", source_path)
    print("Compiled:", built.output)
    print(engine.conclusions[0].as_dict())

# A disposable folder keeps this complete demonstration repeatable.
with TemporaryDirectory() as directory:
    save_and_run(Path(directory) / "My Ting files café")
# For files you want to keep, replace the two lines above with:
# save_and_run(Path.home() / "ting_examples")

Compile when rules or schemas change. Reuse the program for new facts and independent sessions. Facts under tests: are used only to check the rules; they are not the inputs of your application. An Engine begins with any fixed seeds in its program. Supply observations for the current case with declare().

Source held in memory

compile_source(text) performs the same compilation and tests without writing files. This complete example can run in a script or notebook.

from ting_reasoner import Engine, compile_source

source = """# Edit this file, then call compile_file("orchard.ting").
ting source 1
requires semantics 1
module orchard version "1.0.0" role program

schemas:
  schema Fruit:
    name: text required
    sweet: boolean required

conclusions:
  conclusion SweetFruit:
    name: text required

rules:
  rule recognize_sweet_fruit:
    when:
      match Fruit:
        name: bind name
        sweet: true
    then:
      conclude SweetFruit:
        name: use name

tests:
  test apple_is_sweet:
    given:
      fact Fruit:
        name: "apple"
        sweet: true
      fact Fruit:
        name: "lemon"
        sweet: false
    expect:
      conclusion SweetFruit:
        name: "apple"
    firings: 1
"""
built = compile_source(source)
assert built.tests.passed and built.tests.total == 1
engine = Engine(built.program)
Fruit = built.schema("Fruit")
engine.declare(Fruit(name="pear", sweet=True))
engine.run()
assert engine.conclusions[0]["name"] == "pear"
print(engine.conclusions[0].as_dict())

Edit complete source in the IDE

The Full source editor has a compact toolbar for Undo, Redo, Cut, Copy, Paste, Select all and Find. Apply to tabs accepts the edited document; Cancel preserves the current document. The IDE uses visible labels and offline help without hover tooltips; accessible control descriptions remain available.

Saved file information

Open the IDE’s File information tab to see which .ting file you are working with. It shows the filename and full path, saved module name/version/role, UTF-8 byte size, line endings, total lines, blank lines and comment-only lines. The timestamp records when the IDE last successfully opened or saved that copy, in your local timezone.

Read the categories

The category list counts local fact types, entity types with keys, declared fields, reference fields, output types, classifications, fixed facts, rules, decisions, tests, import blocks and exports. Fields include both fact and output declarations. Entity types are a subset of fact types; reference fields are a subset of fields. An import block counts once even when it lists several names. Private declarations are included.

For the complete entity_library.ting example, the saved counts are 3 fact types, 1 entity type, 8 fields, 4 reference fields, 1 output type, 3 fixed facts, 1 rule, 2 tests and 4 exports. Test inputs and expected outputs are not counted again as schema or fixed-fact declarations.

Saved contents and unsaved edits

Counts describe the last successfully opened or saved bytes. If you edit any section or apply Full source changes, the badge says Unsaved edits and the saved counts stay unchanged. Save refreshes them; Save As also changes the displayed path. If you edit again while a save is running, the tab reports the saved snapshot and continues to mark the newer edits as unsaved. A failed save retains the previous information.

A new document says Not saved yet until its first successful save. External file changes are checked when saving or reopening; this tab does not continuously monitor disk changes. The counts are a text summary, not a successful compiler check. Imported declarations, runtime facts and inferred results are excluded. Use Check and Tests to validate behavior. Viewing the tab does not edit your source.

Write Ting rules

Each readable .ting file defines a module: a named collection of declarations. Begin with the three header lines shown in Compile readable source. The ting source 1 and requires semantics 1 lines select the language; the module line supplies your own name, three-part knowledge version and role. That module version describes your knowledge and is independent of the installed Ting version. Use program for a runnable program, library for reusable declarations with explicit exports, and test for validation. The role determines which object compilation returns; see Compiled libraries before compiling a library. A test module is not deployable.

Use two spaces per indentation level, UTF-8 text, and # comments outside quoted strings. Strings use double quotes and JSON escapes. Top-level sections such as schemas: and rules: may be reordered, but each occurs at most once. Names are ASCII identifiers; quoted values can contain Unicode. No expressions execute Python.

Scalar types are text, integer, float, boolean, bytes and null. Unions use text | null required. There are no implicit defaults or numeric conversions. true, 1 and 1.0 remain different values. Bytes use bytes "01ff"; floats must be finite.

Pattern operators are eq, ne, lt, le, gt, ge, between, one_of. Write between [10,20] or one_of [1,2,3] for literal lists. Bind a field and use separate test conditions for several comparisons. Every alternative must establish the same bindings; references to missing bindings and recursion through negation fail compilation.

Add requires feature arrays after the module header for fields such as cells: array i64[2,2] required and values such as array i64[2,2] [1,2,3,4]. Supported element types are bool, i64 and f64; see Arrays for construction and supported operations.

Embedded tests

Each named test runs in a fresh bounded session. Add given seed to include fixed program facts; otherwise the test starts without them. given: supplies facts; expect: lists the exact conclusions; an empty expect block requires none. expect facts: checks the exact final fact collection. firings: optionally checks a count. The expected status defaults to "quiescent". Tests belonging to an imported library do not run as part of the importing program; compile and test each library separately.

By default a test permits at most 1,000,000 candidate rows, 1,000,000 constraint evaluations, 100,000 activations and 10,000 firings. Pass a finite test_budget to change those limits. A failed test prevents that output file from being published. run_tests=False explicitly skips tests; zero tests is reported as zero tests.

Inspection and explanation

parse_source() retains exact text, comments, locations and syntax nodes. format_source() preserves statements and comments while normalizing line endings. analyze_source() returns typed declarations, the resolved graph and semantic identity. AuthoringBackend groups formatting, analysis, test and build operations for applications that provide their own editor.

built.origin_for(conclusion.rule) returns the rule declaration and source line. Keep the build result for this lookup. For a complete example and the distinction between rule proofs and source locations, read Explain a conclusion.

For fixed reference facts, use named seeds with requires feature seed-facts. For explicit nominal types and parent membership, use classifications. Reference fields, units, enums, named export sets and temporal declarations are not supported. Unsupported declarations produce errors rather than being ignored.

Handle build errors

Catch SourceError around compile_file(), compile_source(), save_tingir() and load_tingir(). Its diagnostic supplies a stable code, readable message, correction hint, and file/line/column when available. as_dict() is JSON-compatible. An embedded test failure also provides test_report.

from pathlib import Path
from ting_reasoner import SourceError, compile_file

project_dir = Path.home() / "ting_examples"
source_path = project_dir / "orchard.ting"
output_path = project_dir / "compiled" / "orchard.tingir"
try:
    built = compile_file(source_path, output_path)
except SourceError as error:
    print(error)
    details = error.as_dict()  # For an editor, log or application response.
else:
    print(built.output, built.digest)

Some errors concern the whole program and have no source span; check for None before reading location fields. Invalid Python argument types or option values can raise TypeError or ValueError. Inspect built.tests to distinguish passed tests, zero tests and a skipped test run.

from ting_reasoner import SourceError, SourceSnapshot, compile_source

source = """ting source 1
requires semantics 1
module example version "1.0.0" role program
schemas:
  schema Item:
    value: mystery required
"""
try:
    compile_source(SourceSnapshot(source, "example.ting"))
except SourceError as error:
    diagnostic = error.diagnostic
    assert diagnostic.code == "TING-SRC-006"
    assert diagnostic.span.line == 6
    print(error)
    print(error.as_dict())
else:
    raise AssertionError("Invalid type should fail")

Protect output and control resources

Compilation and tests finish before publication. Source files cannot be used as compiled destinations, including known path aliases. Existing outputs survive a failed rebuild. Use a destination directory controlled by your application; path checks are not a defense against concurrent malicious directory changes.

SourceLimits defaults: 1 MiB per document, 8 MiB across modules, 64 modules, 50,000 lines per document, 16 KiB per line, 100,000 nodes across the graph, depth 32, and 128 integer digits. IRLimits separately bounds the emitted program. These are logical limits, not exact process-memory accounting.

Several source files

Before this section: Compile readable source and Write Ting rules.

For one program composed from several files, put reusable declarations in library modules with explicit exports. The root imports specific names and exact module versions. Supply a module-to-file mapping to the API; the compiler does no directory or network search.

The complete in-memory example below defines both modules. To use the file-based call first, save its types_source text as fruit_types.ting and its main_source text as orchard_modular.ting in your chosen folder. Matching files are also available in the source bundle. This call requires those two saved files:

from pathlib import Path
from ting_reasoner import compile_file

project_dir = Path.home() / "ting_examples"
output_dir = project_dir / "compiled"
output_dir.mkdir(parents=True, exist_ok=True)
built = compile_file(
    project_dir / "orchard_modular.ting",
    output_dir / "orchard_modular.tingir",
    modules={"orchard.types": project_dir / "fruit_types.ting"},
)

The mapping key orchard.types is the module name declared inside the provider. Its value is the actual file path. Each module file may live in a different folder; pass that folder's full path explicitly. Relative module paths are resolved from Path.cwd(), just like other relative file arguments.

Importing a schema alone does not activate rules or seeds. Import the rule you need. Its declared requires: dependencies bring in the named helper rules and seeds automatically; Ting does not infer undeclared dependencies from matching schema names. See Make a rule’s dependencies explicit for a complete example. Aliases use schema Fruit as Produce and retain the same qualified runtime identity. Wildcards, re-exports, dynamic imports, version ranges, and import cycles are rejected. The linked program is checked against one set of global IRLimits.

from ting_reasoner import Engine, MemoryResolver, compile_source

types_source = """ting source 1
requires semantics 1
module orchard.types version "1.0.0" role library

exports:
  schema Fruit
  conclusion SweetFruit
  rule recognize_sweet_fruit

schemas:
  schema Fruit:
    name: text required
    sweet: boolean required

conclusions:
  conclusion SweetFruit:
    name: text required

rules:
  rule recognize_sweet_fruit:
    when:
      match Fruit:
        name: bind name
        sweet: true
    then:
      conclude SweetFruit:
        name: use name
"""
main_source = """ting source 1
requires semantics 1
module orchard.app version "1.0.0" role program

imports:
  from orchard.types version "1.0.0":
    schema Fruit
    conclusion SweetFruit
    rule recognize_sweet_fruit

tests:
  test pear_is_sweet:
    given:
      fact Fruit:
        name: "pear"
        sweet: true
    expect:
      conclusion SweetFruit:
        name: "pear"
    firings: 1
"""
built = compile_source(main_source,
    resolver=MemoryResolver({"orchard.types": types_source}))
assert built.tests.passed
engine = Engine(built.program)
engine.declare(built.schema("Fruit")(name="apple", sweet=True))
engine.run()
assert engine.conclusions[0]["name"] == "apple"

The source bundle includes matching fruit_types.ting and orchard_modular.ting files, plus a separate equipment-temperature example.

Use the same path rules for every source and output. Relative module paths start at the process’s working directory, as explained under relative filenames.

Build several independent programs

compile_files() takes a list of existing source paths and an output directory. It builds each source separately and returns an ordered receipt for each. To try this example, save or copy orchard.ting and equipment.ting from the source bundle's examples into your chosen folder:

from pathlib import Path
from ting_reasoner import compile_files

project_dir = Path.home() / "ting_examples"
output_dir = project_dir / "compiled"
output_dir.mkdir(parents=True, exist_ok=True)
source_paths = [project_dir / "orchard.ting", project_dir / "equipment.ting"]
reports = compile_files(source_paths, output_dir)
for report in reports:
    print(report.source, report.output, report.ok)
    if not report.ok:
        print(report.as_dict()["diagnostics"])

The destinations are output_dir / "orchard.tingir" and output_dir / "equipment.tingir". Each receipt includes the full source and output paths. Pass overwrite=True to compile_files() only when replacing previous outputs is intended. Inspect report.ok and report.diagnostics; a failed file does not stop later files. The output directory must exist. Output names must be unique, including case, and must not alias any source.

from pathlib import Path
from tempfile import TemporaryDirectory
from ting_reasoner import compile_files

valid = """ting source 1
requires semantics 1
module empty version "1.0.0" role program
"""
with TemporaryDirectory() as directory:
    source_dir = Path(directory) / "My rules"
    output_dir = Path(directory) / "Compiled programs"
    source_dir.mkdir(parents=True, exist_ok=True)
    output_dir.mkdir(parents=True, exist_ok=True)
    (source_dir / "valid.ting").write_text(valid, encoding="utf-8")
    (source_dir / "invalid.ting").write_text("bad source", encoding="utf-8")
    source_paths = [source_dir / "invalid.ting", source_dir / "valid.ting"]
    reports = compile_files(source_paths, output_dir)
    assert [report.ok for report in reports] == [False, True]
    assert Path(reports[1].output) == output_dir / "valid.tingir"
    assert (output_dir / "valid.tingir").is_file()
    assert not (output_dir / "invalid.tingir").exists()
    for report in reports:
        print(report.as_dict())

Each file is published atomically. The whole batch is not one filesystem transaction, so earlier successful outputs remain if a later root fails. The default maximum is 256 roots; use max_files to choose another explicit bound. Receipts do not retain full compiled plans or test facts.

Portable program files

To use a compiled program on another computer, install a compatible Ting version, copy the file, and pass its actual path to load_tingir(). Follow Install and verify on that computer before running your application. Preserve the editable source when you need to change the rules.

Use compile_file() to create a compiled file from readable source, and load_tingir() to load it. The loader checks the file's format and compatibility. If loading reports an unsupported format, use a compatible Ting installation or recompile the original source. Renaming a file extension does not convert its contents.

from ting_reasoner import load_ting, save_tingir
save_tingir(load_ting("compiled_rules.ting"), "compiled_rules.tingir")

Optional command line

The source commands call the same public authoring backend as Python applications. Run the following commands from the folder containing orchard.ting, or replace the filenames with full paths. Quote paths that contain spaces. The -o argument is the full compiled filename, and its parent directory must already exist. Use repeated --module NAME=PATH arguments for explicit imports.

ting-build check orchard.ting --json
ting-build test-source orchard.ting --json
ting-build compile orchard.ting -o orchard.tingir --json

Exit status 0 means success, 1 means source/test failure, and 2 means argument, I/O or compiled-artifact failure. ting-build build takes a .tingspec.json RuleSpec project, a different input format from readable .ting source. Use ting-build build --help for its options and a .tingir output. HistoryStore manages checkpoints for RuleSpec JSON projects; it does not store readable .ting source history.

Save a program created through Python

This complete example creates a knowledge base, saves it to a temporary compiled file, loads it, and declares one fact. No source file is required.

from pathlib import Path
from tempfile import TemporaryDirectory
from ting_reasoner import Engine, FactSchema, KnowledgeBase, load_tingir, save_tingir

Fruit = FactSchema("fruit", name=str)
knowledge = KnowledgeBase((Fruit,))
with TemporaryDirectory() as directory:
    path = Path(directory) / "fruit.tingir"
    save_tingir(knowledge, path)
    engine = Engine(load_tingir(path))
    engine.declare(Fruit(name="apple"))
    assert len(engine.facts) == 1

Include fixed knowledge

Before this section: Compile readable source; its two-file example also uses source imports.

Put stable facts, such as equipment limits or reference data, in a program's knowledge: region. Each named seed is a typed, immutable fact. It is stored in the compiled program and available when an Engine is created, including in independent process workers.

1. Choose a folder and define reusable knowledge

Create a folder for these two source files and print their full paths:

from pathlib import Path

project_dir = Path.home() / "ting_examples" / "inspection"
project_dir.mkdir(parents=True, exist_ok=True)
print("Save thresholds here:", project_dir / "thresholds.ting")
print("Save the program here:", project_dir / "inspection.ting")

Save the following source at project_dir / "thresholds.ting". The value is illustrative; choose limits appropriate to your application. The feature declaration makes the file's requirements explicit. Export both the schema and the seed when consumers need both.

ting source 1
requires semantics 1
module equipment.limits version "1.0.0" role library
requires feature seed-facts

exports:
  schema Threshold
  seed inspection_limit

schemas:
  schema Threshold:
    metric: text required
    maximum: integer required

knowledge:
  seed inspection_limit: Threshold:
    metric: "vibration"
    maximum: 20

tests:
  test fixed_limit:
    given seed
    expect facts:
      fact Threshold:
        metric: "vibration"
        maximum: 20

2. Use the knowledge in an inspection rule

Save this source at project_dir / "inspection.ting". The rule compares a session reading with the stored threshold. Its result is a typed record containing the device, measurement and limit.

ting source 1
requires semantics 1
module equipment.inspection version "1.0.0" role program
requires feature seed-facts

imports:
  from equipment.limits version "1.0.0":
    schema Threshold
    seed inspection_limit

schemas:
  schema Reading:
    device: text required
    metric: text required
    value: integer required

conclusions:
  conclusion InspectionRequired:
    device: text required
    measured: integer required
    limit: integer required

rules:
  rule inspection_required:
    when:
      match Threshold:
        metric: bind metric
        maximum: bind maximum
      match Reading:
        device: bind device
        metric: eq use metric
        value: bind measured
      test gt:
        use measured
        use maximum
    then:
      conclude InspectionRequired:
        device: use device
        measured: use measured
        limit: use maximum

tests:
  test above_limit:
    given seed
    given:
      fact Reading:
        device: "pump_a"
        metric: "vibration"
        value: 25
    expect:
      conclusion InspectionRequired:
        device: "pump_a"
        measured: 25
        limit: 20
    firings: 1
  test at_limit:
    given seed
    given:
      fact Reading:
        device: "pump_a"
        metric: "vibration"
        value: 20
    expect:
    firings: 0
  test without_program_knowledge:
    given:
      fact Reading:
        device: "pump_a"
        metric: "vibration"
        value: 25
    expect:
    firings: 0

3. Compile and evaluate a reading

from ting_reasoner import Engine, compile_file, load_tingir

# project_dir is the folder selected in step 1.
output_dir = project_dir / "compiled"
output_dir.mkdir(parents=True, exist_ok=True)
built = compile_file(project_dir / "inspection.ting",
    output_dir / "inspection.tingir",
    modules={"equipment.limits": project_dir / "thresholds.ting"})
engine = Engine(load_tingir(built.output))
Reading = engine.schema("equipment.inspection.Reading")
engine.declare(Reading(device="pump_a", metric="vibration", value=25))
result = engine.run()
if result.status.value == "quiescent":
    print([item.as_dict() for item in engine.conclusions])
# [{"device": "pump_a", "measured": 25, "limit": 20}]

Check the run status before treating a result as complete. In this example, no conclusion means the inspection rule did not match. It does not assert that the equipment is safe or that every needed observation was supplied.

Keep stable reference facts in knowledge: and supply changing observations with declare(). Importing a schema alone does not import its provider’s seeds. In this example, import seed inspection_limit explicitly. An imported rule also brings in seeds listed in its requires: dependencies; see rule dependencies. An alias such as seed inspection_limit as local_name changes the source spelling while preserving the qualified runtime name.

Lifetime and error handling

engine.seed_facts lists named seeds. engine.seed_handle(name) looks up an exact qualified name and raises KeyError if absent. engine.seed_names_for(handle) and snapshot.seed_names_for(handle) identify fixed knowledge among a proof's supporting facts. An empty tuple means the handle is not a seed in that session. Handles remain session-local.

Keep the build result to locate a declaration with built.origin_for("equipment.limits.inspection_limit", kind="seed"), or a value with kind="seed-field" and name equipment.limits.inspection_limit.maximum. The compiled file preserves seed identities and values; source locations require the source analysis.

Declare fixed knowledge through Python

from ting_reasoner import (
    Engine, FactSchema, ImmutableKnowledgeError, KnowledgeBase, SeedFact,
)

Threshold = FactSchema("threshold", metric=str, maximum=int)
knowledge = KnowledgeBase(seeds=(
    SeedFact("inspection_limit", Threshold(metric="vibration", maximum=20)),
))
engine = Engine(knowledge)
handle = engine.seed_handle("inspection_limit")
before = engine.snapshot()
try:
    engine.retract(handle)
except ImmutableKnowledgeError as error:
    assert error.code == "TING-KNOWLEDGE-001"
    print(error.as_dict())
else:
    raise AssertionError("Fixed knowledge must be protected")
assert engine.snapshot() == before
engine.reset()
assert engine.seed_facts == knowledge.seeds
assert engine.snapshot().seed_names_for(handle) == ("inspection_limit",)

Tests and resource limits

Embedded tests start without program seeds unless they include given seed. Use it with given: to test fixed knowledge and changing observations together. Test fixtures never become deployed facts. A failed test protects any existing compiled output.

Seeds accept typed literals, including supported array values; they cannot execute expressions or callbacks. IRLimits defaults to at most 65,536 named seeds and 1,000,000 seed field values. Overall source, payload, constant and verification-work limits also apply. RuleSpec JSON cannot represent seeds; use readable .ting or the Python API for this knowledge.

from ting_reasoner import Engine, MemoryResolver, compile_source

limits = """ting source 1
requires semantics 1
module equipment.limits version "1.0.0" role library
requires feature seed-facts

exports:
  schema Threshold
  seed inspection_limit

schemas:
  schema Threshold:
    metric: text required
    maximum: integer required

knowledge:
  seed inspection_limit: Threshold:
    metric: "vibration"
    maximum: 20

tests:
  test fixed_limit:
    given seed
    expect facts:
      fact Threshold:
        metric: "vibration"
        maximum: 20
"""
inspection = """ting source 1
requires semantics 1
module equipment.inspection version "1.0.0" role program
requires feature seed-facts

imports:
  from equipment.limits version "1.0.0":
    schema Threshold
    seed inspection_limit

schemas:
  schema Reading:
    device: text required
    metric: text required
    value: integer required

conclusions:
  conclusion InspectionRequired:
    device: text required
    measured: integer required
    limit: integer required

rules:
  rule inspection_required:
    when:
      match Threshold:
        metric: bind metric
        maximum: bind maximum
      match Reading:
        device: bind device
        metric: eq use metric
        value: bind measured
      test gt:
        use measured
        use maximum
    then:
      conclude InspectionRequired:
        device: use device
        measured: use measured
        limit: use maximum

tests:
  test above_limit:
    given seed
    given:
      fact Reading:
        device: "pump_a"
        metric: "vibration"
        value: 25
    expect:
      conclusion InspectionRequired:
        device: "pump_a"
        measured: 25
        limit: 20
    firings: 1
  test at_limit:
    given seed
    given:
      fact Reading:
        device: "pump_a"
        metric: "vibration"
        value: 20
    expect:
    firings: 0
  test without_program_knowledge:
    given:
      fact Reading:
        device: "pump_a"
        metric: "vibration"
        value: 25
    expect:
    firings: 0
"""
built = compile_source(inspection,
    resolver=MemoryResolver({"equipment.limits": limits}))
assert built.tests.passed and built.tests.total == 3
engine = Engine(built.program)
Reading = built.schema("Reading")
engine.declare(Reading(device="pump_a", metric="vibration", value=25))
result = engine.run()
assert result.status.value == "quiescent"
assert engine.conclusions[0].as_dict() == {
    "device": "pump_a", "measured": 25, "limit": 20,
}
assert engine.seed_names_for(engine.seed_handle("equipment.limits.inspection_limit"))
engine.reset()
assert len(engine.facts) == 1 and not engine.conclusions

Classifications and inheritance

Before this section: Facts and types; the file example also uses Ting source declarations.

A schema describes record fields. A classification explicitly states what kind of thing those values represent. A pump and a fan may have the same fields, but sharing fields does not make a fan a pump. Declare membership when your application has the evidence to do so.

A namespace is the dot-separated module name, such as equipment.types. It keeps names distinct: equipment.types.Pump is the fully qualified classification name. A module may contain schemas, classifications, rules, conclusions and fixed facts together. There is no separate classification file type.

Declare classifications in a .ting file

Save this text as UTF-8 at a path of your choice, for example Path.home() / "ting_project" / "editable" / "equipment_types.ting". Create the parent folders first. The complete example in Compiled libraries writes and compiles the files for you.

ting source 1
requires semantics 1
module equipment.types version "1.0.0" role library
requires feature classifications

exports:
  schema Equipment
  class Machine
  class Pump
  class CentrifugalPump

schemas:
  schema Equipment:
    device: text required
    vibration: integer required

classifications:
  class Machine applies_to Equipment
  class Pump is Machine
  class CentrifugalPump is Pump

requires feature classifications enables these declarations. applies_to Equipment identifies the record schema. is Pump declares a parent; class Hybrid is Pump, Fan declares multiple parents when they share the same record schema. Cycles, unknown parents, conflicting definitions and incompatible schemas are rejected.

exports: lists the declarations another application may select. Unlisted declarations are private, but Ting can include a private declaration when a selected public rule explicitly requires it. The header's role library makes compilation return a LibraryBuild; no additional classification-file flag is needed.

Use classifications through Python

from ting_reasoner import Classification, Engine, FactSchema, KnowledgeBase

Equipment = FactSchema("equipment.Equipment", device=str, vibration=int)
Machine = Classification("equipment.Machine", Equipment)
Pump = Classification("equipment.Pump", Equipment, parents=(Machine,))
knowledge = KnowledgeBase(classifications=(Pump,))  # Includes its parents.
engine = Engine(knowledge)

record = Equipment(device="pump_a", vibration=25)
engine.declare(record)
engine.run()
assert engine.facts == (record,)  # Shape alone does not establish membership.

membership = Pump.classify(record)  # Explicitly classify these exact values.
engine.declare(membership)
engine.run()  # Derives Machine membership and the underlying Equipment record.
assert Pump.is_a(Machine)
assert Machine(device="pump_a", vibration=25) in engine.facts
assert len(engine.facts) == 3

Membership, inference and changing values

Membership belongs to exact record values. Classifying device "pump_a" with vibration 25 does not classify a later record with vibration 30. There is no implicit entity identifier or reference relationship. Classify the new values explicitly when updating input.

engine.run() derives parent memberships and the underlying record through ordinary deterministic rules. These steps consume the same firing and work budgets as other rules. Merely declaring a child does not immediately populate all parents. A stopped or exhausted run can therefore have incomplete inheritance; check its status and resume with an appropriate budget. Use engine.reset() to start a new case; named seeds are retained.

A source rule uses match Pump: in the same way as match Equipment:, followed by field constraints. A derived classification uses derive Pump:. Classification facts also work in named seeds and embedded tests. Generated inheritance rules use the reserved namespace ting.classification.

Compile and import selected knowledge

Before this section: Source imports, fixed knowledge and classifications.

Use compiled libraries when different applications need different parts of the same knowledge. You can keep several .tingir files in different folders, then select only the required public classifications, schemas, rules or fixed facts. A public export is a declaration made available to other modules. Selecting it also includes its declared dependencies: the other declarations it needs.

The header determines what is compiled

Both compiled kinds use the .tingir extension. A binary header identifies the kind automatically. A compiled library offers declarations for selection; an executable is a program ready to run. Pass linked.program to Engine after selection. A CompiledLibrary itself is not executable.

Make a rule's dependencies explicit

Save this source as maintenance.ting. It uses the equipment_types.ting source from Classifications and inheritance. Its import refers to the module declared inside equipment_types.ting; the Python mapping below supplies that file's actual path.

ting source 1
requires semantics 1
module equipment.maintenance version "1.0.0" role library
requires feature classifications
requires feature seed-facts

imports:
  from equipment.types version "1.0.0":
    class Pump

exports:
  rule inspect_pump
  rule record_pump
  conclusion InspectionRequired

schemas:
  schema Limit:
    maximum: integer required

conclusions:
  conclusion InspectionRequired:
    device: text required
    measured: integer required
  conclusion PumpRecorded:
    device: text required

knowledge:
  seed vibration_limit: Limit:
    maximum: 20

rules:
  rule inspect_pump:
    requires:
      seed vibration_limit
    when:
      match Pump:
        device: bind device
        vibration: bind measured
      match Limit:
        maximum: bind maximum
      test gt:
        use measured
        use maximum
    then:
      conclude InspectionRequired:
        device: use device
        measured: use measured
  rule record_pump:
    when:
      match Pump:
        device: bind device
    then:
      conclude PumpRecorded:
        device: use device

tests:
  test high_vibration:
    given seed
    given:
      fact Pump:
        device: "pump_a"
        vibration: 25
    expect:
      conclusion InspectionRequired:
        device: "pump_a"
        measured: 25
      conclusion PumpRecorded:
        device: "pump_a"

Selecting inspect_pump automatically includes the Pump type, its parents, its record schema and the typed conclusion. The rule's requires: block also includes the private vibration_limit seed. You can list rule helper_name there when a helper rule is essential. Naming a schema alone does not select every rule that might produce facts of that schema; declare those rule dependencies explicitly.

Complete example: compile libraries, select a rule and explain its result

This example is self-contained. It creates temporary folders, writes both sources, compiles them, selects the inspection rule and runs it. The temporary files are removed on exit. For files you want to keep, replace the temporary-directory setup with project_dir = Path.home() / "ting_project" and keep the remaining operations in your application.

from pathlib import Path
from tempfile import TemporaryDirectory
from ting_reasoner import (
    CompiledCatalog, CompiledLibrary, Engine, LibraryError, SourceError,
    compile_file, load_tingir, save_tingir,
)

types_source = """ting source 1
requires semantics 1
module equipment.types version "1.0.0" role library
requires feature classifications

exports:
  schema Equipment
  class Machine
  class Pump
  class CentrifugalPump

schemas:
  schema Equipment:
    device: text required
    vibration: integer required

classifications:
  class Machine applies_to Equipment
  class Pump is Machine
  class CentrifugalPump is Pump
"""
maintenance_source = """ting source 1
requires semantics 1
module equipment.maintenance version "1.0.0" role library
requires feature classifications
requires feature seed-facts

imports:
  from equipment.types version "1.0.0":
    class Pump

exports:
  rule inspect_pump
  rule record_pump
  conclusion InspectionRequired

schemas:
  schema Limit:
    maximum: integer required

conclusions:
  conclusion InspectionRequired:
    device: text required
    measured: integer required
  conclusion PumpRecorded:
    device: text required

knowledge:
  seed vibration_limit: Limit:
    maximum: 20

rules:
  rule inspect_pump:
    requires:
      seed vibration_limit
    when:
      match Pump:
        device: bind device
        vibration: bind measured
      match Limit:
        maximum: bind maximum
      test gt:
        use measured
        use maximum
    then:
      conclude InspectionRequired:
        device: use device
        measured: use measured
  rule record_pump:
    when:
      match Pump:
        device: bind device
    then:
      conclude PumpRecorded:
        device: use device

tests:
  test high_vibration:
    given seed
    given:
      fact Pump:
        device: "pump_a"
        vibration: 25
    expect:
      conclusion InspectionRequired:
        device: "pump_a"
        measured: 25
      conclusion PumpRecorded:
        device: "pump_a"
"""

# This demonstration removes its temporary folders when it finishes.
# In your application use Path.home() / "ting_project", or your chosen folder.
with TemporaryDirectory() as temporary:
    project_dir = Path(temporary)
    source_dir = project_dir / "editable"
    types_dir = project_dir / "compiled_types"
    rules_dir = project_dir / "compiled_rules"
    for folder in (source_dir, types_dir, rules_dir):
        folder.mkdir(parents=True, exist_ok=True)

    types_path = source_dir / "equipment_types.ting"
    maintenance_path = source_dir / "maintenance.ting"
    with types_path.open("x", encoding="utf-8") as file:
        file.write(types_source)
    with maintenance_path.open("x", encoding="utf-8") as file:
        file.write(maintenance_source)

    try:
        types_build = compile_file(types_path, types_dir / "types.tingir")
        rules_build = compile_file(
            maintenance_path, rules_dir / "maintenance.tingir",
            modules={"equipment.types": types_path},  # Editable source path.
        )
    except SourceError as error:
        print(error.as_dict())
        raise

    assert types_build.kind == "library"  # Determined by the .ting header.
    opened = load_tingir(types_build.output)
    assert isinstance(opened, CompiledLibrary)
    print(opened.module, opened.version)
    print([(item.kind, item.qualified_name) for item in opened.exports])

    catalog = CompiledCatalog({
        "equipment.types": types_build.output,  # Compiled paths at runtime.
        "equipment.maintenance": rules_build.output,
    })
    try:
        linked = catalog.link(
            classifications=("equipment.types.CentrifugalPump",),
            rules=("equipment.maintenance.inspect_pump",),
        )
    except LibraryError as error:
        print(error.code, error.message, error.hint, error.chain)
        raise

    # Required parents, schema, typed conclusion and vibration_limit are loaded.
    # The unrelated record_pump business rule is not part of this program.
    assert len(linked.knowledge_base.rules) == 1
    Pump = linked.classification("equipment.types.CentrifugalPump")
    engine = Engine(linked.program)
    engine.declare(Pump(device="pump_a", vibration=25))
    engine.run()
    assert len(engine.conclusions) == 1
    assert engine.conclusions[0].as_dict() == {"device": "pump_a", "measured": 25}
    explanation = linked.explain(engine.snapshot(), engine.conclusions[0])
    assert explanation.proofs
    print(linked.usage.sections, linked.usage.section_bytes)

    # Keep the selected executable if this application always uses these imports.
    program_path = project_dir / "pump_inspection.tingir"
    save_tingir(linked.program, program_path)
    another_session = Engine(load_tingir(program_path))
    assert not another_session.conclusions

    # Directly requesting a private seed produces a structured selection error.
    try:
        catalog.link(seeds=("equipment.maintenance.vibration_limit",))
    except LibraryError as error:
        assert error.code == "TING-LIB-003"
        assert error.chain
    else:
        raise AssertionError("Private exports must not be directly selectable")

Choose persistent paths

from pathlib import Path
from ting_reasoner import CompiledCatalog

project_dir = Path.home() / "ting_project"
catalog = CompiledCatalog({
    "equipment.types": project_dir / "compiled_types" / "types.tingir",
    "equipment.maintenance": project_dir / "compiled_rules" / "maintenance.tingir",
})
linked = catalog.link(rules=("equipment.maintenance.inspect_pump",))

The files must already exist at those paths. The temporary example above deletes its files when it finishes; use its persistent-folder variation before running this continuation. Follow the path instructions when choosing another location. Relative catalog paths are anchored to Path.cwd() when the catalog is created.

During compilation, modules={...} maps imported module names to editable .ting sources. During runtime selection, CompiledCatalog({...}) maps names to compiled .tingir libraries or already opened CompiledLibrary objects. Runtime selection does not need the original source files. Compiling new source rules against only a binary provider is not supported; retain its source for that operation.

Select exactly what your application needs

Use tuples or lists of strings; the comma in ("module.Name",) creates a one-item tuple. Passing a bare string is an error. You may combine keywords. Duplicate selections are harmless, and selection order does not change the linked program. linked.imports lists the complete selected dependency closure as (module, kind, name) tuples. linked.classification() and linked.schema() require fully qualified names.

linked.program is reusable across independent Engines and with run_batch(). Link once when the application starts, then reuse that object for many cases. linked.usage reports the amount of knowledge read for that selection. Its counts are useful when comparing selections; they do not measure total memory or elapsed time.

Verification and compatibility

Opening a library verifies its bounded directory, identity and section descriptors. It does not inspect unread payloads. Linking verifies the selected payloads and their dependency closure before returning an executable. Call library.verify_all() to read and verify every payload explicitly; this checks that file, without resolving other libraries. An unselected damaged section may therefore remain unnoticed until selected or fully verified.

Dependencies pin both an exact module version and a semantic contract digest. A changed provider can require recompiling its consumers even when its version string was not changed. Source comments and file locations do not change that semantic contract. SHA-256 checks detect corruption and identity changes; they do not authenticate who created a file. No source parser, Python import, network lookup or folder scan runs while selecting compiled knowledge.

Selected source fragments retain declaration text and original file labels for linked.explain(), even if those source files are absent. These are direct proof explanations. Saving just linked.program stores the executable without this source map; retain the LinkedBuild or reload and link the original libraries when source explanations are required.

Handle errors and bound work

Catch SourceError around source compilation and load_tingir(). Catch LibraryError around catalog selection, CompiledLibrary.open() and verify_all(). LibraryError exposes code, message, hint, chain and as_dict(). The chain names the selected export and dependencies leading to the problem. Wrong Python argument types raise TypeError or ValueError.

from ting_reasoner import CompiledCatalog, IRLimits, LibraryLimits

# Reuse the catalog paths from the preceding example.
catalog = CompiledCatalog({
    "equipment.types": project_dir / "compiled_types" / "types.tingir",
    "equipment.maintenance": project_dir / "compiled_rules" / "maintenance.tingir",
}, limits=LibraryLimits(max_selected_bytes=8 * 1024 * 1024, max_sections=100))
linked = catalog.link(rules=("equipment.maintenance.inspect_pump",),
                      ir_limits=IRLimits(max_rules=100))

LibraryLimits bounds file, directory, section, build and selected byte sizes; module and section counts; dependency traversal and depth; and retained source nodes. IRLimits bounds each selected Compact-IR payload and the final executable. An IR policy supplied to load_tingir(limits=...) is retained for lazy payload verification; an explicit link(ir_limits=...) replaces it for that operation. ExecutionBudget separately bounds reasoning. Classification graphs have a hard maximum of 4,096 declarations and 64 levels; lower IRLimits may tighten these.

A failed link returns no executable and changes no existing Engine. Compilation validates and runs embedded tests before publishing a file; existing outputs require overwrite=True. Compile libraries independently to run each library's embedded tests. A consumer's build does not automatically execute the provider's tests.

Reuse entities from compiled libraries

An indexed library lets an application select a named type, rule or fixed entity without loading unrelated payloads. Entity identities remain qualified keys; source paths, row addresses and handles are not part of those identities.

Write and compile a reusable linked-equipment library

Save this complete source as entity_library.ting. The two pumps reference each other. The private spare is unrelated. The exported rule explicitly requires pump_a; selecting that rule also selects pump_b to keep the backup relationship valid. Both embedded tests include fixed knowledge.

ting source 1
requires semantics 1
module entity_library version "1.0.0" role library
requires feature entity-references
requires feature seed-facts

schemas:
  schema Asset key(serial):
    serial: text required
    backup: reference Asset required
    location: text required
  schema Request:
    asset: reference Asset required
  schema Alarm:
    asset: reference Asset required
    code: text required

conclusions:
  conclusion SelectedAlarm:
    asset: reference Asset required
    code: text required

knowledge:
  seed pump_a: Asset:
    serial: "pump-a"
    backup: ref Asset ["pump-b"]
    location: "north"
  seed pump_b: Asset:
    serial: "pump-b"
    backup: ref Asset ["pump-a"]
    location: "south"
  seed spare: Asset:
    serial: "spare"
    backup: ref Asset ["spare"]
    location: "store"

rules:
  rule select_alarm:
    requires:
      seed pump_a
    when:
      match Request:
        asset: bind selected
      match Alarm:
        asset: eq use selected
        code: bind code
    then:
      conclude SelectedAlarm:
        asset: use selected
        code: use code

tests:
  test selected_reference:
    given seed
    given:
      fact Request:
        asset: ref Asset ["pump-a"]
      fact Alarm:
        asset: ref Asset ["pump-a"]
        code: "temperature"
      fact Alarm:
        asset: ref Asset ["pump-b"]
        code: "pressure"
    expect:
      conclusion SelectedAlarm:
        asset: ref Asset ["pump-a"]
        code: "temperature"
    firings: 1
  test all_alternative_matches:
    given seed
    given:
      fact Request:
        asset: ref Asset ["pump-a"]
      fact Alarm:
        asset: ref Asset ["pump-a"]
        code: "temperature"
      fact Alarm:
        asset: ref Asset ["pump-a"]
        code: "pressure"
    expect:
      conclusion SelectedAlarm:
        asset: ref Asset ["pump-a"]
        code: "temperature"
      conclusion SelectedAlarm:
        asset: ref Asset ["pump-a"]
        code: "pressure"
    firings: 2

exports:
  schema Asset
  rule select_alarm
  seed pump_a
  conclusion SelectedAlarm

Compile with ting-build compile entity_library.ting -o entity_library.tingir, or use the IDE with General → Library and Compile. The compiler validates all fixed reference targets and runs the embedded tests before publishing the file. The File information tab describes the saved source and its local declaration counts.

Select and run without the source

Run this complete script beside the compiled library. You may remove the editable source first. A schema selection is useful when the application supplies its own entities; a rule selection can carry explicitly required fixed knowledge.

from ting_reasoner import CompiledCatalog, Engine, EntityRef, load_tingir

# The .ting source is not needed to load or select this library.
library = load_tingir("entity_library.tingir")
catalog = CompiledCatalog({"entity_library": library})

# A type-only selection includes target types but activates no fixed facts.
types = catalog.link(schemas=("entity_library.Asset",))
assert types.knowledge_base.seeds == ()
assert types.knowledge_base.rules == ()

# The rule explicitly requires pump_a. Its backup link also needs pump_b.
linked = catalog.link(rules=("entity_library.select_alarm",))
# Selected bytes include each shared program payload only once.
assert linked.usage.programs > 0
assert 0 < linked.usage.program_bytes < linked.usage.section_bytes
engine = Engine(linked.program)
assert len(engine.facts) == 2
assert ("entity_library", "seed", "spare") not in linked.imports

Request = linked.schema("entity_library.Request")
Alarm = linked.schema("entity_library.Alarm")
selected = EntityRef("entity_library.Asset", ("pump-a",))
engine.declare(Request(asset=selected), Alarm(asset=selected, code="temperature"))
engine.run()
result = engine.conclusions[0]
print(result["code"], engine.resolve(result["asset"])["location"])
# Output: temperature north
assert len(engine.explain(result)[0].fact_ids) == 2
assert linked.explain(engine, result).proofs
assert linked.origin_for("entity_library.pump_b", kind="seed").name == "entity_library.pump_b"

Understand which declarations are loaded

Schema and conclusion selections include the target types required by their reference fields. They do not implicitly select seeds or business rules. Selecting a seed follows its reference values to the fixed target facts, including forward links and cycles; unrelated fixed facts remain unselected. If several named fixed facts describe the same referenced entity, their names travel together. Rules bring their declared requirements and typed outputs. Private declarations can satisfy dependencies but cannot be requested directly as public exports.

Cross-module targets require explicit catalog mappings and the exact compatible versions and semantic contracts used at compilation. Source imports still use explicit source mappings. CompiledCatalog performs runtime selection from compiled files; it does not parse source or search for modules.

Verification, costs and failure behavior

The compiler writes entity libraries in indexed container format 4. Identical closed program fragments share one payload within a module, using program format 9 / semantics 10 for entities or format 10 / semantics 11 for binding-based traversal. Each declaration retains its own source text, export visibility and dependency list. For a cycle in which every seed reaches the same graph, those seeds share the compiled graph. The linker reads and verifies that payload once during each link, then checks each selected declaration against it. No verification cache survives between link operations.

Seed dependencies name immediate fixed targets; the catalog follows them to select the complete graph. Shared storage does not export private seeds, activate unrelated facts or remove source evidence. Differing closures can still overlap in storage: for example, successive roots of a long one-way chain contain different reachable graphs. Such cases can still grow quadratically and remain subject to the configured limits.

Opening verifies the directory. Linking verifies selected sections and their dependencies; library.verify_all() checks every payload. linked.usage.sections counts selected declarations. section_bytes counts their metadata plus each selected shared program once; programs and program_bytes identify the shared program portion. These last two counters are zero for older containers with embedded programs. LibraryLimits constrain file, section, build, dependency depth and selected bytes. The section limit applies to one declaration’s metadata plus its program, even when that program is shared. IRLimits applies to every unique selected program and the final linked program. Large or deeply connected graphs can require deliberate limit increases. A missing dependency, mismatched definition, wrong key, unresolved fixed target or corrupt selected payload rejects linking before an Engine is created. These byte limits bound serialized payloads, not total Python memory. Readers also accept existing entity libraries in format 3. Non-entity libraries retain format 1 or 2. Older Ting versions that lack the format-4 reader reject new entity libraries; update the runtime and compiler together. Module versions and provider contracts still must match exactly.

RuleSpec version 1 still cannot serialize entity references. Match target facts explicitly when their properties affect a rule; Python resolve() does not add evidence or a hidden dependency. Use explicit follow conditions to match target properties with their fact support. Dynamic action construction is available in program profile 11/12; it creates identity without target support. Implicit dereference expressions are unsupported.

Reuse a compiled program

Compile rules once, then reuse the returned CompactIR program object for independent cases. The program is immutable; each Engine has its own facts, handles and results.

The first two rows assume role program. For role library, use CompiledCatalog and reuse linked.program. The path variables are introduced in Compile readable source. A program contains rules and fixed knowledge; a session adds observations for one case. Reusing program avoids repeating its verification when the bytes and verification limits are unchanged. Passing raw bytes or a knowledge base to each new Engine repeats that work.

from ting_reasoner import CompactIR, Engine, FactSchema, KnowledgeBase, Rule, bind, emit, use

Reading = FactSchema("reading", device=str, value=int)
rule = Rule("record", (Reading.pattern(device=bind("device"), value=bind("value")),),
    (emit("recorded", device=use("device"), value=use("value")),))
knowledge = KnowledgeBase((Reading,), (rule,))
program = CompactIR(knowledge.compile())  # Compile and verify once here.

first = Engine(program)
second = Engine(program)  # A separate session using the same program.
first.declare(Reading(device="pump_a", value=25))
second.declare(Reading(device="pump_b", value=10))
first.run()
second.run()
assert first.conclusions[0]["device"] == "pump_a"
assert second.conclusions[0]["device"] == "pump_b"
first.reset()
assert not first.conclusions
assert second.conclusions[0]["device"] == "pump_b"

first and second are two Python variables holding separate Engine objects. Resetting first clears its session facts and results, while retaining any compiled fixed knowledge. It does not affect second. Fact handles belong to their own sessions; never transfer them between Engines.

For consecutive cases in one session, engine.reset(*facts) can also reuse its program and replace session inputs. The * expands a collection of facts into individual arguments. Save a snapshot before reset if you need to retain the old result.

For many independent cases, pass the same program to batch execution. Use Choose an execution mode to select between a reusable Engine, separate Engines and workers.

Loading with a more permissive IRLimits policy does not raise the Engine’s default limits. Keep runnable programs within those defaults; reduce the selected knowledge if Engine construction rejects an oversized program. See IRLimits for the limit fields.

Explain a conclusion

Before this section: Run and inspect; source locations also require a build from readable source.

Keep the exact conclusion returned by the engine and pass it to engine.explain(conclusion). The result contains its proof edges. You can also inspect a fact handle, a conclusion name, or all recorded provenance. A string selects a conclusion name, not a rule name. A snapshot exposes the same explanation interface after the live session changes. The short continuation below uses engine from Your first reasoner, after its successful run.

conclusion = engine.conclusions[0]
proofs = engine.explain(conclusion)
for proof in proofs:
    print(proof)

snapshot = engine.snapshot()
assert snapshot.explain(conclusion) == proofs

Proofs identify the rule, bindings and fact support. Multiple justifications may support the same logical output. Match and optimization metrics describe physical work; they are not confidence scores or probabilistic evidence.

Find the source behind a result

When you compile readable source, retain the returned SourceBuild object, called built in these examples. built.explain(engine, conclusion) adds source locations and fact values to a recorded proof. It returns a SourceExplanation and does not run inference or modify the session.

A proof edge records which rule and direct fact supports produced an output. A binding is a value captured by a rule. The explanation connects this runtime evidence to the source declaration. It is structured data that your application can inspect or display.

Complete example, including a saved snapshot

This example includes its readable source and needs no files. The URI memory:orchard is a label for that text. When using compile_file(source_path), the source map instead retains the file location supplied by the file reader.

import json
from ting_reasoner import Engine, ExplanationError, ExplanationLimits, SourceSnapshot, compile_source

# This complete example keeps source in memory; it does not read or write a file.
source = """# Edit this file, then call compile_file("orchard.ting").
ting source 1
requires semantics 1
module orchard version "1.0.0" role program

schemas:
  schema Fruit:
    name: text required
    sweet: boolean required

conclusions:
  conclusion SweetFruit:
    name: text required

rules:
  rule recognize_sweet_fruit:
    when:
      match Fruit:
        name: bind name
        sweet: true
    then:
      conclude SweetFruit:
        name: use name

tests:
  test apple_is_sweet:
    given:
      fact Fruit:
        name: "apple"
        sweet: true
      fact Fruit:
        name: "lemon"
        sweet: false
    expect:
      conclusion SweetFruit:
        name: "apple"
    firings: 1
"""
built = compile_source(SourceSnapshot(source, uri="memory:orchard"))
Fruit = built.schema("Fruit")
engine = Engine(built.program)
engine.declare(Fruit(name="pear", sweet=True))
run = engine.run()
assert run.status.value == "quiescent"
snapshot = engine.snapshot()

for conclusion in snapshot.conclusions:
    explanation = built.explain(snapshot, conclusion,
        limits=ExplanationLimits(max_proofs=20, max_supports=100))
    for proof in explanation.proofs:
        print(proof.rule.name, proof.rule.span.uri, proof.rule.span.line)
        for support in proof.supports:
            print(support.handle.id, support.fact.as_dict())
    print(json.dumps(explanation.as_dict(), indent=2))

# A saved snapshot remains usable after the live session changes.
conclusion = snapshot.conclusions[0]  # The example's pear produces one result.
engine.reset()
assert built.explain(snapshot, conclusion).proofs
try:
    built.explain(engine, conclusion)
except ExplanationError as error:
    assert error.code == "TING-EXPLAIN-002"  # Not present in the current session.
    print(error.message, error.hint)
else:
    raise AssertionError("A missing conclusion must be reported")

The nested loops first visit conclusions, then their proofs, then supporting facts. json.dumps(..., indent=2) converts the dictionary to formatted JSON text for display. snapshot.epoch identifies the captured state within that session. Epochs and numeric handles do not identify a session globally.

Fixed knowledge and exact statement locations

For a program containing named seeds, support.seeds identifies their declarations, including imported providers. An empty tuple means the support has no fixed-knowledge names. Equal seed aliases can share one fact while retaining several names. The continuation below requires built from Include fixed knowledge, rather than the orchard build above. It locates the threshold field in that program:

# With built from the equipment inspection example:
origin = built.origin_for("equipment.limits.Threshold.maximum", kind="schema-field")
statement = built.analysis.node(origin.node_id)
print(statement.text, origin.span.uri, origin.span.line)

Handle explanation errors

Catch ExplanationError and inspect error.code, error.message, error.hint or error.as_dict(). Invalid argument types raise TypeError; invalid limit values raise ValueError. Limits default to 100,000 inspected snapshot records, 1,000 selected proofs, 10,000 direct support references and 10,000 statement references. Repeated references count toward the output limits. These limits do not bound snapshot creation time or JSON byte size.

Interpret the evidence correctly

Check RunResult.status to know whether inference finished. Explanations describe recorded direct evidence; they do not prove that every possible justification was found, recursively expand a support graph, or provide a confidence score. The rule’s source statements include all branches and actions. Their presence does not identify which OR branch matched, which action produced a particular output, or provide an absence witness for a negated condition.

The compiled program digest must match. Source comments and line breaks can change without changing compiled bytes, so equivalent source may supply a different map. explanation.sources records the exact documents used for that map. Digests identify content; they are not signatures. Snapshots and builds are trusted in-process objects, not independently authenticated evidence.

An executable .tingir program does not retain the readable source map. Keep built in your application, or preserve the .ting files and rebuild them to recover source context. If you only retain the executable, engine.explain() still provides rule proofs. Compiled libraries can retain selected source fragments for linked.explain(); saving just linked.program does not preserve that source map.

Keep an earlier explanation

Call engine.snapshot() to save an immutable view of the current facts, conclusions and explanations. Later changes to the Engine do not change this view. Ask the saved snapshot for its explanation; the current Engine may no longer contain that conclusion.

from ting_reasoner import Engine, FactSchema, KnowledgeBase, Rule, bind, emit, use

Reading = FactSchema("reading", sensor=str)
rule = Rule("record_sensor", Reading.pattern(sensor=bind("sensor")),
            [emit("Seen", sensor=use("sensor"))])
engine = Engine(KnowledgeBase([Reading], [rule]))
handle = engine.declare(Reading(sensor="sensor-1"))
engine.run()

saved = engine.snapshot()
conclusion = saved.conclusions[0]
engine.retract(handle)
engine.run()
assert engine.conclusions == ()

# The saved result still describes the earlier facts and their explanation.
assert conclusion["sensor"] == "sensor-1"
for proof in saved.explain(conclusion):
    print(proof.rule, proof.fact_ids, dict(proof.bindings))

This example removes the only input fact. The current Engine then has no conclusion, while saved still explains the earlier result. A snapshot keeps its facts and proof records alive for as long as your application retains it. Keep only the snapshots you need, and release references to old snapshots when finished with them. Use del saved to remove that variable; any other references must also be released before Python can reclaim the saved state.

Complete universal evidence

for_all(pattern) checks every fact of that schema. Every fact must satisfy the pattern. A positive universal condition records every witness handle in order; it does not summarize the evidence or keep only the first witness. The complete example below uses two named rules over the same twenty observations. They emit one deduplicated conclusion with two distinct proof edges.

from ting_reasoner import (
    Engine, FactSchema, KnowledgeBase, Rule, emit, eq, for_all,
)

Sample = FactSchema("Sample", number=int, passed=bool)
all_pass = for_all(Sample.pattern(passed=eq(True)))
rules = [
    Rule("inspection", all_pass, [emit("Ready", label="complete")]),
    Rule("review", all_pass, [emit("Ready", label="complete")]),
]
engine = Engine(KnowledgeBase([Sample], rules))
handles = engine.declare(*(Sample(number=i, passed=True) for i in range(20)))
engine.run()

conclusion = engine.conclusions[0]
proofs = engine.explain(conclusion)
expected_handles = tuple(handle.id for handle in handles)
assert len(proofs) == 2
assert all(proof.fact_ids == expected_handles for proof in proofs)
print(len(proofs), len(proofs[0].fact_ids))  # 2 proof edges, 20 witnesses

saved = engine.snapshot()
engine.replace(handles[-1], Sample(number=19, passed=False))
engine.run()
assert engine.conclusions == ()
assert saved.explain(conclusion) == proofs
print(len(saved.explain(conclusion)))  # the saved explanation still has 2 edges

The first output is 2 20: two proof edges, each with all twenty handles. After one observation fails, the current session has no Ready conclusion, while the saved snapshot still explains its earlier result. Rule names and proof edges remain distinct even when their immutable evidence can share storage.

An empty domain satisfies a universal condition. If your application requires at least one observation, combine exists(Sample.pattern()) and the universal condition with all_of(...). A handle can occur more than once when the same fact supports several conditions; this represents its roles, not independent observations.

Large explanations and saved snapshots retain their complete evidence. Ting may share equal immutable support lists through bounded internal caches; reaching a cache limit uses ordinary complete storage. This does not impose a proof-memory or process-RAM limit. MemoryPolicy retains its stated numeric-buffer scope. Release snapshots when your application no longer needs them, and bound the amount of evidence your rules request. Do not rely on Python object identity to compare explanations.

Universal checks and counterexamples

Use for_all(pattern) when every fact of that schema must satisfy a condition. Use not_(for_all(pattern)) when a failing observation should produce a result. A positive universal condition retains every ordered witness; a negated universal condition records the first counterexample in evaluation order.

from ting_reasoner import (
    Engine, FactSchema, KnowledgeBase, Rule, emit, eq, for_all, not_,
)

Sample = FactSchema("Sample", number=int, passed=bool)
rule = Rule("request_review", not_(for_all(Sample.pattern(passed=eq(True)))),
            [emit("ReviewNeeded", label="failed observation")])
engine = Engine(KnowledgeBase([Sample], [rule]))
engine.declare(*(Sample(number=i, passed=True) for i in range(24)))
failed = engine.declare(Sample(number=24, passed=False),
                        Sample(number=25, passed=False))
engine.run()
conclusion = engine.conclusions[0]
assert engine.explain(conclusion)[0].fact_ids == (failed[0].id,)
saved = engine.snapshot()
print(engine.explain(conclusion)[0].fact_ids)  # (25,)

engine.retract(failed[0])
engine.run()
assert engine.explain(engine.conclusions[0])[0].fact_ids == (failed[1].id,)
assert saved.explain(conclusion)[0].fact_ids == (failed[0].id,)
print(engine.explain(engine.conclusions[0])[0].fact_ids)  # (26,)

engine.retract(failed[1])
engine.run()
assert engine.conclusions == ()

The twenty-four passing observations do not support ReviewNeeded. The first failed observation does. Removing it makes the next counterexample support the current result; the saved snapshot retains its earlier explanation. Removing both failures makes the negated universal condition false.

An empty domain satisfies the positive universal condition. Its negation does not fire. If at least one observation is required, combine exists(Sample.pattern()) with the positive condition using all_of(...). To report every failing observation individually, use an ordinary Sample.pattern(passed=eq(False)) rule instead of a negated universal.

Universal conditions check the whole schema domain, including its fixed and derived facts. They do not implicitly select a subgroup. Execution limits can stop an incomplete check; an exhausted budget is not evidence that all observations pass or that a counterexample exists. Inspect RunResult.status before using a result as a completed decision. Complete positive evidence and retained snapshots still require storage proportional to their output; numeric memory limits keep their documented scope.

Keep valid results when inputs change

Before this section: Input changes and explanations.

A result can have several valid reasons. A pump may need inspection because of excessive temperature and independently because of excessive vibration. Clearing one alarm should preserve the result while the other alarm remains.

Pass maintenance=MaintenancePolicy() when creating an Engine to retain grounded results across input changes. A result is grounded when a chain of rule applications reaches current input facts, fixed seeds, or a rule whose conditions need no facts. Circular reasoning alone cannot preserve a result.

Define the rules in readable source

This complete .ting program derives one inspection requirement per device. Several alarm facts can support the same requirement. The final rule emits the application's structured inspection result.

ting source 1
requires semantics 1
module inspection version "1.0.0" role program

schemas:
  schema Alarm:
    device: text required
    cause: text required
  schema InspectionNeeded:
    device: text required

conclusions:
  conclusion Inspect:
    device: text required

rules:
  rule identify_need:
    when:
      match Alarm:
        device: bind device
    then:
      derive InspectionNeeded:
        device: use device
  rule request_inspection:
    when:
      match InspectionNeeded:
        device: bind device
    then:
      conclude Inspect:
        device: use device

tests:
  test independent_causes:
    given:
      fact Alarm:
        device: "pump_a"
        cause: "temperature"
      fact Alarm:
        device: "pump_a"
        cause: "vibration"
    expect:
      conclusion Inspect:
        device: "pump_a"
    firings: 3

Complete Python example

This example includes the source string, so it can run without creating files. compile_source() returns the compiled program and the source information used by built.explain(). The two values returned by engine.declare() are handles identifying the input facts to remove later.

from ting_reasoner import Engine, MaintenancePolicy, compile_source

source = """ting source 1
requires semantics 1
module inspection version "1.0.0" role program

schemas:
  schema Alarm:
    device: text required
    cause: text required
  schema InspectionNeeded:
    device: text required

conclusions:
  conclusion Inspect:
    device: text required

rules:
  rule identify_need:
    when:
      match Alarm:
        device: bind device
    then:
      derive InspectionNeeded:
        device: use device
  rule request_inspection:
    when:
      match InspectionNeeded:
        device: bind device
    then:
      conclude Inspect:
        device: use device

tests:
  test independent_causes:
    given:
      fact Alarm:
        device: "pump_a"
        cause: "temperature"
      fact Alarm:
        device: "pump_a"
        cause: "vibration"
    expect:
      conclusion Inspect:
        device: "pump_a"
    firings: 3
"""
built = compile_source(source)  # Validates the rules and their embedded test.
Alarm = built.schema("Alarm")
engine = Engine(built.program, maintenance=MaintenancePolicy())

# Two facts independently justify the same inspection.
heat, vibration = engine.declare(
    Alarm(device="pump_a", cause="temperature"),
    Alarm(device="pump_a", cause="vibration"),
)
engine.run()
assert len(engine.conclusions) == 1
before = engine.snapshot()  # Keep this if you need the earlier evidence.
previous_conclusion = engine.conclusions[0]

engine.retract(heat)
report = engine.maintenance_report
assert report.mode == "retained"
assert report.retained_conclusions == 1
assert len(engine.conclusions) == 1  # The vibration cause still supports it.
for change in report.changes:
    print(change.status, change.previous_supports, change.remaining_supports)

# run() checks whether more consequences are now possible.
result = engine.run()
assert result.status.value == "quiescent"
assert result.firings == 0  # Existing valid derivations did not fire again.
assert built.explain(engine, engine.conclusions[0]).proofs

engine.retract(vibration)
assert engine.conclusions == ()  # The final grounding cause disappeared.
assert engine.maintenance_report.withdrawn_conclusions == 1
engine.run()

# A saved snapshot still explains the earlier result after withdrawal.
assert built.explain(before, previous_conclusion).proofs

After the first removal, the derived inspection requirement changes from two direct supports to one. Its conclusion still has one direct support: the inspection requirement itself. Direct support counts describe proof edges; they are not confidence percentages and do not establish that observations are independent.

Use a source file in your chosen folder

from pathlib import Path
from ting_reasoner import Engine, MaintenancePolicy, compile_file, load_tingir

project_dir = Path.home() / "ting_examples"
source_dir = project_dir / "editable"
compiled_dir = project_dir / "compiled"
source_dir.mkdir(parents=True, exist_ok=True)
compiled_dir.mkdir(parents=True, exist_ok=True)
source_path = source_dir / "inspection_supports.ting"
compiled_path = compiled_dir / "inspection_supports.tingir"
print("Save the complete .ting source above at:", source_path)

# After saving that text as UTF-8 in your editor:
built = compile_file(source_path, compiled_path)
engine = Engine(load_tingir(compiled_path), maintenance=MaintenancePolicy())
Alarm = built.schema("Alarm")

Save the complete source above at the printed path before running the compilation line. Use the file-path instructions for other folders, and overwrite=True only to replace an existing compiled file. See Handle build errors for compilation failures.

Support maintenance is a Python session option. The rules remain ordinary portable .ting knowledge, and the same compiled program can be used with or without this option. For a selectively linked library, use Engine(linked.program, maintenance=MaintenancePolicy()). Inheritance rules for classifications are eligible too.

What happens during an update?

Retention uses only rule applications already recorded. If you remove an old cause and insert a different cause in the same transaction, an old result may disappear until run() proves the new cause sufficient. A budget-limited prior run may also have left alternative supports undiscovered.

A retained derived fact keeps its negative handle while it remains a derived row. Explicitly declaring that value makes it a base fact with a positive handle, and supporting references are updated. Removing a base assertion that is still justified by rules creates a derived handle. Reacquire handles when facts change roles.

When a conclusion's original representative proof disappears, Ting chooses a remaining proof and updates its rule and fact_ids. Reacquire the current Conclusion from engine.conclusions. Its name and values can remain the same while that evidence changes. Save an EngineSnapshot beforehand to inspect earlier evidence later.

Read the update report

engine.maintenance_report is an immutable report for the latest committed input update. Duplicate declarations, missing-handle retractions, unchanged replacements and failed updates leave the previous report intact. With maintenance disabled it is None.

The report describes the committed update, before the next run. A recompute or reset report contains summary counts and no individual change records. If a fact loses its rule proofs but remains explicitly asserted as a base fact, its status is retained with zero remaining rule supports.

Supported conditions and recomputation

Retention applies when every loaded rule branch uses positive match conditions and optional test comparisons. Positive AND and OR branches, repeated patterns, recursion, fixed seeds and classification inheritance are supported. Cycles without grounding cannot keep one another alive after their last grounding reason disappears.

If the loaded program contains any negated condition, exists or for_all, input updates use the normal recomputation path: the new inputs are committed, inferred state is cleared, and run() rebuilds it. The report identifies this as unsupported_conditions. This applies to the whole loaded program. Omitting maintenance or setting enabled=False uses the usual recomputation behavior throughout.

Set limits and handle an update failure

from ting_reasoner import (
    Engine, FactSchema, KnowledgeBase, MaintenanceLimitError,
    MaintenancePolicy, Rule, bind, emit, use,
)

Reading = FactSchema("reading", sensor=str)
knowledge = KnowledgeBase(rules=(
    Rule("record", (Reading.pattern(sensor=bind("sensor")),),
         (emit("observed", sensor=use("sensor")),)),
))
# Zero proofs is deliberately tiny, to demonstrate an update limit error.
policy = MaintenancePolicy(max_proofs=0, on_limit="raise")
engine = Engine(knowledge, maintenance=policy)
handle = engine.declare(Reading(sensor="north"))
engine.run()
before = engine.snapshot()

try:
    engine.retract(handle)
except MaintenanceLimitError as error:
    assert error.code == "TING-MAINT-001"
    assert error.limit == "max_proofs"
    assert engine.snapshot() == before  # The input and prior result survive.
    print(error.as_dict())
else:
    raise AssertionError("Expected the deliberately low limit to reject this update")

# The default on_limit="recompute" commits the base update and clears inferred
# state when retention exceeds a limit. run() then rebuilds the consequences.
fallback = Engine(knowledge, maintenance=MaintenancePolicy(max_proofs=0))
first = fallback.declare(Reading(sensor="north"))
fallback.run()
fallback.declare(Reading(sensor="south"))
assert fallback.maintenance_report.mode == "recompute"
assert fallback.maintenance_report.reason == "limit:max_proofs"
assert not fallback.conclusions
fallback.run()
assert len(fallback.conclusions) == 2

Limits are non-negative exact integers; Boolean values are rejected. They bound logical work and records, not exact Python heap bytes or elapsed time. Wrong argument types raise TypeError or ValueError. MaintenanceLimitError exposes code, limit, maximum, used and a JSON-safe as_dict(). Catch it around declare, retract, replace or transaction commit. A rejected transaction requires a new transaction to retry.

With on_limit="recompute", the input update succeeds using the established recomputation behavior. Check the report when your application needs to know which path ran. on_limit="raise" concerns resource limits; programs with unsupported conditions still use recomputation. Allocation or other staging failures propagate before publication, leaving the prior state usable.

Run after changing inputs

Keep the same Engine while updating facts for one case. Group related changes in a transaction, then call engine.run() after the transaction closes. Check result.status.value == "quiescent" before treating inference as complete. For "firing_limit", raise the firing allowance; for "budget_exhausted", inspect result.exhausted_budget. See Work limits for retry behavior.

RunResult counts work performed by that call. A call may report zero new firings while valid conclusions remain in engine.conclusions. Read engine.maintenance_report to see what the input update retained or withdrew. Use run(budget=ExecutionBudget(...)) when your application needs explicit limits on candidate rows, constraint checks, activations and firings.

An input update that changes the committed inputs clears engine.trace; current proof records still explain retained results. Save an EngineSnapshot before an update if you need to inspect the previous state. Reacquire conclusions and fact handles after an update, especially when a fact changes between an asserted input and a derived result.

Use run_batch() for independent cases. Each case starts a fresh session. To retain results as the inputs of one case change, keep its Engine and enable maintenance=MaintenancePolicy().

Make and inspect a decision

Before this section: Run and inspect and Write Ting rules.

Use decide() when your application needs a structured answer about a finite set of options. You supply a compiled program, a DecisionRequest, and the case's input facts. Ting evaluates the rules in a fresh isolated session, checks the evidence for every required condition, and returns a DecisionResult.

The selection policy is unique-feasible-v1: select an option only when it is the single feasible candidate and every other candidate is conclusively infeasible. All requested conditions are mandatory. Two feasible options remain a tie; their order does not choose a winner. The program supplies evidence through ordinary portable conclusion declarations.

Define the evidence in a .ting program

This example chooses equipment with capacity of at least 10 units and a passed inspection. The complete Python example below includes the source text directly and needs no source file. If you prefer an editable file, first choose its full path:

from pathlib import Path

source_path = Path.home() / "ting_examples" / "editable" / "equipment_decision.ting"
source_path.parent.mkdir(parents=True, exist_ok=True)
print("Save the following UTF-8 text at:", source_path)

Replace the folder components with your preferred location if needed. This creates the folder and prints the filename; it does not write the source text. Use your editor’s Save As command to save the following complete text at that printed path.

ting source 1
requires semantics 1
module equipment.choice version "1.0.0" role program

schemas:
  schema Capacity:
    device: text required
    available: integer required
  schema Inspection:
    device: text required
    passed: boolean required

conclusions:
  conclusion Evidence:
    goal: text required
    candidate: text required
    requirement: text required
    satisfied: boolean required
    reason: text required

rules:
  rule capacity_sufficient:
    when:
      match Capacity:
        device: bind device
        available: ge 10
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "capacity"
        satisfied: true
        reason: "CAPACITY_AT_LEAST_10"
  rule capacity_insufficient:
    when:
      match Capacity:
        device: bind device
        available: lt 10
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "capacity"
        satisfied: false
        reason: "CAPACITY_BELOW_10"
  rule inspection_passed:
    when:
      match Inspection:
        device: bind device
        passed: eq true
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "inspection"
        satisfied: true
        reason: "INSPECTION_PASSED"
  rule inspection_failed:
    when:
      match Inspection:
        device: bind device
        passed: eq false
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "inspection"
        satisfied: false
        reason: "INSPECTION_FAILED"

tests:
  test sufficient_capacity:
    given:
      fact Capacity:
        device: "pump_a"
        available: 15
    expect:
      conclusion Evidence:
        goal: "choose_equipment"
        candidate: "pump_a"
        requirement: "capacity"
        satisfied: true
        reason: "CAPACITY_AT_LEAST_10"
    firings: 1

The five fields of Evidence form the decision evidence contract. A conclusion with satisfied: true supports one requirement; false explicitly opposes it. An absent conclusion does neither. reason is an author-written code that your application can display or map to its own labels.

The designated evidence conclusion must have exactly these fields. Other conclusion names are ignored. Well-formed evidence for a different goal is ignored. Evidence for this goal with an unknown candidate or requirement raises DecisionError, making spelling mistakes visible. Every conclusion on the designated channel is checked for the correct shape, even if it names a different goal.

Complete Python example

This example runs without creating a source file. compile_source() validates the declarations and their embedded test; built.schema() retrieves the fact constructors. A tuple such as ("capacity", "inspection") lists the names in order. A one-item tuple needs a trailing comma, for example ("capacity",).

from ting_reasoner import (
    DecisionError, DecisionLimits, DecisionOutcome, DecisionRequest,
    DecisionStatus, compile_source, decide,
)

source = """ting source 1
requires semantics 1
module equipment.choice version "1.0.0" role program

schemas:
  schema Capacity:
    device: text required
    available: integer required
  schema Inspection:
    device: text required
    passed: boolean required

conclusions:
  conclusion Evidence:
    goal: text required
    candidate: text required
    requirement: text required
    satisfied: boolean required
    reason: text required

rules:
  rule capacity_sufficient:
    when:
      match Capacity:
        device: bind device
        available: ge 10
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "capacity"
        satisfied: true
        reason: "CAPACITY_AT_LEAST_10"
  rule capacity_insufficient:
    when:
      match Capacity:
        device: bind device
        available: lt 10
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "capacity"
        satisfied: false
        reason: "CAPACITY_BELOW_10"
  rule inspection_passed:
    when:
      match Inspection:
        device: bind device
        passed: eq true
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "inspection"
        satisfied: true
        reason: "INSPECTION_PASSED"
  rule inspection_failed:
    when:
      match Inspection:
        device: bind device
        passed: eq false
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "inspection"
        satisfied: false
        reason: "INSPECTION_FAILED"

tests:
  test sufficient_capacity:
    given:
      fact Capacity:
        device: "pump_a"
        available: 15
    expect:
      conclusion Evidence:
        goal: "choose_equipment"
        candidate: "pump_a"
        requirement: "capacity"
        satisfied: true
        reason: "CAPACITY_AT_LEAST_10"
    firings: 1
"""
built = compile_source(source)
Capacity = built.schema("Capacity")
Inspection = built.schema("Inspection")
request = DecisionRequest(
    goal="choose_equipment",
    candidates=("pump_a", "pump_b"),
    requirements=("capacity", "inspection"),
    evidence_name="equipment.choice.Evidence",
    closed=True,
)

facts = (
    Capacity(device="pump_a", available=15),
    Inspection(device="pump_a", passed=True),
    Capacity(device="pump_b", available=5),
)
result = decide(built.program, request, facts=facts)
assert result.status is DecisionStatus.COMPLETE
assert result.outcome is DecisionOutcome.SELECTED
assert result.selected_option == "pump_a"
print(result.selected_option)

# Inspect the rule evidence behind pump_a's capacity assessment.
capacity_check = result.candidates[0].constraints[0]
evidence = capacity_check.support[0]
print(evidence["reason"])  # CAPACITY_AT_LEAST_10
explanation = built.explain(result.snapshot, evidence)
assert explanation.proofs

# Export the structured summary for your application.
import json
print(json.dumps(result.as_dict(), indent=2))

The result selects pump_a. Pump_b is excluded by its known capacity failure, so its missing inspection cannot rescue it or block pump_a. Every condition remains visible in the candidate records, including that unknown inspection.

Understand the request

The request is immutable and accepts at most 4,096 candidates and 256 requirements. An empty closed candidate list yields NO_FEASIBLE_OPTION within that empty domain. With no requirements, each listed candidate is feasible by definition. Use nonempty requirements whenever your application intends to impose conditions.

Your .ting rules contain the domain knowledge and emit evidence. You can construct the request in Python for each case, or store a named decision declaration in the .ting source and retrieve it from the compiled file. Both use the same evidence and outcome contract. Use Store and import named decisions when the request should travel with the saved program.

Read the result before acting

For each requirement, SUPPORTED means positive evidence only, REFUTED means negative evidence only, UNKNOWN means neither, and CONFLICTED means both. A candidate with any REFUTED hard requirement is INFEASIBLE. Otherwise a conflict takes precedence over missing evidence; all other candidates are UNKNOWN or FEASIBLE. A conflict alone does not authorize exclusion. No support count becomes a confidence percentage, and duplicate derivations do not constitute voting.

The overall outcome reports CONFLICTED before NEEDS_DATA when both problems affect viable candidates; reason_codes and condition records preserve the additional blockers. Missing or conflicting evidence on an independently excluded candidate does not block the remaining choice.

Handle missing data, conflicts, ties and incomplete runs

from ting_reasoner import (
    DecisionError, DecisionLimits, DecisionOutcome, DecisionRequest,
    DecisionStatus, compile_source, decide,
)

source = """ting source 1
requires semantics 1
module equipment.choice version "1.0.0" role program

schemas:
  schema Capacity:
    device: text required
    available: integer required
  schema Inspection:
    device: text required
    passed: boolean required

conclusions:
  conclusion Evidence:
    goal: text required
    candidate: text required
    requirement: text required
    satisfied: boolean required
    reason: text required

rules:
  rule capacity_sufficient:
    when:
      match Capacity:
        device: bind device
        available: ge 10
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "capacity"
        satisfied: true
        reason: "CAPACITY_AT_LEAST_10"
  rule capacity_insufficient:
    when:
      match Capacity:
        device: bind device
        available: lt 10
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "capacity"
        satisfied: false
        reason: "CAPACITY_BELOW_10"
  rule inspection_passed:
    when:
      match Inspection:
        device: bind device
        passed: eq true
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "inspection"
        satisfied: true
        reason: "INSPECTION_PASSED"
  rule inspection_failed:
    when:
      match Inspection:
        device: bind device
        passed: eq false
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "inspection"
        satisfied: false
        reason: "INSPECTION_FAILED"

tests:
  test sufficient_capacity:
    given:
      fact Capacity:
        device: "pump_a"
        available: 15
    expect:
      conclusion Evidence:
        goal: "choose_equipment"
        candidate: "pump_a"
        requirement: "capacity"
        satisfied: true
        reason: "CAPACITY_AT_LEAST_10"
    firings: 1
"""
built = compile_source(source)
Capacity = built.schema("Capacity")
Inspection = built.schema("Inspection")
request = DecisionRequest(
    goal="choose_equipment",
    candidates=("pump_a", "pump_b"),
    requirements=("capacity", "inspection"),
    evidence_name="equipment.choice.Evidence",
    closed=True,
)

capacity_a = Capacity(device="pump_a", available=15)
capacity_b = Capacity(device="pump_b", available=5)

# Missing inspection is unknown: it is not an automatic failure.
missing = decide(built.program, request, facts=(capacity_a, capacity_b))
assert missing.status is DecisionStatus.COMPLETE
assert missing.outcome is DecisionOutcome.NEEDS_DATA
assert missing.selected_option is None

# Opposing reports for the same requirement remain visible.
conflict = decide(built.program, request, facts=(
    capacity_a, capacity_b,
    Inspection(device="pump_a", passed=True),
    Inspection(device="pump_a", passed=False),
))
assert conflict.outcome is DecisionOutcome.CONFLICTED
check = conflict.candidates[0].constraints[1]
assert check.support and check.opposition

# If both pumps qualify, Ting preserves the tie.
tie = decide(built.program, request, facts=(
    capacity_a, Capacity(device="pump_b", available=12),
    Inspection(device="pump_a", passed=True),
    Inspection(device="pump_b", passed=True),
))
assert tie.outcome is DecisionOutcome.MULTIPLE_VALID_OPTIONS
assert tie.feasible_options == ("pump_a", "pump_b")
assert tie.selected_option is None

# A firing limit is incomplete computation, not missing information.
limited = decide(built.program, request, facts=(capacity_a, capacity_b), max_firings=0)
assert limited.status is DecisionStatus.BUDGET_EXHAUSTED
assert limited.outcome is DecisionOutcome.UNDETERMINED
assert limited.candidates == ()
print(limited.exhausted_limit)

COMPLETE with NEEDS_DATA is valid: Ting finished evaluating the information supplied, and determined that it is insufficient. A computation limit instead returns BUDGET_EXHAUSTED with UNDETERMINED. Never infer an application action from selected_option alone without handling the other outcomes your workflow permits.

Choose source and compiled file paths

from pathlib import Path
from ting_reasoner import DecisionRequest, compile_file, decide, load_tingir

project_dir = Path.home() / "ting_examples"
source_dir = project_dir / "editable"
compiled_dir = project_dir / "compiled"
source_dir.mkdir(parents=True, exist_ok=True)
compiled_dir.mkdir(parents=True, exist_ok=True)
source_path = source_dir / "equipment_decision.ting"
compiled_path = compiled_dir / "equipment_decision.tingir"
print("Save the complete .ting text above at:", source_path)

# After saving that UTF-8 file in your editor:
built = compile_file(source_path, compiled_path)
program = load_tingir(compiled_path)
request = DecisionRequest("choose_equipment", ("pump_a", "pump_b"),
    ("capacity", "inspection"), "equipment.choice.Evidence", closed=True)
Capacity, Inspection = built.schema("Capacity"), built.schema("Inspection")
result = decide(program, request, facts=(
    Capacity(device="pump_a", available=15),
    Inspection(device="pump_a", passed=True),
    Capacity(device="pump_b", available=5),
))
print(result.outcome.value, result.selected_option)

This continuation requires the evidence source saved earlier in this section. Use the file-path instructions to choose another folder, and overwrite=True only when replacing an existing compiled file. Keep the build result for schema helpers and source explanations.

Set limits and handle errors

from ting_reasoner import (
    DecisionError, DecisionLimits, DecisionOutcome, DecisionRequest,
    DecisionStatus, compile_source, decide,
)

source = """ting source 1
requires semantics 1
module equipment.choice version "1.0.0" role program

schemas:
  schema Capacity:
    device: text required
    available: integer required
  schema Inspection:
    device: text required
    passed: boolean required

conclusions:
  conclusion Evidence:
    goal: text required
    candidate: text required
    requirement: text required
    satisfied: boolean required
    reason: text required

rules:
  rule capacity_sufficient:
    when:
      match Capacity:
        device: bind device
        available: ge 10
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "capacity"
        satisfied: true
        reason: "CAPACITY_AT_LEAST_10"
  rule capacity_insufficient:
    when:
      match Capacity:
        device: bind device
        available: lt 10
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "capacity"
        satisfied: false
        reason: "CAPACITY_BELOW_10"
  rule inspection_passed:
    when:
      match Inspection:
        device: bind device
        passed: eq true
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "inspection"
        satisfied: true
        reason: "INSPECTION_PASSED"
  rule inspection_failed:
    when:
      match Inspection:
        device: bind device
        passed: eq false
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "inspection"
        satisfied: false
        reason: "INSPECTION_FAILED"

tests:
  test sufficient_capacity:
    given:
      fact Capacity:
        device: "pump_a"
        available: 15
    expect:
      conclusion Evidence:
        goal: "choose_equipment"
        candidate: "pump_a"
        requirement: "capacity"
        satisfied: true
        reason: "CAPACITY_AT_LEAST_10"
    firings: 1
"""
built = compile_source(source)
Capacity = built.schema("Capacity")
Inspection = built.schema("Inspection")
request = DecisionRequest(
    goal="choose_equipment",
    candidates=("pump_a", "pump_b"),
    requirements=("capacity", "inspection"),
    evidence_name="equipment.choice.Evidence",
    closed=True,
)

# A typo must not silently turn an observation into missing evidence.
try:
    decide(built.program, request, facts=(Capacity(device="pump_typo", available=15),))
except DecisionError as error:
    assert error.code == "TING-DEC-003"
    print(error.as_dict())
else:
    raise AssertionError("Expected an out-of-domain evidence error")

# Limit the assessment separately from the inference run.
limited = decide(built.program, request,
    facts=(Capacity(device="pump_a", available=15),),
    limits=DecisionLimits(max_work=0))
assert limited.status is DecisionStatus.BUDGET_EXHAUSTED
assert limited.exhausted_limit == "max_work"
assert limited.selected_option is None

Limits accept nonnegative exact integers; Boolean values are rejected. They bound assessment records and logical work, not exact heap usage or elapsed time. DecisionLimits applies after inference except for request and case-input checks. To bound inference itself, use budget=ExecutionBudget(...) or max_firings=..., never both. With neither argument, decide() allows at most 10,000 firings. An explicit ExecutionBudget uses its own ceilings; unspecified fields are unbounded.

An incomplete decision has empty candidates and feasible_options and selected_option=None; partial assessments are not published as final conclusions. Its snapshot can contain partial inference evidence. Retry decide() with adequate limits to start a fresh evaluation. assess_decision() can reassess an already completed batch session with a larger assessment allowance without rerunning inference.

Source, compiled-file and engine validation errors retain their existing exception types. Catch SourceError around compilation and DecisionError around request creation or assessment as appropriate. Unexpected execution faults propagate as exceptions; they are never converted into NO_FEASIBLE_OPTION. decide() owns a new session, so an error cannot alter a separate live Engine.

Evaluate independent cases with multiprocessing

decide() uses the current process. For multiple independent cases, run the program through run_batch() or iter_batch(), then pass each SessionResult to assess_decision(). The inference work can run in workers; the parent performs the deterministic evidence assessment.

from ting_reasoner import (
    DecisionRequest, ExecutionPolicy, InferenceJob,
    assess_decision, compile_source, run_batch,
)

source = """ting source 1
requires semantics 1
module equipment.choice version "1.0.0" role program

schemas:
  schema Capacity:
    device: text required
    available: integer required
  schema Inspection:
    device: text required
    passed: boolean required

conclusions:
  conclusion Evidence:
    goal: text required
    candidate: text required
    requirement: text required
    satisfied: boolean required
    reason: text required

rules:
  rule capacity_sufficient:
    when:
      match Capacity:
        device: bind device
        available: ge 10
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "capacity"
        satisfied: true
        reason: "CAPACITY_AT_LEAST_10"
  rule capacity_insufficient:
    when:
      match Capacity:
        device: bind device
        available: lt 10
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "capacity"
        satisfied: false
        reason: "CAPACITY_BELOW_10"
  rule inspection_passed:
    when:
      match Inspection:
        device: bind device
        passed: eq true
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "inspection"
        satisfied: true
        reason: "INSPECTION_PASSED"
  rule inspection_failed:
    when:
      match Inspection:
        device: bind device
        passed: eq false
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "inspection"
        satisfied: false
        reason: "INSPECTION_FAILED"

tests:
  test sufficient_capacity:
    given:
      fact Capacity:
        device: "pump_a"
        available: 15
    expect:
      conclusion Evidence:
        goal: "choose_equipment"
        candidate: "pump_a"
        requirement: "capacity"
        satisfied: true
        reason: "CAPACITY_AT_LEAST_10"
    firings: 1
"""

def main():
    built = compile_source(source)
    Capacity = built.schema("Capacity")
    Inspection = built.schema("Inspection")
    request = DecisionRequest("choose_equipment", ("pump_a", "pump_b"),
        ("capacity", "inspection"), "equipment.choice.Evidence", closed=True)
    jobs = (
        InferenceJob((Capacity(device="pump_a", available=15),
                      Inspection(device="pump_a", passed=True),
                      Capacity(device="pump_b", available=5))),
        InferenceJob((Capacity(device="pump_a", available=15),
                      Capacity(device="pump_b", available=5))),
    )
    batch = run_batch(built.program, jobs,
        policy=ExecutionPolicy(mode="owned", max_workers=2, minimum_parallel_facts=0))
    results = tuple(assess_decision(session, request) for session in batch.results)
    assert results[0].selected_option == "pump_a"
    assert results[1].outcome.value == "NEEDS_DATA"
    for result in results:
        print(result.outcome.value, result.selected_option)

if __name__ == "__main__":
    main()

Save the example as a .py file and run it from a terminal. The main guard supports spawned workers on Windows, Linux and macOS. The example sets minimum_parallel_facts=0 only to demonstrate workers on tiny inputs; process startup can outweigh their benefit. Use measured thresholds for your workloads. Each result is bound to its own snapshot. For different candidate lists, use the corresponding request when assessing each session.

Save and explain a decision result

Keep the returned result and its snapshot when your application needs to show why an option passed or failed. See Explain a conclusion for reading the rule evidence and Reuse a compiled program for evaluating further cases.

Support and opposition contain actual Conclusion records, so all underlying rule proofs remain available through the snapshot. as_dict() includes each evidence record’s index in snapshot.conclusions. Keep the snapshot if you need to resolve those references later. The request digest identifies request contents and order; program_digest identifies the compiled program. snapshot_epoch and session_sequence are local labels, not globally unique snapshot identities. The summary is not an independent verification certificate.

This policy evaluates hard requirements and preserves ties. It does not rank preferences, resolve authority conflicts, expire observations, construct candidate domains, or search for counterfactual changes. Domain closure and the relevance of input observations remain explicit application responsibilities.

Store a decision in a compiled file

A named decision keeps its candidate list, hard requirements, evidence channel and selection policy with the reasoning program. Write it in a readable .ting file, compile that file, then retrieve the request from the resulting .tingir file. Your application supplies each case’s facts and calls decide(). Loading a request does not run rules or choose an option.

This section uses the equipment rules from Make and inspect a decision. The complete examples include their source text so you can run each example on its own. See that section for outcome handling, inference limits and evidence inspection.

Where the files go

Save the first complete Python example as named_decision.py and run it with python named_decision.py (or py named_decision.py on Windows). Path.cwd() is the folder your terminal is currently using. The example creates a ting_projects/named equipment folder inside it and prints the full source and compiled paths.

To choose another folder, replace the folder = ... line with folder = Path.home() / "Ting Files", or use a path from the file-path instructions.

mkdir(parents=True, exist_ok=True) creates missing folders. write_text(..., encoding="utf-8") creates the readable source file, replacing that example file if it already exists. compile_file() creates the compiled file after validation and embedded tests succeed. It does not create missing parent folders. The example passes overwrite=True so it can replace its previous compiled output; omit this argument when existing compiled files should be protected.

Declare the decision in .ting source

Add requires feature decisions immediately after the module header. Add this decisions: region alongside schemas:, conclusions: and rules:. Use two spaces for each indentation level. In source, Boolean literals are lowercase true and false; Python uses True and False.

decisions:
  decision ChooseEquipment:
    goal: "choose_equipment"
    candidates: ["pump_a", "pump_b"]
    requirements: ["capacity", "inspection"]
    evidence: Evidence
    closed: true
    policy: "unique-feasible-v1"
    requires:
      rule capacity_sufficient
      rule capacity_insufficient
      rule inspection_passed
      rule inspection_failed

Every field except logical_time is mandatory. Candidate and requirement identifiers follow the DecisionRequest text limits: nonempty valid Unicode, at most 256 UTF-8 bytes and no ASCII control characters. Empty lists are permitted: an empty closed candidate universe is infeasible within that empty universe; an empty requirement list imposes no conditions. Use these cases only when they describe your intended scope.

The compiler checks that dependency names exist. You must list the rules and fixed facts needed for the intended decision, including producer rules declared in a rule’s own requires: block. Ting does not discover missing domain knowledge or assume that all rules in a folder should run. Schema/classification definitions required by the selected rules are included automatically. Session observations still come from your application.

Complete example: save, compile, load and decide

from pathlib import Path
from ting_reasoner import (
    CompiledCatalog, DecisionOutcome, Engine, SourceError,
    compile_file, compile_source, decide, load_tingir,
)

source = """ting source 1
requires semantics 1
module equipment.choice version "1.0.0" role program
requires feature decisions

schemas:
  schema Capacity:
    device: text required
    available: integer required
  schema Inspection:
    device: text required
    passed: boolean required

conclusions:
  conclusion Evidence:
    goal: text required
    candidate: text required
    requirement: text required
    satisfied: boolean required
    reason: text required

rules:
  rule capacity_sufficient:
    when:
      match Capacity:
        device: bind device
        available: ge 10
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "capacity"
        satisfied: true
        reason: "CAPACITY_AT_LEAST_10"
  rule capacity_insufficient:
    when:
      match Capacity:
        device: bind device
        available: lt 10
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "capacity"
        satisfied: false
        reason: "CAPACITY_BELOW_10"
  rule inspection_passed:
    when:
      match Inspection:
        device: bind device
        passed: eq true
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "inspection"
        satisfied: true
        reason: "INSPECTION_PASSED"
  rule inspection_failed:
    when:
      match Inspection:
        device: bind device
        passed: eq false
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "inspection"
        satisfied: false
        reason: "INSPECTION_FAILED"

tests:
  test sufficient_capacity:
    given:
      fact Capacity:
        device: "pump_a"
        available: 15
    expect:
      conclusion Evidence:
        goal: "choose_equipment"
        candidate: "pump_a"
        requirement: "capacity"
        satisfied: true
        reason: "CAPACITY_AT_LEAST_10"
    firings: 1

decisions:
  decision ChooseEquipment:
    goal: "choose_equipment"
    candidates: ["pump_a", "pump_b"]
    requirements: ["capacity", "inspection"]
    evidence: Evidence
    closed: true
    policy: "unique-feasible-v1"
    requires:
      rule capacity_sufficient
      rule capacity_insufficient
      rule inspection_passed
      rule inspection_failed
"""

# Change this one folder if you want to store the files elsewhere.
folder = Path.cwd() / "ting_projects" / "named equipment"
folder.mkdir(parents=True, exist_ok=True)
source_path = folder / "equipment.ting"
compiled_path = folder / "equipment.tingir"
source_path.write_text(source, encoding="utf-8")

# overwrite=True allows this example to be run again in the same folder.
built = compile_file(source_path, compiled_path, overwrite=True)
assert built.decision("ChooseEquipment").closed is True
print("Source:", source_path)
print("Compiled:", built.output)

# Everything below also works in a separate Python program.
program = load_tingir(compiled_path)
print(program.decision_names)
request = program.decision("equipment.choice.ChooseEquipment")

# Use the loaded program's schemas to construct correctly typed input facts.
schemas = Engine(program)
Capacity = schemas.schema("equipment.choice.Capacity")
Inspection = schemas.schema("equipment.choice.Inspection")
result = decide(program, request, facts=(
    Capacity(device="pump_a", available=15),
    Inspection(device="pump_a", passed=True),
    Capacity(device="pump_b", available=5),
))
assert result.outcome is DecisionOutcome.SELECTED
assert result.selected_option == "pump_a"
print(result.outcome.value, result.selected_option)

The final line prints SELECTED pump_a. pump_a passes both requirements; pump_b fails capacity, so its absent inspection does not restore its eligibility. The returned value is a DecisionResult, with the same outcome and error behavior as a request constructed directly in Python.

built.decision("ChooseEquipment") accepts names visible in the source, including import aliases. A loaded CompactIR uses exact qualified names: program.decision("equipment.choice.ChooseEquipment"). Its decision_names property lists the available names. Neither method changes the program or runs inference.

To run on another computer, copy the compiled program, install a Ting version that supports its format, and set compiled_path to the file’s actual location on that computer. The loading-and-running part of the example needs no source file or compiler call. The inspection Engine supplies the schemas; decide() owns its separate reasoning session. Both reuse the same verified program.

Import one decision from a compiled library

Use role library in the source header and export decision ChooseEquipment. Other declarations can stay private. CompiledCatalog.link(decisions=("equipment.choice.ChooseEquipment",)) includes that request, its evidence declaration, and the rule/seed dependencies you listed. The comma creates a one-element Python tuple. It does not select unrelated exported decisions or every rule in the library.

from pathlib import Path
from ting_reasoner import (
    CompiledCatalog, DecisionOutcome, Engine, SourceError,
    compile_file, compile_source, decide, load_tingir,
)

source = """ting source 1
requires semantics 1
module equipment.choice version "1.0.0" role program
requires feature decisions

schemas:
  schema Capacity:
    device: text required
    available: integer required
  schema Inspection:
    device: text required
    passed: boolean required

conclusions:
  conclusion Evidence:
    goal: text required
    candidate: text required
    requirement: text required
    satisfied: boolean required
    reason: text required

rules:
  rule capacity_sufficient:
    when:
      match Capacity:
        device: bind device
        available: ge 10
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "capacity"
        satisfied: true
        reason: "CAPACITY_AT_LEAST_10"
  rule capacity_insufficient:
    when:
      match Capacity:
        device: bind device
        available: lt 10
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "capacity"
        satisfied: false
        reason: "CAPACITY_BELOW_10"
  rule inspection_passed:
    when:
      match Inspection:
        device: bind device
        passed: eq true
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "inspection"
        satisfied: true
        reason: "INSPECTION_PASSED"
  rule inspection_failed:
    when:
      match Inspection:
        device: bind device
        passed: eq false
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "inspection"
        satisfied: false
        reason: "INSPECTION_FAILED"

tests:
  test sufficient_capacity:
    given:
      fact Capacity:
        device: "pump_a"
        available: 15
    expect:
      conclusion Evidence:
        goal: "choose_equipment"
        candidate: "pump_a"
        requirement: "capacity"
        satisfied: true
        reason: "CAPACITY_AT_LEAST_10"
    firings: 1

decisions:
  decision ChooseEquipment:
    goal: "choose_equipment"
    candidates: ["pump_a", "pump_b"]
    requirements: ["capacity", "inspection"]
    evidence: Evidence
    closed: true
    policy: "unique-feasible-v1"
    requires:
      rule capacity_sufficient
      rule capacity_insufficient
      rule inspection_passed
      rule inspection_failed
"""

# Two compiled libraries can live in different folders.
root = Path.cwd() / "ting_projects" / "decision libraries"
primary_folder = root / "primary rules"
backup_folder = root / "backup rules"
for folder in (primary_folder, backup_folder):
    folder.mkdir(parents=True, exist_ok=True)

# The header and export determine the library's contents and public entry.
library_source = source.replace("role program", "role library")
library_source += "\nexports:\n  decision ChooseEquipment\n"
primary_source = primary_folder / "equipment.ting"
primary_source.write_text(library_source, encoding="utf-8")
backup_source = backup_folder / "equipment.ting"
backup_source.write_text(library_source.replace("equipment.choice", "equipment.backup"), encoding="utf-8")
primary = compile_file(primary_source, overwrite=True)
backup = compile_file(backup_source, overwrite=True)

catalog = CompiledCatalog({
    "equipment.choice": primary.output,
    "equipment.backup": backup.output,
})
linked = catalog.link(decisions=("equipment.choice.ChooseEquipment",))
request = linked.decision("equipment.choice.ChooseEquipment")
assert linked.program.decision_names == ("equipment.choice.ChooseEquipment",)
assert linked.usage.modules == 1  # The backup library was not opened.

Capacity = linked.schema("equipment.choice.Capacity")
Inspection = linked.schema("equipment.choice.Inspection")
result = decide(linked.program, request, facts=(
    Capacity(device="pump_a", available=15),
    Inspection(device="pump_a", passed=True),
    Capacity(device="pump_b", available=5),
))
assert result.selected_option == "pump_a"
assert len(linked.knowledge_base.rules) == 4
print(linked.imports)  # Includes the decision's private required rules.

# Retain linked if your application needs the compiled source explanations.
evidence = result.candidates[0].constraints[0].support[0]
assert linked.explain(result.snapshot, evidence).proofs

A catalog maps each exact module name to its compiled file path. You can supply files from several folders and select several decisions in one link call. Every external dependency also needs a catalog entry matching its compiled version and contract. The linker does not scan folders automatically. linked.imports records the selected declarations and dependencies; linked.usage reports the sections and bytes read.

namespaces=("equipment.choice",) selects every public export in that module, including decisions and their dependencies. Selecting one decision reads the container directory and the selected sections; unrelated payloads are not verified by that operation. Use CompiledLibrary.verify_all() when you need to check every payload. Keep linked for source-linked explanations, or save linked.program with save_tingir() for a standalone executable file. The standalone program retains requests and rules, while the separate linked source map remains with linked.

Retrieving a request from a whole program does not filter that program’s rules. Compile and link a selected library subset when rule isolation is required. All conclusions matching the request’s goal and evidence channel must use candidate/requirement names from that request. Different decisions can use separate goals or evidence channels to keep their evidence domains distinct.

Import a source declaration

Source imports support the same named entry and an optional local alias:

ting source 1
requires semantics 1
module application version "1.0.0" role program
requires feature decisions
imports:
  from equipment.choice version "1.0.0":
    decision ChooseEquipment as EquipmentChoice

Pass modules={"equipment.choice": provider_source_path} to compile_file(), where provider_source_path is the actual library .ting path. The build then supports built.decision("EquipmentChoice"); after loading its compiled file, use program.decision("equipment.choice.ChooseEquipment"). An alias does not rename the portable declaration. With in-memory text, use compile_source(..., resolver=MemoryResolver({...})) as shown in Source modules.

Handle declaration, file and lookup errors

from pathlib import Path
from ting_reasoner import (
    CompiledCatalog, DecisionOutcome, Engine, SourceError,
    compile_file, compile_source, decide, load_tingir,
)

source = """ting source 1
requires semantics 1
module equipment.choice version "1.0.0" role program
requires feature decisions

schemas:
  schema Capacity:
    device: text required
    available: integer required
  schema Inspection:
    device: text required
    passed: boolean required

conclusions:
  conclusion Evidence:
    goal: text required
    candidate: text required
    requirement: text required
    satisfied: boolean required
    reason: text required

rules:
  rule capacity_sufficient:
    when:
      match Capacity:
        device: bind device
        available: ge 10
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "capacity"
        satisfied: true
        reason: "CAPACITY_AT_LEAST_10"
  rule capacity_insufficient:
    when:
      match Capacity:
        device: bind device
        available: lt 10
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "capacity"
        satisfied: false
        reason: "CAPACITY_BELOW_10"
  rule inspection_passed:
    when:
      match Inspection:
        device: bind device
        passed: eq true
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "inspection"
        satisfied: true
        reason: "INSPECTION_PASSED"
  rule inspection_failed:
    when:
      match Inspection:
        device: bind device
        passed: eq false
    then:
      conclude Evidence:
        goal: "choose_equipment"
        candidate: use device
        requirement: "inspection"
        satisfied: false
        reason: "INSPECTION_FAILED"

tests:
  test sufficient_capacity:
    given:
      fact Capacity:
        device: "pump_a"
        available: 15
    expect:
      conclusion Evidence:
        goal: "choose_equipment"
        candidate: "pump_a"
        requirement: "capacity"
        satisfied: true
        reason: "CAPACITY_AT_LEAST_10"
    firings: 1

decisions:
  decision ChooseEquipment:
    goal: "choose_equipment"
    candidates: ["pump_a", "pump_b"]
    requirements: ["capacity", "inspection"]
    evidence: Evidence
    closed: true
    policy: "unique-feasible-v1"
    requires:
      rule capacity_sufficient
      rule capacity_insufficient
      rule inspection_passed
      rule inspection_failed
"""

from dataclasses import replace
from ting_reasoner import DecisionError, SourceSnapshot

# A declaration error is rejected before a compiled artifact is published.
invalid_source = source.replace("closed: true", "closed: 1")
try:
    compile_source(SourceSnapshot(invalid_source, uri="memory:equipment.ting"))
except SourceError as error:
    diagnostic = error.diagnostic
    assert diagnostic.code == "TING-SRC-006"
    print(diagnostic.code, diagnostic.message)
    print(diagnostic.span.uri, diagnostic.span.line)
    print(diagnostic.hint)
else:
    raise AssertionError("An integer must not stand in for true or false")

program = compile_source(source).program
try:
    program.decision("equipment.choice.Misspelled")
except KeyError:
    print("Available decisions:", program.decision_names)

# Requests are immutable. A validated copy can represent a different case.
original = program.decision("equipment.choice.ChooseEquipment")
open_request = replace(original, closed=False, logical_time=120)
assert original.closed is True and open_request.closed is False
assert original.digest != open_request.digest
try:
    replace(original, candidates=("pump_a", "pump_a"))
except DecisionError as error:
    assert error.code == "TING-DEC-001"
    print(error.as_dict())

A failed compilation leaves an existing compiled output intact. Compilation protects existing destinations by default. Requests are frozen objects; dataclasses.replace() creates and validates a new request for an application-controlled variation. That copy has its own digest and does not rewrite the compiled declaration. It is not an authorization mechanism or a signed policy. Keep the original request when the compiled candidate domain is the intended contract.

Build the same portable definition directly in Python

from pathlib import Path
from ting_reasoner import (
    DecisionDefinition, DecisionRequest, FactSchema, KnowledgeBase, Rule,
    bind, decide, emit, load_tingir, save_tingir, use,
)

Inspection = FactSchema("equipment.Inspection", device=str, passed=bool)
rule = Rule("inspect", (Inspection.pattern(device=bind("device"), passed=bind("passed")),),
    (emit("equipment.Evidence", goal="choose", candidate=use("device"),
          requirement="inspection", satisfied=use("passed"), reason="INSPECTED"),))
request = DecisionRequest("choose", ("pump_a", "pump_b"), ("inspection",),
                          "equipment.Evidence", closed=True)
knowledge = KnowledgeBase(rules=(rule,), decisions=(
    DecisionDefinition("equipment.Choose", request),
))
folder = Path.cwd() / "ting_projects" / "python decisions"
folder.mkdir(parents=True, exist_ok=True)
path = folder / "inspection.tingir"
save_tingir(knowledge, path, overwrite=True)
program = load_tingir(path)
result = decide(program, program.decision("equipment.Choose"), facts=(
    Inspection(device="pump_a", passed=True),
    Inspection(device="pump_b", passed=False),
))
assert result.selected_option == "pump_a"

DecisionDefinition(name, request) adds a named request to KnowledgeBase(..., decisions=(...)). The knowledge base includes the rules you supply. Named source imports and exports are available through the .ting library format; a Python definition alone does not declare a selective source dependency list. Both approaches can save a named decision in a compiled program. Use readable .ting source to declare selective library imports and dependencies. RuleSpec JSON does not support named decisions.

Reuse a request and set declaration limits

Retrieve the request once with program.decision(name) and reuse it across independent cases. For batch results, pass it to assess_decision(session, request), as shown in Make and inspect a decision. IRLimits.max_decisions defaults to 4,096; max_decision_items defaults to 100,000 total candidate plus requirement entries across the table. These bound compiled declarations; DecisionLimits and ExecutionBudget separately bound runtime assessment and inference.

Fixed-shape arrays

NDValue holds an immutable grid of values in row-major order (each row follows the previous one). Supported element types are bool, signed 64-bit i64, and finite 64-bit f64. A schema specifies dtype and exact shape.

from ting_reasoner import Engine, FactSchema, KnowledgeBase, NDField, NDValue

Grid = FactSchema("grid", (NDField("cells", "i64", (2, 2)),))
value = NDValue.from_flat([1, 2, 3, 4], dtype="i64", shape=(2, 2))
engine = Engine(KnowledgeBase((Grid,)))
engine.declare(Grid(cells=value))
assert engine.facts[0].values[0].tobytes() == value.tobytes()

Whole-array equality, inequality, membership, binding and derivation are supported. Shape, element count and byte limits are checked before construction. Use array_sum() to add numeric array values. Broadcasting, axis reductions, masks and general array arithmetic are not supported. Scalar Python integers and array i64 elements deliberately have different range contracts.

Add array values

Use array_sum() to add every element of an integer or floating-point array. The array stays unchanged. The sum follows the stored row order, starting from zero. An empty array returns zero of the same numeric type.

from ting_reasoner import NDValue, MemoryPolicy, array_sum, numeric_run

samples = NDValue.from_flat([1.0, 2.0, 3.0, 4.0], dtype="f64", shape=(4,))
with numeric_run(MemoryPolicy(max_managed_bytes=1024)) as usage:
    total = array_sum(samples)
    assert total == 10.0
print(total)
print(usage.report().peak_registered_bytes)  # 40 bytes

An integer sum must stay within the signed 64-bit range after each addition. A floating-point sum must remain finite. A calculation that exceeds either limit raises ValidationError; it does not return a wrapped number or infinity. Boolean arrays, choosing one axis, broadcasting and other arithmetic are not available.

Use a sum in a rule

Declare array-sum as a required feature. Bind an array in when, then write sum use samples in an output field. Use an integer output for i64, or float for f64. This expression is supported in the then block.

ting source 1
requires semantics 1
module sensor_totals version "1.0.0" role program
requires feature arrays
requires feature array-sum

schemas:
  schema Reading:
    name: text required
    samples: array f64[4] required

conclusions:
  conclusion Total:
    name: text required
    value: float required

rules:
  rule total_reading:
    when:
      match Reading:
        name: bind name
        samples: bind samples
    then:
      conclude Total:
        name: use name
        value: sum use samples

tests:
  test four_samples:
    given:
      fact Reading:
        name: "pump"
        samples: array f64[4] [1.0,2.0,3.0,4.0]
    expect:
      conclusion Total:
        name: "pump"
        value: 10.0
    firings: 1

Save this example as numeric_sum.ting, then call compile_file("numeric_sum.ting") from Python. Import that function with from ting_reasoner import compile_file. Compilation runs the embedded test, which expects 10.0, before saving numeric_sum.tingir beside the source file. See Compile readable source for a complete example of loading the result and supplying input facts. Older Ting installations cannot open the newly compiled sum; keep the editable source when transferring a program.

Share and inspect array storage

An Engine can reuse one immutable NDValue when several facts contain exactly the same array. Sharing checks the element type, shape and every canonical byte. Different shapes, types and positive/negative floating-point zero remain distinct. Each fact keeps its own handle, and every rule proof remains available in its established order.

Sharing is automatic for Python inputs, fixed knowledge and rule outputs, including programs loaded from .tingir. It needs no source declaration or setting. Continue to compare arrays with ==; Python object identity with is is not part of the value contract.

from ting_reasoner import (
    ArraySum, Engine, FactSchema, Field, KnowledgeBase, NDField, NDValue,
    Rule, bind, emit, use,
)

Reading = FactSchema("Reading", [
    Field("number", int), NDField("samples", "i64", (4,)),
])
rules = [
    Rule(name, Reading.pattern(number=bind("number"), samples=bind("samples")),
         [emit("Total", number=use("number"), value=ArraySum(use("samples")))])
    for name in ("primary", "verification")
]
engine = Engine(KnowledgeBase([Reading], rules))

def samples(values):
    return NDValue.from_flat(values, dtype="i64", shape=(4,))

# Construct each input array separately. Different numbers make distinct facts.
handles = engine.declare(
    Reading(number=1, samples=samples([1, 2, 3, 4])),
    Reading(number=2, samples=samples([1, 2, 3, 4])),
)
assert handles[0] != handles[1]
engine.run()
saved = engine.snapshot()
before = engine.array_storage_report
assert before.array_value_slots == 2
assert before.referenced_payload_bytes == 64
assert before.unique_payload_bytes == 32
assert before.shared_payload_bytes == 32
assert [row["value"] for row in engine.conclusions] == [10, 10]
assert all(len(engine.explain(row)) == 2 for row in engine.conclusions)

# Replace a complete fact; an immutable array is never edited in place.
engine.replace(handles[1], Reading(number=2, samples=samples([10, 20, 30, 40])))
engine.run()
after = engine.array_storage_report
assert after.unique_payload_bytes == 64
assert [row["value"] for row in engine.conclusions] == [10, 100]
assert saved.fact_items[1][1]["samples"].to_flat_tuple() == (1, 2, 3, 4)
print(before.as_dict())
print(after.as_dict())

Read the storage report

engine.array_storage_report returns an immutable ArrayStorageReport for arrays referenced by the current base and derived fact columns. It does not run inference or retain the arrays it inspects. Read a new report after an update; a saved report keeps its original numbers. Use as_dict() to obtain a detached dictionary suitable for logging.

Snapshots, cache capacity and memory limits

The sharing cache uses weak references. A cache entry alone does not keep an unused array alive. Your own variables, retained snapshots, explanations and compiled knowledge can still retain arrays for their promised lifetimes. Those references are outside the current fact-column payload totals. A report with zero current payload bytes does not imply that every old array has been released.

When the cache is full, newly encountered arrays use ordinary storage; admitted arrays can still be reused. Unequal values with the same fingerprint use ordinary storage. If allocating optional cache metadata fails, the cache is cleared and disabled for that Engine, and values continue through ordinary storage. Facts, alternative proofs and snapshots are not removed to fit the cache.

This is a storage inspection, not a memory limit or a process-RAM measurement. The numerical MemoryPolicy settings cover numeric operations; the optional session store-buffer limit applies separately. StoreMemoryReport and the compaction buffer limit cover their documented store buffers. Do not add their totals together as a whole-application memory figure, or interpret shared_payload_bytes as a measured reduction in laptop RAM. See Numeric run limits and Inspect fact storage for those separate scopes.

Numeric run limits

MemoryPolicy limits the array inputs and working buffers used by numerical operations. Set max_managed_bytes in bytes; one MiB is 1,048,576 bytes. Set max_numeric_work to limit the array elements and numeric filter checks processed during the run.

Use Engine(knowledge, memory=MemoryPolicy(...)), or a numeric_run() block as shown above to share a limit across several operations. Read engine.memory_report or usage.report() after the operation. The peak is the largest combined size of numeric buffers in use at once. Python object estimates are reported separately.

This is not a limit on all memory used by Ting. Facts, indexes, proofs, saved snapshots, Python itself and first-use preparation need additional memory. Independent process workers have separate limits; a shared limit across workers is not available. Automatic disk storage and recalculation to save memory are not available. See Keep an earlier explanation for managing saved snapshot references.

If a numeric limit is insufficient, Ting raises ResourceLimitError. Treat the result as incomplete. No part of the failing rule’s output is published; earlier completed rules remain committed. Increase the limit or reduce the input before retrying.

From a terminal, use:

ting-build --json test-source numeric_sum.ting --memory-bytes 1048576 --numeric-work 1000000

Read numeric filter usage

The following example selects three readings from 600 input facts. between() includes both endpoints. Ting can process eligible constant comparisons on floating-point fields in batches. It preserves the input order and uses private working memory; you do not need a different rule or import to enable this behavior.

from ting_reasoner import (
    Engine, FactSchema, KnowledgeBase, MemoryPolicy, Rule,
    between, bind, emit, use,
)

Reading = FactSchema("Reading", number=int, value=float)
rule = Rule(
    "select_readings",
    Reading.pattern(number=bind("number"), value=between(400.0, 402.0)),
    [emit("Selected", number=use("number"))],
)
engine = Engine(KnowledgeBase([Reading], [rule]),
                memory=MemoryPolicy(max_managed_bytes=4096))
# The asterisk supplies each generated fact as a separate declare() argument.
engine.declare(*(Reading(number=i, value=float(i)) for i in range(600)))
engine.run()
assert [row["number"] for row in engine.conclusions] == [400, 401, 402]
usage = engine.memory_report
assert usage.current_registered_bytes == 0
print(usage.peak_registered_bytes)
print(dict(usage.peak_by_category))
print(usage.filter_buffer_reuses)

peak_by_category separates input buffers, scalar outputs, filter result bitmaps and working masks. Category peaks may occur at different times. The result bitmap stores membership using one bit per physical row. peak_filter_bitmap_bytes and peak_filter_scratch_bytes describe these two allocations; they are already included in peak_registered_bytes, so do not add them to that total again. filter_buffer_allocations counts created filter buffers, and filter_buffer_reuses counts subsequent uses of a working mask within an operation. No filter buffers are cached between operations. Small inputs, unsupported comparisons and runs with an ExecutionBudget retain the established scalar matching path; their filter-buffer values can be zero.

Large matching sets

Ting may retain matching row references in an optional cache so several rules can reuse a selection. Cache construction is bounded. When the cache cannot hold a complete selection, Ting uses direct matching and still considers all relevant facts. This can take longer; it does not turn an incomplete selection into a negative answer or remove facts, outputs or explanations.

These reference lists are indexes and are outside both numerical and session store-buffer limits. The numeric report continues to describe numerical buffers. For large applications, account separately for facts, indexes, output records and saved snapshots.

The first numerical operation may take a few seconds to prepare. Later operations normally reuse this preparation. Ting installs NumPy and Numba for these operations; no separate acceleration option is needed. A read-only cache location can require preparation again on the next launch.

Compare several floating-point fields

Place a constant comparison on each field that must pass. A matching reading must satisfy every comparison. Conditions follow the schema field order, and a failed condition stops that reading’s remaining checks. between() includes both endpoints.

In this program, first must be at least zero and second must be between 490.0 and 499.0. Both positive and negative zero pass the ordering comparison. Equality and inequality comparisons distinguish their signs, so eq(-0.0) selects negative zero specifically.

Save and compile the complete program

ting source 1
requires semantics 1
module float_filtering version "1.0.0" role program

schemas:
  schema Reading:
    number: integer required
    first: float required
    second: float required

conclusions:
  conclusion Selected:
    number: integer required

rules:
  rule primary:
    when:
      match Reading:
        number: bind number
        first: ge -0.0
        second: between [490.0,499.0]
    then:
      conclude Selected:
        number: use number
  rule confirmation:
    when:
      match Reading:
        number: bind number
        first: ge -0.0
        second: between [490.0,499.0]
    then:
      conclude Selected:
        number: use number

tests:
  test lower_endpoint:
    given:
      fact Reading:
        number: 490
        first: -0.0
        second: 490.0
    expect:
      conclusion Selected:
        number: 490
    firings: 2
  test upper_endpoint:
    given:
      fact Reading:
        number: 499
        first: 0.0
        second: 499.0
    expect:
      conclusion Selected:
        number: 499
    firings: 2
  test first_field_rejects:
    given:
      fact Reading:
        number: 495
        first: -1.0
        second: 495.0
    firings: 0
  test second_field_rejects:
    given:
      fact Reading:
        number: 500
        first: 0.0
        second: 500.0
    firings: 0

Save it as float_filtering.ting. In the IDE, Check validates its types, Test checks the endpoints and both rejection cases, and Compile writes the compiled file after those tests pass. From a terminal, use:

ting-build compile float_filtering.ting -o float_filtering.tingir

Run the compiled file from Python

Save this Python example beside the compiled file. It supplies 601 readings, checks all ten selected numbers, and verifies that both rules remain visible as separate explanations. The source file is not needed to run the compiled program.

from pathlib import Path
from ting_reasoner import Engine, MemoryPolicy, load_tingir

# Compile float_filtering.ting before running this file in the same folder.
program = load_tingir(Path("float_filtering.tingir"))
engine = Engine(program, memory=MemoryPolicy(max_managed_bytes=8192))
Reading = engine.schema("float_filtering.Reading")
engine.declare(*(Reading(number=i, first=(-0.0 if i % 2 else 0.0),
                        second=float(i)) for i in range(601)))
engine.run()
saved = engine.snapshot()
numbers = [row["number"] for row in saved.conclusions]
proof_counts = [len(engine.explain(row)) for row in saved.conclusions]
assert numbers == list(range(490, 500))
assert proof_counts == [2] * 10
assert engine.memory_report.current_registered_bytes == 0
print(numbers)
print(proof_counts)

Ting automatically batches eligible constant float comparisons on sufficiently large inputs. This preserves results, input order and logical matching counts. Small inputs, other field types and runs with an ExecutionBudget use direct matching. First-use numerical preparation can dominate a small case; measure the complete application case and repeated runs separately. See Numeric run limits for the scope of the buffer limit and Measure a complete case for an executable timing example.

Inspect fact storage

Read engine.store_memory_report when you want to inspect the current physical fact store. Ting creates an immutable StoreMemoryReport for you. Reading it does not change facts, results or the engine epoch, and does not create public copies of every fact.

from ting_reasoner import Engine, FactSchema, KnowledgeBase

Reading = FactSchema("Reading", number=int, value=float)
engine = Engine(KnowledgeBase([Reading]))
handles = engine.declare(
    Reading(number=1, value=10.0),
    Reading(number=2, value=20.0),
    Reading(number=3, value=30.0),
)
saved = engine.snapshot()
before = engine.store_memory_report
engine.retract(handles[0])
# Inspect storage after withdrawal, then compact any remaining unused rows.
withdrawn = engine.store_memory_report
engine.compact_store()
after = engine.store_memory_report
assert after.current_buffer_payload_bytes <= withdrawn.current_buffer_payload_bytes
assert after.current_buffer_payload_bytes < before.current_buffer_payload_bytes
assert len(saved.fact_items) == 3
print(after.as_dict())

current_buffer_payload_bytes is the combined size of the stored numeric or byte field buffers and row buffers. Read column_buffer_payload_bytes and row_buffer_payload_bytes for those two parts. Any unused rows still present are included. An update may rebuild storage automatically; call compact_store() to remove remaining unused rows. estimated_container_overhead_bytes separately estimates container headers, reference slots and spare capacity; its value can differ between Python implementations. object_value_slots counts slots in object columns, including unused slots.

This report is an inspection, not a memory limit or a reading of total process memory. It excludes the values referenced by object columns, including text, arbitrary-precision integers and arrays, as well as indexes, proofs, saved snapshots, the compiled plan and numerical preparation. Its coverage and excluded fields state that scope. It is separate from engine.memory_report, which reports numeric operations and their limits. Read the store report occasionally when reviewing storage; it visits the store’s containers each time.

compact_store() preserves live facts, handles and saved snapshots. Keeping a snapshot keeps its values available even after compaction or reset. Release snapshots when you no longer need them; see Keep an earlier explanation.

Limit session store buffers

Set MemoryPolicy(max_store_buffer_bytes=...) when constructing an Engine to bound the payload of its numeric and byte fact columns and row metadata. The value is a non-negative integer in bytes; None, the default, applies no session store-buffer limit. It includes unused physical rows and the extra payload copied or appended during a change.

A change may need more room than its final facts occupy. Ting keeps the current store available while preparing an update. Unchanged shared chunks count once; copied chunks count separately until publication. In this example each row contributes 30 payload bytes. Adding the second row needs 30 bytes for the existing row, 30 for its copied chunk and 30 for the new row. The next addition would need 150 bytes, so it fails before changing the session.

from ting_reasoner import (
    Engine, FactSchema, KnowledgeBase, MemoryPolicy, ResourceLimitError,
    Rule, bind, emit, gt, use,
)

Reading = FactSchema("Reading", number=int, value=float, ready=bool)
rule = Rule("select", Reading.pattern(number=bind("number"), value=gt(1.0)),
            [emit("Selected", number=use("number"))])
engine = Engine(KnowledgeBase([Reading], [rule]),
                memory=MemoryPolicy(max_store_buffer_bytes=90))
engine.declare(Reading(number=1, value=1.0, ready=True))
engine.declare(Reading(number=2, value=2.0, ready=True))
engine.run()
saved = engine.snapshot()

try:
    engine.declare(Reading(number=3, value=3.0, ready=True))
except ResourceLimitError as error:
    assert error.resource == "session store buffer bytes"
    assert (error.requested, error.limit) == (150, 90)
else:
    raise AssertionError("The store limit should reject this change")
assert engine.snapshot() == saved
report = engine.store_allocation_report
assert report.staged_buffer_payload_bytes == 0
print(report.current_buffer_payload_bytes, report.peak_reserved_buffer_payload_bytes)
# Output: 60 90. The two rows occupy 60 payload bytes; the update peak was 90.

engine.reset()
assert len(saved.fact_items) == 2  # A snapshot retains its own public values.
print(engine.store_allocation_report.current_buffer_payload_bytes)  # 0

Read the admission report

engine.store_allocation_report returns an immutable StoreAllocationReport. Reading it uses maintained counters and does not scan the store. current_buffer_payload_bytes describes initialized payload in the committed store. peak_reserved_buffer_payload_bytes is the largest reservation bound since this Engine was created, including failed attempts. It can exceed actual allocated payload because admission precedes allocation. staged_buffer_payload_bytes is zero after an operation returns or raises. allocation_reservations counts accepted growth and copy reservations; limit_bytes records the configured cap. Reset releases the current store payload but preserves this lifetime peak.

This cap does not cover total application memory. Container headers, spare capacity, values in object columns (including text, integers and arrays), indexes, proofs, snapshots, error tracebacks, the compiled plan, numerical working buffers and process RAM are outside it. Read coverage and excluded on the report. Use store_memory_report for a separate container inspection; do not add its current payload to the admission report because both describe the same current buffers.

Handle an insufficient limit

ResourceLimitError means the requested operation did not complete. A failed declaration, replacement, retraction, transaction publication, reset or compaction leaves the committed session unchanged. A failing rule publishes none of that rule’s consequences; earlier completed firings remain committed. Reservations are released on errors and interruptions. Allow room for update overlap when choosing a limit. To use a different policy, create a new Engine with the larger limit and redeclare the intended input facts.

The session limit applies throughout the Engine’s lifetime, including immutable seeds and derived facts. The optional max_buffer_bytes on transactions and compaction adds an operation-specific limit; both checks apply. max_managed_bytes and max_numeric_work continue to limit numerical operations independently.

Run a compiled program

Save the complete source from Compare several floating-point fields as float_filtering.ting. Compile it in the IDE, or run ting-build compile float_filtering.ting -o float_filtering.tingir in that directory. Then save the following Python code beside the compiled file and run it. The source file is no longer required. The IDE and compiler terminal commands use their usual defaults; configure this session limit in the Python Engine that runs the compiled program.

from ting_reasoner import Engine, MemoryPolicy, ResourceLimitError, load_tingir

program = load_tingir("float_filtering.tingir")
engine = Engine(program, memory=MemoryPolicy(
    max_managed_bytes=8192, max_store_buffer_bytes=22237,
))
Reading = engine.schema("float_filtering.Reading")
engine.declare(*(Reading(number=i, first=0.0, second=float(i)) for i in range(601)))
engine.run()
saved = engine.snapshot()
assert sorted(item["number"] for item in engine.conclusions) == list(range(490, 500))
assert engine.store_allocation_report.current_buffer_payload_bytes == 22237

try:
    engine.declare(Reading(number=601, first=0.0, second=601.0))
except ResourceLimitError:
    pass
else:
    raise AssertionError("The store limit should reject this change")
assert engine.snapshot() == saved
print(engine.store_allocation_report.current_buffer_payload_bytes)  # 22237

Independent jobs

Pass the same policy through InferenceJob(memory=MemoryPolicy(max_store_buffer_bytes=...)) when using run_batch() or iter_batch(). Each job receives its own session limit, whether it runs inline or in a process. Its SessionResult.store_allocation_report records that session’s counters. These caps do not combine into a shared limit across workers or across multiple Engines. Engines created inside numeric_run() inherit the policy but still have separate store limits.

Inspect exact equality lookups

When a rule matches a field against a constant or an earlier binding, Ting prepares an exact equality index for that field. The index maps each canonical value to an ordered sequence of matching fact rows. A key with one row uses a compact immutable sequence; adding another matching row expands the sequence. Removing rows can make it compact again. The returned facts, conclusions and explanations retain their logical order.

This equipment example joins two ordinary text fields. Such a join permits several Assets with the same serial and does not enforce target existence or entity-key uniqueness. Both matching facts produce their own complete evidence.

from ting_reasoner import Engine, FactSchema, KnowledgeBase, Rule, bind, emit, eq, use

Request = FactSchema("Request", serial=str)
Asset = FactSchema("Asset", serial=str, location=str)
rule = Rule("locate", [Request.pattern(serial=bind("serial")),
                       Asset.pattern(serial=eq(use("serial")), location=bind("location"))],
            [emit("Located", serial=use("serial"), location=use("location"))])
engine = Engine(KnowledgeBase([Request, Asset], [rule]))
engine.declare(Request(serial="pump-a"), Asset(serial="pump-a", location="north"),
               Asset(serial="pump-b", location="south"))
engine.run()
saved = engine.snapshot()
report = engine.equality_index_memory_report
print(report.posting_buckets, report.singleton_buckets, report.row_reference_slots)
# Output: 2 2 2

extra = engine.declare(Asset(serial="pump-a", location="backup"))
engine.run()
assert [item["location"] for item in engine.conclusions] == ["north", "backup"]
assert all(len(engine.explain(item)) == 1 for item in engine.conclusions)
report = engine.equality_index_memory_report
print(report.posting_buckets, report.singleton_buckets,
      report.multirow_buckets, report.row_reference_slots)
# Output: 2 1 1 3
assert len(saved.conclusions) == 1

engine.retract(extra)
engine.run()
assert engine.equality_index_memory_report.singleton_buckets == 2
assert engine.conclusions[0]["location"] == "north"

Understand the report

engine.equality_index_memory_report returns an immutable EqualityIndexMemoryReport. The report inspects the current indexes under the session lock; reading it takes time proportional to the number of posting buckets. indexed_fields counts fields with prepared equality dictionaries. posting_buckets counts their distinct indexed keys. singleton_buckets and multirow_buckets classify those keys by matching-row count. row_reference_slots counts retained row references, including separate references when several fields are indexed.

The three estimated_*_bytes fields report Python-specific shallow sizes for posting containers, dictionaries and their sum. Keys, key values, row-reference integers, other indexes, routes, temporary update overlap and saved snapshots are excluded. These estimates are separate from initialized store payload and numeric reservations. Neither existing MemoryPolicy byte limit admits these index containers. Read the report’s coverage and excluded fields; do not interpret the estimate as process RAM, peak usage or a whole-Engine limit.

Author and compile the same lookup

Save this complete source as indexed_equipment.ting. In the IDE use Check, Tests and Compile, or run ting-build compile indexed_equipment.ting -o indexed_equipment.tingir. The three tests check one match, several matches and an unmatched request. The second pattern reads the serial bound by the first pattern.

ting source 1
requires semantics 1
module indexed_equipment version "1.0.0" role program

schemas:
  schema Request:
    serial: text required
  schema Asset:
    serial: text required
    location: text required

conclusions:
  conclusion Located:
    serial: text required
    location: text required

rules:
  rule locate:
    when:
      match Request:
        serial: bind serial
      match Asset:
        serial: eq use serial
        location: bind location
    then:
      conclude Located:
        serial: use serial
        location: use location

tests:
  test single_match:
    given:
      fact Request:
        serial: "pump-a"
      fact Asset:
        serial: "pump-a"
        location: "north"
    expect:
      conclusion Located:
        serial: "pump-a"
        location: "north"
    firings: 1
  test several_matches:
    given:
      fact Request:
        serial: "pump-a"
      fact Asset:
        serial: "pump-a"
        location: "north"
      fact Asset:
        serial: "pump-a"
        location: "south"
    expect:
      conclusion Located:
        serial: "pump-a"
        location: "north"
      conclusion Located:
        serial: "pump-a"
        location: "south"
    firings: 2
  test no_matching_asset:
    given:
      fact Request:
        serial: "pump-b"
      fact Asset:
        serial: "pump-a"
        location: "north"
    expect:
    firings: 0

Save the following Python script beside the compiled program and run it. The original source is no longer required. Compiled execution uses the same indexes and provides the same inspection report.

from ting_reasoner import Engine, load_tingir

engine = Engine(load_tingir("indexed_equipment.tingir"))
Request = engine.schema("indexed_equipment.Request")
Asset = engine.schema("indexed_equipment.Asset")
engine.declare(Request(serial="pump-a"), Asset(serial="pump-a", location="north"),
               Asset(serial="pump-a", location="backup"))
engine.run()
assert [item["location"] for item in engine.conclusions] == ["north", "backup"]
report = engine.equality_index_memory_report
print(report.posting_buckets, report.multirow_buckets, report.row_reference_slots)
# Output: 1 1 2

Compact fact storage

Use engine.compact_store() outside a transaction to remove unused physical fact rows after updates. It returns StoreMetrics. Fact order, handles, conclusions, explanations and the engine epoch stay unchanged. A saved snapshot continues to contain its original facts.

Compaction builds a replacement while the current store remains available. Supply the optional max_buffer_bytes keyword to check the combined old and replacement buffer payload before the replacement is allocated. The value must be a non-negative integer in bytes; None, the default, applies no compaction buffer limit. This setting is independent of numerical limits. If a session store-buffer limit is also configured, both store checks apply.

from ting_reasoner import (
    Engine, FactSchema, KnowledgeBase, ResourceLimitError,
)

Reading = FactSchema("Reading", number=int, value=float)
engine = Engine(KnowledgeBase([Reading]))
handles = engine.declare(
    *(Reading(number=i, value=float(i)) for i in range(300))
)
saved = engine.snapshot()
engine.retract(handles[-1])
before_compaction = engine.snapshot()

# Zero bytes cannot hold the old and replacement buffers together.
try:
    engine.compact_store(max_buffer_bytes=0)
except ResourceLimitError as error:
    print(error.resource, error.requested, error.limit)
assert engine.snapshot() == before_compaction

# Removing unused rows needs no more payload than two current stores.
maximum = 2 * engine.store_memory_report.current_buffer_payload_bytes
metrics = engine.compact_store(max_buffer_bytes=maximum)
report = engine.store_compaction_report
assert report is not None
assert report.overlap_buffer_payload_bytes <= maximum
assert report.peak_staged_rows == 1
assert metrics.tombstone_rows == 0
assert engine.snapshot() == before_compaction
assert len(saved.fact_items) == 300
print(report.as_dict())

Read the completed operation

engine.store_compaction_report returns an immutable StoreCompactionReport after a successful compaction. input_buffer_payload_bytes includes the old store’s unused rows; output_buffer_payload_bytes describes the dense replacement. overlap_buffer_payload_bytes is their combined payload while both stores exist. rows_copied counts live base and derived rows. peak_staged_rows describes temporary row-value tuples: at most one row is staged during copying. The reference mapping and new indexes still grow with the number of live rows.

buffer_limit_bytes records the supplied limit. estimated_remap_container_bytes estimates the mapping container itself, separately from the limit; it excludes the integers referenced by that container. Read coverage and excluded before interpreting these values.

The limit covers numeric/byte field buffers and row-buffer payload lengths. Container headers, spare allocation capacity, object-column values such as text and arrays, reference mappings, indexes, proofs, saved snapshots and process memory require additional space. This limit cannot guarantee that the whole operation fits in a particular amount of laptop RAM.

Handle a limit or allocation failure

An insufficient buffer limit raises ResourceLimitError with resource, requested and limit values. Python can separately raise MemoryError for allocations outside this limit. A failure before publication leaves the current store, handles, outputs and last successful compaction report intact. Increase the appropriate limit or reduce retained application data before retrying; compaction does not discard live facts or explanations to fit.

If there are no unused rows, the call returns current metrics without copying anything or replacing the report. The report is initially None, and a complete store replacement, including reset(), clears it. Holding a report retains its numbers independently of later changes.

Choose an execution mode

Choose how to organize your cases before changing performance settings. A case is one set of observations evaluated by the rules.

Measure the operation your application actually performs, including loading, declaring facts, inference and inspecting results. Check that the results and proofs remain correct. Larger snapshots cost more to store and transfer; request only the work and outputs your application needs.

Measure a complete case

Reuse a verified compiled program for independent cases. The example below prepares its rules once, then measures each case from Engine creation through fact construction, declaration, inference and the required conclusions and proofs. It reports compilation separately and retains the first numerical execution separately from the seven repeated measurements. Every measurement uses the same inputs and checks the expected ordered answers and both supporting rules.

from statistics import median
from time import perf_counter
from ting_reasoner import (
    Engine, FactSchema, KnowledgeBase, MemoryPolicy, Rule,
    between, bind, compile_knowledge_base, emit, use,
)

prepare_started = perf_counter()
Reading = FactSchema("Reading", number=int, value=float)
rules = [
    Rule(name, Reading.pattern(number=bind("number"), value=between(490.0, 499.0)),
         [emit("Selected", number=use("number"))])
    for name in ("primary", "confirmation")
]
program = compile_knowledge_base(KnowledgeBase([Reading], rules))
prepare_seconds = perf_counter() - prepare_started

def evaluate_case():
    started = perf_counter()
    engine = Engine(program, memory=MemoryPolicy(max_managed_bytes=8192))
    engine.declare(*(Reading(number=i, value=float(i)) for i in range(600)))
    result = engine.run()
    conclusions = engine.conclusions
    proofs = [engine.explain(row) for row in conclusions]
    elapsed = perf_counter() - started
    assert result.firings == 20
    assert [row["number"] for row in conclusions] == list(range(490, 500))
    assert all(len(edges) == 2 for edges in proofs)
    usage = engine.memory_report
    assert usage.current_registered_bytes == 0
    return elapsed, usage

# Keep first-use preparation visible, then measure repeated cases.
first_seconds, _ = evaluate_case()
samples = [evaluate_case()[0] for _ in range(7)]
_, usage = evaluate_case()
print({"program_preparation_seconds": prepare_seconds,
       "first_case_seconds": first_seconds,
       "repeated_case_median_seconds": median(samples),
       "repeated_case_range_seconds": [min(samples), max(samples)]})
print(usage.as_dict())

between(490.0, 499.0) includes both endpoints. Both rules use the same constant filter; the resulting ten conclusions each retain two proofs. Eligible constant comparisons on float columns use numeric filtering automatically. Integer, binding-dependent and structured logical-budget paths keep their established matching behavior. No compiler option or source-format change is needed to use these paths.

The first numeric operation may compile a fixed kernel. Later operations can reuse compatible prepared code. Imports, CPU load, dependency versions and cache conditions affect latency; compare the same application operation under the same conditions. The reported median is an observation on your machine. Small cases can vary substantially, so retain the range and repeat with representative input sizes.

Choose the view your application needs. engine.facts returns immutable values; engine.fact_items also returns handles for updates. engine.snapshot() retains a consistent view of facts, conclusions and explanations. Save a returned tuple in a local variable when several checks need the same view, and request a new view after changing inputs.

The numeric report measures registered numeric buffers. It excludes facts, selected-row references, indexes, proofs, snapshots, compilation and total process RAM. Read Numeric run limits for that boundary. Keep complete evidence when your application needs explanations, and release saved snapshots once their history is no longer needed. Measure those choices as part of the application workload.

Pruning and performance

Pass an OptimizationPolicy to Engine when you want to set limits on optional search indexes. Start with the defaults, then inspect the counters using facts and rules representative of your application.

from ting_reasoner import (
    Engine, FactSchema, KnowledgeBase, OptimizationPolicy, Rule,
    bind, emit, ge, use,
)

Reading = FactSchema("reading", sensor=str, value=int)
rule = Rule("high_reading",
            (Reading.pattern(sensor=bind("sensor"), value=ge(35)),),
            (emit("high_reading", sensor=use("sensor")),))
knowledge = KnowledgeBase((Reading,), (rule,))
engine = Engine(knowledge, optimization=OptimizationPolicy(
    minimum_rows=32,
    max_indexes=32,
    max_cache_cells=1_048_576,
    max_build_work=4_194_304,
))
engine.declare(*(Reading(sensor=str(i), value=i) for i in range(40)))
result = engine.run()
assert result.status.value == "quiescent"
assert len(engine.conclusions) == 5
print(engine.match_metrics)
print(engine.optimization_metrics)

minimum_rows sets the smallest candidate collection considered for these indexes. max_indexes limits their number; max_cache_cells limits stored index entries, and max_build_work limits the work used to construct them. These are work and entry limits, not a limit on the whole Python process's memory. Reaching an optional index limit can make a query slower but does not omit valid matches.

Read engine.match_metrics after run() to see candidate-row and constraint counts. engine.optimization_metrics.pruned_rows reports avoided candidate checks, and build_rows reports rows examined while constructing indexes. The example emits five high-reading conclusions. Counters may be zero when an index is unnecessary. Use OptimizationPolicy(enabled=False) when comparing execution with these optional indexes disabled; it does not disable every optimization in Ting.

For repeated positive-rule updates, enable MaintenancePolicy() as described in Keep valid results when inputs change. When writing a join, include fields that identify the intended relationship, as in Rules and conditions. Matching every fact against every other fact can create many combinations even with indexing.

Independent sessions in parallel

Before this section: Reuse a compiled program and Work limits.

Use batch execution for separate cases that share a knowledge base. Each case gets its own Engine, fact memory, handles and proofs. Results return in input order regardless of which process finishes first.

from ting_reasoner import (
    ExecutionPolicy, FactSchema, InferenceJob, KnowledgeBase,
    Rule, bind, emit, run_batch, use,
)

def main():
    Fruit = FactSchema("fruit", name=str, sweet=bool)
    rule = Rule("sweet", (Fruit.pattern(name=bind("name"), sweet=True),),
                (emit("sweet_fruit", name=use("name")),))
    knowledge = KnowledgeBase((Fruit,), (rule,))
    jobs = [
        InferenceJob((Fruit(name="apple", sweet=True),)),
        InferenceJob((Fruit(name="lemon", sweet=False),)),
    ]
    result = run_batch(knowledge, jobs, policy=ExecutionPolicy(
        mode="owned", max_workers=2, max_in_flight=2,
        minimum_parallel_facts=0,  # Force processes for this tiny demonstration.
    ))
    assert [len(item.snapshot.conclusions) for item in result.results] == [1, 0]
    for item in result.results:
        print(item.sequence, item.snapshot.conclusions)

if __name__ == "__main__":
    main()

Save the complete example as parallel_example.py and run it from a terminal. The if __name__ == "__main__": guard prevents a new worker from launching more workers as it loads the script. Use inline mode in environments that cannot launch processes. Worker processes are part of your application; they do not isolate untrusted code.

Read results one at a time

This complete example uses inline execution and prints results for apple, then pear. closing() releases the iterator’s resources if you stop early. For process workers, use the main-guard structure in the preceding example.

from contextlib import closing
from ting_reasoner import (
    ExecutionPolicy, FactSchema, InferenceJob, KnowledgeBase,
    Rule, bind, emit, iter_batch, use,
)

Item = FactSchema("item", name=str)
rule = Rule("record", (Item.pattern(name=bind("name")),),
            (emit("recorded", name=use("name")),))
knowledge = KnowledgeBase((Item,), (rule,))
jobs = [InferenceJob((Item(name="apple"),)),
        InferenceJob((Item(name="pear"),))]
policy = ExecutionPolicy(mode="inline", max_in_flight=2, chunk_size=1)
with closing(iter_batch(knowledge, jobs, policy=policy)) as results:
    for result in results:
        assert result.run.status.value == "quiescent"
        for conclusion in result.snapshot.conclusions:
            print(result.sequence, conclusion.as_dict())
# closing() also cleans up if a loop exits early with break or an exception.

max_in_flight × chunk_size bounds submitted/prefetched jobs; each job is limited to 100,000 facts. It does not bound the total bytes in facts or output. run_batch retains all results until the whole batch succeeds, with a configurable total-job ceiling. iter_batch yields ordered results and allows earlier results to remain usable if a later job fails.

Cancellation accepts an object with is_set(), such as a threading Event. It stops submissions and publication at coordinator boundaries. Already-running jobs finish; cancellation is not a hard process kill or a deadline. Owned-pool cleanup waits for those jobs. Use logical job budgets to bound their work.

BatchExecutionError.sequence identifies a failed job. Caller-owned engines are never mutated. Worker startup and executor failures propagate; Ting does not silently hide a failed process launch. A borrowed executor remains usable afterward.

Batch execution does not split one live reasoning session across cores. It improves throughput only when independent work outweighs process and serialization costs. For CPU work, a borrowed process pool is generally more appropriate than a thread pool.

Troubleshooting

Exceptions preserve category-specific information: schema/rule validation errors, compilation/IR verification errors, engine errors, stale transactions, history errors and batch errors. Catch the specific category you can recover from. Use build errors, decision errors, or batch errors for recovery examples, and the exception reference for their base class.

Assurance and verification

Ting provides a scoped assurance profile for typed reference construction, snapshot index ownership, storage inspection and their verification controls. Requirements connect to design explanations, code paths, named tests, failure scenarios and unresolved issues. The checker produces reverse links so an implementation or test can be traced back to its requirement. These records support engineering assessment; they do not establish NASA certification or a deployment’s safety classification.

Check a delivered source release

Extract the source archive and install its declared dependencies. Run these commands from its root. The supplied baseline identifies the controlled files and required test inventory; verify it before interpreting results.

python -m tools.assurance
python -m tools.run_tests --output tests.json
python -m tools.assurance --results tests.json --output assurance.json

The first command checks profile links and configuration integrity. The second executes all required cases and records case IDs, outcomes, failure details, environment versions and loaded-package hashes. The third checks that evidence against the baseline. An error in one command must not be hidden by a later successful command. On headless Linux, set QT_QPA_PLATFORM=offscreen for GUI tests; this does not replace native desktop validation.

With --results, exit code 0 means consistent evidence and a passing complete required suite; 1 means configuration or evidence is invalid; 2 means required-suite acceptance is unmet. Without results, exit 0 means only structural and configuration checks passed. Skips, expected failures, missing cases and stale runtime bytes cannot be accepted. Neither a passing checker nor a content hash authenticates an untrusted producer or grants NASA approval.

Maintain application evidence

Keep the exact rule source, knowledge inputs, compiled program digest, Ting/dependency versions, run settings, expected outputs, actual outcomes and explanations together. Record changes and their effects. A reference constructor identifies an entity; only an explicit match or follow adds the target fact as a premise. Use the complete construction examples to see that distinction.

Applications must define their response to incomplete or conflicting decisions and failed operations. A work budget counts operations; it does not promise a deadline. Numerical-byte and graph-entry limits have stated coverage and do not bound whole-process memory. The entity storage report is a shallow diagnostic measurement, not byte admission or a whole-heap bound. Retain failed results and validate on the intended platform.

Standards and applicability

• NPR 7150.2D — NASA Software Engineering Requirements
• NASA-STD-8739.8B — Software Assurance and Software Safety Standard

The profile applies selected traceability, configuration and testing controls from these references. Full compliance requires an intended-use assessment, applicable requirements and approved tailoring, complete product/system traceability, accountable authorities, appropriate safety evidence and independent assurance where required. For identified safety-critical components, ordinary unit or branch coverage does not demonstrate the required 100% Modified Condition/Decision Coverage (MC/DC).

NASA Software Engineering Handbook topic 7.25, Artificial Intelligence And Software Engineering, discusses deterministic verification. Topic 8.25, Artificial Intelligence And Software Assurance, addresses data-driven and agentic systems; its applicability depends on an application’s integration. The NASA insignia in the reference panel identifies the source agency only. Ting is independently developed.

Availability and limits

The package provides typed facts, declarative conditions, verified compiled programs, deterministic inference, base-fact transactions, immutable snapshots, explanations, fixed-shape arrays, bounded optional pruning and independent-session multiprocessing. The decision API evaluates hard requirements over an explicit candidate list, preserving unknowns, conflicts and ties.

For changing observations, follow the input-update rules. The API does not provide aggregate operators, temporal or backward reasoning, exact process-memory budgets, or parallel matching inside one Engine. Implement application actions in Python after checking the returned results.

The supported application scope excludes banking and financial use. See the package’s LICENSE and LICENSING.md files for the terms and project policy.

API reference

Look up a name here after choosing the relevant tutorial. The reference lists every symbol exported by ting_reasoner, including advanced and optional APIs; you do not need all of them to build a reasoner. Read Understand the Python calls for signature notation. Each usage link returns to the related explanation and examples.

__version__ = '0.10.0a21'

Action(kind: 'str', target: 'FactSchema | str', values: 'tuple[tuple[str, Value | BindingRef], ...]') -> None

all_of(*conditions: 'Condition') -> 'Condition'

AllCondition(conditions: 'tuple[Condition, ...]') -> None

analyze_source(source: 'str | SourceSnapshot', *, resolver: 'SourceResolver | None' = None, limits: 'SourceLimits | None' = None, ir_limits: 'IRLimits | None' = None) -> 'SourceAnalysis'

any_of(*conditions: 'Condition') -> 'Condition'

AnyCondition(conditions: 'tuple[Condition, ...]') -> None

array_sum(value: 'NDValue', *, memory: 'MemoryLedger | None' = None) -> 'int | float'

report = engine.array_storage_report
print(report.unique_payload_bytes)
print(report.shared_payload_bytes)
print(report.as_dict())

ArraySum(value: 'BindingRef | NDValue') -> None

assess_decision(session: 'SessionResult', request: 'DecisionRequest', *, limits: 'DecisionLimits | None' = None) -> 'DecisionResult'

AuthoringBackend()

BatchCancelledError

BatchExecutionError(sequence: 'int', message: 'str') -> 'None'

BatchMetrics(jobs: 'int' = 0, submitted_chunks: 'int' = 0, inline_jobs: 'int' = 0, peak_in_flight: 'int' = 0, pool_created: 'bool' = False, nested_inline: 'bool' = False) -> None

BatchResult(results: 'tuple[SessionResult, ...]', metrics: 'BatchMetrics') -> None

between(low: 'Value | BindingRef', high: 'Value | BindingRef') -> 'Constraint'

bind(name: 'str') -> 'Constraint'

BindingRef(name: 'str') -> None

BudgetUsage(candidate_rows: 'int' = 0, constraint_evaluations: 'int' = 0, activations: 'int' = 0, firings: 'int' = 0) -> None

CandidateAssessment(candidate: 'str', feasibility: 'CandidateFeasibility', constraints: 'tuple[ConstraintAssessment, ...]') -> None

CandidateFeasibility(*values)

canonical_source_bytes(spec: 'RuleSpec', *, limits: 'RuleSpecLimits | None' = None) -> 'bytes'

CheckpointRevision(revision: 'str', parent: 'str | None', restored_from: 'str | None', source_object: 'str', rulespec_object: 'str', artifact_object: 'str', source_digest: 'str', semantic_digest: 'str', artifact_digest: 'str', artifact_payload_digest: 'str', artifact_size: 'int', ting_version: 'str', format_version: 'int', semantics_version: 'int', rulespec_version: 'int', test_passed: 'bool', test_passed_count: 'int', test_failed_count: 'int') -> None

CLASS_FORMAT_VERSION = 6

CLASS_SEMANTICS_VERSION = 7

Classification(name: 'str', record_schema: 'FactSchema', *, parents: 'Iterable[Classification]' = ()) -> 'None'

CommitReceipt(delta: 'DeltaBatch', resolutions: 'tuple[tuple[StagedHandle, FactHandle | None], ...]' = (), *, buffer_report: 'TransactionBufferReport | None' = None) -> None

CompactIR(data: 'bytes | bytearray | memoryview', *, limits: 'IRLimits | None' = None) -> 'None'

CompilationError

compile(kb: 'KnowledgeBase', *, limits: 'IRLimits | None' = None) -> 'bytes'

compile_file(path: 'str | os.PathLike[str]', output: 'str | os.PathLike[str] | None' = None, *, modules: 'Mapping[str, str | os.PathLike[str]] | None' = None, overwrite: 'bool' = False, run_tests: 'bool' = True, limits: 'SourceLimits | None' = None, ir_limits: 'IRLimits | None' = None, test_budget: 'ExecutionBudget | None' = None, library_limits: 'LibraryLimits | None' = None) -> 'SourceBuild | LibraryBuild'

compile_files(paths: 'Iterable[str | os.PathLike[str]]', output_directory: 'str | os.PathLike[str]', *, modules: 'Mapping[str, str | os.PathLike[str]] | None' = None, overwrite: 'bool' = False, run_tests: 'bool' = True, max_files: 'int' = 256, limits: 'SourceLimits | None' = None, ir_limits: 'IRLimits | None' = None, test_budget: 'ExecutionBudget | None' = None, library_limits: 'LibraryLimits | None' = None) -> 'tuple[FileBuildResult, ...]'

compile_knowledge_base(kb: 'KnowledgeBase', *, limits: 'IRLimits | None' = None) -> 'bytes'

compile_rulespec(spec: 'RuleSpec', *, limits: 'IRLimits | None' = None) -> 'bytes'

compile_source(source: 'str | SourceSnapshot', *, resolver: 'SourceResolver | None' = None, limits: 'SourceLimits | None' = None, ir_limits: 'IRLimits | None' = None, run_tests: 'bool' = True, test_budget: 'ExecutionBudget | None' = None, library_limits: 'LibraryLimits | None' = None) -> 'SourceBuild | LibraryBuild'

CompiledCatalog(libraries: 'Mapping[str, str | os.PathLike | CompiledLibrary]', *, limits: 'LibraryLimits | None' = None)

CompiledLibrary(data: 'bytes', *, limits: 'LibraryLimits | None' = None)

Conclusion(name: 'str', values: 'tuple[tuple[str, Value], ...]', rule: 'str', fact_ids: 'tuple[int, ...]') -> None

Condition = ting_reasoner.model.Pattern | ting_reasoner.model.TestCondition | ting_reasoner.model.ExistsCondition | ting_reasoner.model.ForAllCondition | ting_reasoner.model.AllCondition | ting_reasoner.model.AnyCondition

Constraint(operator: 'str', operands: 'tuple[Value | BindingRef, ...]' = ()) -> None

ConstraintAssessment(requirement: 'str', state: 'EvidenceState', support: 'tuple[Conclusion, ...]' = (), opposition: 'tuple[Conclusion, ...]' = ()) -> None

CONSTRUCTION_FORMAT_VERSION = 11

CONSTRUCTION_SEMANTICS_VERSION = 12

decide(knowledge: 'KnowledgeBase | CompactIR | bytes | bytearray | memoryview', request: 'DecisionRequest', *, facts: 'Iterable[Fact]' = (), budget: 'ExecutionBudget | None' = None, max_firings: 'int | None' = None, limits: 'DecisionLimits | None' = None) -> 'DecisionResult'

DECISION_FORMAT_VERSION = 7

DECISION_LIBRARY_FORMAT_VERSION = 2

DECISION_POLICY = 'unique-feasible-v1'

DECISION_SEMANTICS_VERSION = 8

DecisionDefinition(name: 'str', request: 'DecisionRequest') -> None

DecisionError(code: 'str', message: 'str', path: 'str' = '') -> 'None'

DecisionLimits(max_candidates: 'int' = 4096, max_requirements: 'int' = 256, max_inputs: 'int' = 100000, max_conclusions: 'int' = 100000, max_work: 'int' = 1000000) -> None

DecisionOutcome(*values)

DecisionRequest(goal: 'str', candidates: 'tuple[str, ...]', requirements: 'tuple[str, ...]', evidence_name: 'str', closed: 'bool', logical_time: 'int' = 0) -> None

DecisionResult(request: 'DecisionRequest', status: 'DecisionStatus', outcome: 'DecisionOutcome', candidates: 'tuple[CandidateAssessment, ...]', feasible_options: 'tuple[str, ...]', selected_option: 'str | None', reason_codes: 'tuple[str, ...]', usage: 'DecisionUsage', limits: 'DecisionLimits', session: 'SessionResult', exhausted_limit: 'str | None' = None) -> None

DecisionStatus(*values)

DecisionUsage(conclusions_examined: 'int' = 0, candidates_examined: 'int' = 0, requirement_checks: 'int' = 0) -> None

DeltaBatch(base_epoch: 'int', epoch: 'int', added: 'tuple[FactItem, ...]' = (), removed: 'tuple[FactItem, ...]' = (), replacements: 'tuple[ReplacementDelta, ...]' = ()) -> None

derive(schema: 'FactSchema', /, **values: 'Value | BindingRef') -> 'Action'

dump_rulespec(spec: 'RuleSpec', path: 'RuleSpecPath', *, limits: 'RuleSpecLimits | None' = None, overwrite: 'bool' = False) -> 'Path'

emit(name: 'str', /, **values: 'Value | BindingRef') -> 'Action'

Engine(knowledge: 'KnowledgeBase | CompactIR | bytes | bytearray | memoryview', *, firing_limit: 'int' = 10000, optimization: 'OptimizationPolicy | None' = None, maintenance: 'MaintenancePolicy | None' = None, memory: 'MemoryPolicy | None' = None) -> 'None'

EngineError

EngineSnapshot(epoch: 'int', fact_items: 'tuple[FactItem, ...]', conclusions: 'tuple[Conclusion, ...]', trace: 'tuple[TraceEvent, ...]', provenance: 'tuple[Provenance, ...]', conclusion_provenance: 'tuple[tuple[Conclusion, tuple[Provenance, ...]], ...]' = (), seed_names: 'tuple[tuple[str, FactHandle], ...]' = (), program_digest: 'str | None' = None, *, entity_schemas: 'tuple' = ()) -> None

ENTITY_FORMAT_VERSION = 9

ENTITY_LIBRARY_FORMAT_VERSION = 3

ENTITY_SEMANTICS_VERSION = 10

EntityIndexReport(entities: 'int', references: 'int', peak_reserved_entries: 'int', limit_entries: 'int | None', coverage: 'str' = 'one entry per entity plus one per reference edge; staged old/new overlap', excluded: 'str' = 'Python bytes, key payloads, facts, proofs, snapshots, other indexes and process RAM') -> None

EntityLink(handle: 'FactHandle', fact: 'Fact', field: 'str') -> None

EntityRef(target: 'str', key: 'tuple[int | str | bytes, ...]') -> None

EntitySchema(name, fields, /, *, key)

report = engine.entity_storage_report
print(report.live_index_shallow_bytes)
print(report.retained_index_generations)
print(report.as_dict())

eq(value: 'Value | BindingRef') -> 'Constraint'

EqualityIndexMemoryReport(indexed_fields: 'int', posting_buckets: 'int', singleton_buckets: 'int', multirow_buckets: 'int', row_reference_slots: 'int', estimated_posting_container_bytes: 'int', estimated_dictionary_bytes: 'int', estimated_total_container_bytes: 'int', coverage: 'str' = 'current equality posting dictionaries and posting containers', excluded: 'str' = 'keys and their values, row-reference integers, schema routes, identity indexes, optional pruning/join caches, update overlap, facts, proofs, snapshots, JIT and process RAM') -> None

EvidenceState(*values)

ExecutionBudget(max_candidate_rows: 'int | None' = None, max_constraint_evaluations: 'int | None' = None, max_activations: 'int | None' = None, max_firings: 'int | None' = None) -> None

ExecutionPolicy(mode: 'str' = 'inline', max_workers: 'int' = 2, max_in_flight: 'int' = 4, chunk_size: 'int' = 1, minimum_parallel_facts: 'int' = 256, start_method: 'str' = 'spawn', max_jobs: 'int' = 100000) -> None

exists(pattern: 'Pattern') -> 'ExistsCondition'

ExistsCondition(pattern: 'Pattern', negated: 'bool' = False) -> None

ExplanationError(code: 'str', message: 'str', hint: 'str') -> 'None'

ExplanationLimits(max_records: 'int' = 100000, max_proofs: 'int' = 1000, max_supports: 'int' = 10000, max_statements: 'int' = 10000) -> None

Fact(schema: 'FactSchema', values: 'tuple[Value, ...]') -> None

FactHandle(id: 'int') -> None

FactSchema(name: 'str', fields: 'Mapping[str, type | tuple[type, ...]] | Sequence[Field | NDField] | None' = None, /, **field_types: 'type | tuple[type, ...]') -> 'None'

FactSupport(handle: 'FactHandle', fact: 'Fact', schema: 'SourceOrigin', seeds: 'tuple[SourceOrigin, ...]' = ()) -> None

Field(name: 'str', type_: 'type | tuple[type, ...]') -> 'None'

FileBuildResult(source: 'str', output: 'str', digest: 'str | None', diagnostics: 'tuple[SourceDiagnostic, ...]' = ()) -> None

FileResolver(modules: 'Mapping[str, str | os.PathLike[str]]', *, limits: 'SourceLimits | None' = None) -> 'None'

for_all(pattern: 'Pattern') -> 'ForAllCondition'

ForAllCondition(pattern: 'Pattern', negated: 'bool' = False) -> None

format_source(source: 'str | SourceSnapshot | SourceDocument', *, limits: 'SourceLimits | None' = None) -> 'str'

FORMAT_VERSION = 4

ge(value: 'Value | BindingRef') -> 'Constraint'

gt(value: 'Value | BindingRef') -> 'Constraint'

HistoryCorruptionError

HistoryEntry(generation: 'int', revision: 'str', parent: 'str | None', restored_from: 'str | None', timestamp: 'str', message: 'str', operation: 'str', source_digest: 'str', semantic_digest: 'str', artifact_digest: 'str') -> None

HistoryError

HistoryExportError(source_path: 'Path', artifact_path: 'Path', *, source_available: 'bool', artifact_available: 'bool') -> 'None'

HistoryLimits(rulespec_limits: 'RuleSpecLimits' = RuleSpecLimits(max_source_bytes=33554432, max_depth=64, max_collection_items=100000, max_total_nodes=1000000, max_string_bytes=16777216, max_integer_digits=128, max_test_cases=1000, max_facts_per_test=10000), ir_limits: 'IRLimits' = IRLimits(max_total_bytes=16777216, max_body_bytes=8388608, max_integer_digits=128, max_symbols=100000, max_symbol_bytes=4096, max_constants=250000, max_schemas=4096, max_fields_per_schema=4096, max_rules=100000, max_patterns_per_rule=1024, max_constraints_per_pattern=4096, max_operands_per_constraint=4096, max_actions_per_rule=1024, max_values_per_action=4096, max_bindings_per_rule=4096, max_instruction_fuel=5000000, max_nd_rank=8, max_nd_dimension=1000000, max_nd_elements=1000000, max_nd_bytes=8388608, max_seed_facts=65536, max_seed_values=1000000, max_classifications=4096, max_classification_edges=16384, max_classification_depth=64, max_decisions=4096, max_decision_items=100000), max_source_bytes: 'int' = 33554432, max_manifest_bytes: 'int' = 1048576, max_generation_bytes: 'int' = 262144, max_test_report_bytes: 'int' = 4194304, max_history_entries: 'int' = 100000, max_message_bytes: 'int' = 16384, max_test_cases: 'int' = 1000, test_budget: 'ExecutionBudget' = ExecutionBudget(max_candidate_rows=1000000, max_constraint_evaluations=1000000, max_activations=100000, max_firings=10000)) -> None

HistoryRestoreExportError(entry: 'HistoryEntry', source_path: 'Path', artifact_path: 'Path', *, source_available: 'bool', artifact_available: 'bool') -> 'None'

HistoryStore(root: 'HistoryPath', *, limits: 'HistoryLimits | None' = None) -> 'None'

HistoryTestFailure(failed_count: 'int', report: 'Mapping[str, Any]') -> 'None'

ImmutableKnowledgeError(handle_id: 'int', names: 'tuple[str, ...]') -> 'None'

InferenceJob(facts: 'tuple[Fact, ...]', max_firings: 'int | None' = None, budget: 'ExecutionBudget | None' = None, memory: 'MemoryPolicy | None' = None) -> None

IRLimits(max_total_bytes: 'int' = 16777216, max_body_bytes: 'int' = 8388608, max_integer_digits: 'int' = 128, max_symbols: 'int' = 100000, max_symbol_bytes: 'int' = 4096, max_constants: 'int' = 250000, max_schemas: 'int' = 4096, max_fields_per_schema: 'int' = 4096, max_rules: 'int' = 100000, max_patterns_per_rule: 'int' = 1024, max_constraints_per_pattern: 'int' = 4096, max_operands_per_constraint: 'int' = 4096, max_actions_per_rule: 'int' = 1024, max_values_per_action: 'int' = 4096, max_bindings_per_rule: 'int' = 4096, max_instruction_fuel: 'int' = 5000000, max_nd_rank: 'int' = 8, max_nd_dimension: 'int' = 1000000, max_nd_elements: 'int' = 1000000, max_nd_bytes: 'int' = 8388608, max_seed_facts: 'int' = 65536, max_seed_values: 'int' = 1000000, max_classifications: 'int' = 4096, max_classification_edges: 'int' = 16384, max_classification_depth: 'int' = 64, max_decisions: 'int' = 4096, max_decision_items: 'int' = 100000) -> None

IRValidationError

iter_batch(knowledge: 'KnowledgeBase | CompactIR | bytes | bytearray | memoryview', jobs: 'Iterable[InferenceJob]', *, policy: 'ExecutionPolicy | None' = None, executor: 'Executor | None' = None, parallel_budget: 'ParallelBudget | None' = None, cancel: 'Any' = None, optimization: 'OptimizationPolicy | None' = None, firing_limit: 'int' = 10000) -> 'Iterator[SessionResult]'

KnowledgeBase(schemas: 'Iterable[FactSchema]' = (), rules: 'Iterable[Rule]' = (), *, seeds: 'Iterable[SeedFact]' = (), classifications: 'Iterable' = (), decisions: 'Iterable' = ()) -> 'None'

le(value: 'Value | BindingRef') -> 'Constraint'

LIBRARY_FORMAT_VERSION = 1

LIBRARY_MAGIC = b'tlib'

LibraryBuild(library: 'CompiledLibrary', analysis: 'Any', tests: 'Any', output: 'Path | None' = None) -> None

LibraryError(code: 'str', message: 'str', hint: 'str', chain: 'tuple[str, ...]' = ())

LibraryExport(kind: 'str', name: 'str', qualified_name: 'str') -> None

LibraryLimits(max_file_bytes: 'int' = 134217728, max_index_bytes: 'int' = 4194304, max_section_bytes: 'int' = 20971520, max_sections: 'int' = 4096, max_modules: 'int' = 64, max_selected_bytes: 'int' = 67108864, max_build_bytes: 'int' = 134217728, max_dependencies: 'int' = 100000, max_dependency_depth: 'int' = 128, max_source_nodes: 'int' = 100000) -> None

LinkedBuild(program: 'CompactIR', knowledge_base: 'KnowledgeBase', analysis: 'LinkedSourceMap', imports: 'tuple[tuple[str, str, str], ...]', usage: 'LinkUsage') -> None

LinkedSourceMap(program_digest: 'str', semantic_digest: 'str', origins: 'tuple[SourceOrigin, ...]', nodes: 'tuple[SourceNode, ...]', source_identities: 'tuple[tuple[str, str], ...]') -> None

LinkUsage(modules: 'int', sections: 'int', directory_bytes: 'int', section_bytes: 'int', programs: 'int' = 0, program_bytes: 'int' = 0) -> None

load_rulespec(path: 'RuleSpecPath', *, limits: 'RuleSpecLimits | None' = None) -> 'RuleSpec'

load_ting(path: 'TingPath', *, limits: 'IRLimits | None' = None) -> 'CompactIR'

load_tingir(path: 'str | os.PathLike[str]', *, limits: 'IRLimits | None' = None, library_limits: 'LibraryLimits | None' = None) -> 'CompactIR | CompiledLibrary'

lt(value: 'Value | BindingRef') -> 'Constraint'

MaintenanceLimitError(limit: 'str', maximum: 'int', used: 'int')

MaintenancePolicy(enabled: 'bool' = True, max_facts: 'int' = 100000, max_proofs: 'int' = 200000, max_supports: 'int' = 1000000, max_work: 'int' = 4000000, max_changes: 'int' = 1000, on_limit: 'str' = 'recompute') -> None

MaintenanceReport(mode: 'str', reason: 'str', retained_facts: 'int' = 0, withdrawn_facts: 'int' = 0, retained_conclusions: 'int' = 0, withdrawn_conclusions: 'int' = 0, retained_proofs: 'int' = 0, removed_proofs: 'int' = 0, work: 'int' = 0, changes: 'tuple[SupportChange, ...]' = (), changes_truncated: 'bool' = False) -> None

MatchMetrics(index_probes: 'int' = 0, indexed_rows: 'int' = 0, full_scan_rows: 'int' = 0, constraint_evaluations: 'int' = 0, activations_generated: 'int' = 0) -> None

MemoryLedger(policy: 'MemoryPolicy | None' = None)

MemoryPolicy(max_managed_bytes: 'int' = 67108864, max_numeric_work: 'int' = 100000000, *, max_store_buffer_bytes: 'int | None' = None, max_entity_index_entries: 'int | None' = None) -> None

report = engine.memory_report
# Or, after a numeric_run() block:
report = usage.report()

MemoryResolver(modules: 'Mapping[str, str | SourceSnapshot]') -> 'None'

NDDType(*values)

NDField(name: 'str', dtype: 'NDDType | str', shape: 'Iterable[int]') -> 'None'

NDScalar = bool | int | float

NDValue() -> 'None'

ne(value: 'Value | BindingRef') -> 'Constraint'

new_rulespec(name: 'str', description: 'str' = '') -> 'RuleSpec'

normalized_source_view(spec: 'RuleSpec') -> 'dict[str, Any]'

not_(condition: 'Condition') -> 'Condition'

numeric_run(policy: 'MemoryPolicy | None' = None)

one_of(*values: 'Value | BindingRef') -> 'Constraint'

OptimizationMetrics(index_builds: 'int' = 0, index_hits: 'int' = 0, build_rows: 'int' = 0, pruned_rows: 'int' = 0, empty_branches: 'int' = 0, fallbacks: 'int' = 0, cached_cells: 'int' = 0) -> None

OptimizationPolicy(enabled: 'bool' = True, minimum_rows: 'int' = 32, max_indexes: 'int' = 32, max_cache_cells: 'int' = 1048576, max_build_work: 'int' = 4194304) -> None

ParallelBudget(capacity: 'int') -> 'None'

parse_rulespec(source: 'RuleSpecSource', *, limits: 'RuleSpecLimits | None' = None) -> 'RuleSpec'

parse_source(source: 'str | SourceSnapshot', *, uri: 'str' = '<memory>', limits: 'SourceLimits | None' = None) -> 'SourceDocument'

Pattern(schema: 'FactSchema', constraints: 'tuple[tuple[int, Constraint], ...]', negated: 'bool' = False) -> None

Provenance(subject_kind: 'str', subject: 'int | str', rule: 'str', fact_ids: 'tuple[int, ...]', bindings: 'tuple[tuple[str, Value], ...]') -> None

read_source(path: 'str | os.PathLike[str]', *, limits: 'SourceLimits | None' = None) -> 'SourceSnapshot'

ReferenceField(name: 'str', target: 'str') -> None

ReplacementDelta(before_handle: 'FactHandle', before_fact: 'Fact', after_handle: 'FactHandle', after_fact: 'Fact', changed_mask: 'int | None' = None) -> None

ResourceLimitError(resource: 'str', requested: 'int', limit: 'int')

RestoreResult(entry: 'HistoryEntry', source_path: 'Path', artifact_path: 'Path') -> None

Rule(name: 'str', when: 'Iterable[Pattern] | Condition', then: 'Iterable[Action]', *, salience: 'int' = 0, namespace: 'str' = 'default', version: 'str' = '1') -> 'None'

RuleError

RuleSpec(project_name: 'str', description: 'str', knowledge_base: 'KnowledgeBase', tests: 'tuple[RuleSpecTestCase, ...]' = ()) -> None

RULESPEC_FORMAT = 'ting-rulespec'

RULESPEC_FORMAT_VERSION = 1

RuleSpecConclusionExpectation(name: 'str', values: 'tuple[tuple[str, Value], ...]' = ()) -> None

RuleSpecError

RuleSpecLimits(max_source_bytes: 'int' = 33554432, max_depth: 'int' = 64, max_collection_items: 'int' = 100000, max_total_nodes: 'int' = 1000000, max_string_bytes: 'int' = 16777216, max_integer_digits: 'int' = 128, max_test_cases: 'int' = 1000, max_facts_per_test: 'int' = 10000) -> None

RuleSpecTestCase(name: 'str', facts: 'tuple[Fact, ...]' = (), expected_conclusions: 'tuple[RuleSpecConclusionExpectation, ...] | None' = None, expected_facts: 'tuple[Fact, ...] | None' = None, expected_status: 'str | None' = 'quiescent', expected_firings: 'int | None' = None) -> None

RuleSpecTestReport(results: 'tuple[RuleSpecTestResult, ...]', budget: 'ExecutionBudget') -> None

RuleSpecTestResult(name: 'str', passed: 'bool', status: 'str', firings: 'int', failures: 'tuple[str, ...]', facts: 'tuple[Fact, ...]', conclusions: 'tuple[Conclusion, ...]', budget_usage: 'BudgetUsage', exhausted_budget: 'str | None' = None) -> None

run_batch(knowledge: 'KnowledgeBase | CompactIR | bytes | bytearray | memoryview', jobs: 'Iterable[InferenceJob]', *, policy: 'ExecutionPolicy | None' = None, executor: 'Executor | None' = None, parallel_budget: 'ParallelBudget | None' = None, cancel: 'Any' = None, optimization: 'OptimizationPolicy | None' = None, firing_limit: 'int' = 10000) -> 'BatchResult'

run_rulespec_tests(spec: 'RuleSpec', *, ir_limits: 'IRLimits | None' = None, budget: 'ExecutionBudget | None' = None, rulespec_limits: 'RuleSpecLimits | None' = None) -> 'RuleSpecTestReport'

RunResult(status: 'RunStatus', firings: 'int', derived_facts: 'int', conclusions: 'int', trace_events: 'int', budget_usage: 'BudgetUsage' = BudgetUsage(candidate_rows=0, constraint_evaluations=0, activations=0, firings=0), exhausted_budget: 'str | None' = None) -> None

RunStatus(*values)

RuntimePlan(digest: 'str', payload_digest: 'str', symbols: 'tuple[str, ...]', constants: 'tuple[Any, ...]', schemas: 'tuple[RuntimeSchema, ...]', rules: 'tuple[RuntimeRule, ...]', instruction_fuel: 'int', seeds: 'tuple[RuntimeSeed, ...]' = (), format_version: 'int' = 4, semantics_version: 'int' = 5, classifications: 'tuple[RuntimeClassification, ...]' = (), decisions: 'tuple[DecisionDefinition, ...]' = ()) -> None

save_rulespec(spec: 'RuleSpec', path: 'RuleSpecPath', *, limits: 'RuleSpecLimits | None' = None, overwrite: 'bool' = False) -> 'Path'

save_ting(source: 'TingArtifactSource', path: 'TingPath', *, limits: 'IRLimits | None' = None, overwrite: 'bool' = False) -> 'Path'

save_tingir(program: 'KnowledgeBase | CompactIR | CompiledLibrary | bytes | bytearray | memoryview', path: 'str | os.PathLike[str]', *, overwrite: 'bool' = False, limits: 'IRLimits | None' = None) -> 'Path'

SchemaError

SEED_FORMAT_VERSION = 5

SEED_SEMANTICS_VERSION = 6

SeedFact(name: 'str', fact: 'Fact') -> None

semantic_digest(spec: 'RuleSpec') -> 'str'

semantic_view(spec: 'RuleSpec') -> 'dict[str, Any]'

SEMANTICS_VERSION = 5

SessionResult(sequence: 'int', run: 'RunResult', snapshot: 'EngineSnapshot', match_metrics: 'MatchMetrics', optimization_metrics: 'OptimizationMetrics', memory_report: 'MemoryReport | None' = None, *, store_allocation_report: 'StoreAllocationReport | None' = None) -> None

SHARED_ENTITY_LIBRARY_FORMAT_VERSION = 4

source_digest(spec: 'RuleSpec', *, limits: 'RuleSpecLimits | None' = None) -> 'str'

SOURCE_FEATURES = frozenset({'array-sum', 'arrays', 'entity-traversal', 'decisions', 'entity-references', 'entity-construction', 'rules-core', 'seed-facts', 'classifications'})

SOURCE_FORMAT_VERSION = 1

SOURCE_SEMANTICS_VERSION = 1

SourceAnalysis(root: 'SourceModule', modules: 'tuple[SourceModule, ...]', knowledge_base: 'KnowledgeBase', test_spec: 'RuleSpec', semantic_digest: 'str', origins: 'tuple[SourceOrigin, ...]', program_digest: 'str' = '') -> None

SourceBuild(program: 'CompactIR', analysis: 'SourceAnalysis', tests: 'RuleSpecTestReport | None', output: 'Path | None' = None) -> None

SourceDiagnostic(code: 'str', message: 'str', hint: 'str', span: 'SourceSpan | None' = None, node_id: 'str | None' = None, severity: 'str' = 'error') -> None

SourceDocument(snapshot: 'SourceSnapshot', exact_digest: 'str', module: 'str', version: 'str', role: 'str', features: 'tuple[str, ...]', regions: 'tuple[SourceNode, ...]') -> None

SourceError(diagnostic: 'SourceDiagnostic', *, test_report: 'Any' = None) -> 'None'

SourceExplanation(program_digest: 'str', semantic_digest: 'str', sources: 'tuple[tuple[str, str], ...]', epoch: 'int', subject: 'FactSupport | Conclusion | None', proofs: 'tuple[SourceProof, ...]') -> None

SourceLimits(max_source_bytes: 'int' = 1048576, max_total_bytes: 'int' = 8388608, max_modules: 'int' = 64, max_lines: 'int' = 50000, max_line_bytes: 'int' = 16384, max_nodes: 'int' = 100000, max_depth: 'int' = 32, max_integer_digits: 'int' = 128) -> None

SourceModule(document: 'SourceDocument', schemas: 'tuple[tuple[str, FactSchema], ...]', conclusions: 'tuple[tuple[str, FactSchema], ...]', rules: 'tuple[tuple[str, Rule], ...]', exports: 'tuple[tuple[str, str], ...]', imported_rules: 'tuple[Rule, ...]', tests: 'tuple[RuleSpecTestCase, ...]', origins: 'tuple[SourceOrigin, ...]', seeds: 'tuple[tuple[str, SeedFact], ...]' = (), seed_tests: 'tuple[str, ...]' = (), classifications: 'tuple[tuple[str, Classification], ...]' = (), requirements: 'tuple[tuple[str, tuple[tuple[str, str], ...]], ...]' = (), decisions: 'tuple[tuple[str, DecisionDefinition], ...]' = ()) -> None

SourceNode(text: 'str', span: 'SourceSpan', node_id: 'str', children: 'tuple[SourceNode, ...]' = ()) -> None

SourceOrigin(kind: 'str', name: 'str', node_id: 'str', span: 'SourceSpan') -> None

SourceProof(provenance: 'Provenance', rule: 'SourceOrigin', statements: 'tuple[SourceStatement, ...]', supports: 'tuple[FactSupport, ...]') -> None

SourceResolver(*args, **kwargs)

SourceSnapshot(text: 'str', uri: 'str' = '<memory>', revision: 'str' = '') -> None

SourceSpan(uri: 'str', source_digest: 'str', line: 'int', column: 'int', end_column: 'int') -> None

SourceStatement(role: 'str', text: 'str', node_id: 'str', span: 'SourceSpan') -> None

StagedHandle(ordinal: 'int', _owner: 'object') -> None

StaleHistoryError(expected_head: 'str | None', current_head: 'str | None') -> 'None'

StaleTransactionError(expected_epoch: 'int', current_epoch: 'int') -> 'None'

StoreAllocationReport(current_buffer_payload_bytes: 'int', peak_reserved_buffer_payload_bytes: 'int', staged_buffer_payload_bytes: 'int', allocation_reservations: 'int', limit_bytes: 'int | None', coverage: 'str' = 'session numeric/byte store columns and row metadata, including staged copy/growth reservations', excluded: 'str' = 'container headers/spare capacity, object-column values, routes/indexes, proofs, snapshots, error tracebacks, plan, numeric work buffers, JIT and process RAM') -> None

report = engine.store_compaction_report
if report is not None:
    print(report.overlap_buffer_payload_bytes)
    print(report.as_dict())

report = engine.store_memory_report
print(report.current_buffer_payload_bytes)
print(report.as_dict())

StoreMetrics(chunk_size: 'int', row_chunks: 'int', column_chunks: 'int', row_capacity: 'int', live_base_rows: 'int', live_derived_rows: 'int', tombstone_rows: 'int', compactions: 'int') -> None

SupportChange(subject: 'Fact | Conclusion', status: 'str', previous_supports: 'int', remaining_supports: 'int', previous_handle: 'FactHandle | None' = None, current_handle: 'FactHandle | None' = None) -> None

test(operator: 'str', *operands: 'Value | BindingRef') -> 'TestCondition'

TestCondition(operator: 'str', operands: 'tuple[Value | BindingRef, ...]', negated: 'bool' = False) -> None

TING_FILE_SUFFIX = '.tingir'

TING_MAGIC = b'ting'

TingError

TraceEvent(sequence: 'int', kind: 'str', rule: 'str', fact_ids: 'tuple[int, ...]', bindings: 'tuple[tuple[str, Value], ...]', subject: 'int | str | None' = None) -> None

Transaction(*args, **kwargs)

report = receipt.buffer_report
if report is not None:
    print(report.peak_buffer_payload_bound_bytes)
    print(report.as_dict())

TRAVERSAL_FORMAT_VERSION = 10

TRAVERSAL_SEMANTICS_VERSION = 11

use(binding: 'str | BindingRef') -> 'BindingRef'

ValidationError

Value = None | bool | int | float | str | bytes | ting_reasoner.nd.NDValue | ting_reasoner.entities.EntityRef

verify_compact_ir(data: 'bytes | bytearray | memoryview', *, limits: 'IRLimits | None' = None) -> 'RuntimePlan'