LIBRARY GUIDE · 0.8.5

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.

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 or later 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 --no-deps .\ting_reasoner-0.8.5-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 --no-deps ./ting_reasoner-0.8.5-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. Ting needs no C compiler or additional runtime packages.

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.

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.

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())

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 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 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. Broadcasting, reductions, masks and general array arithmetic are not supported. Scalar Python integers and array i64 elements deliberately have different range contracts.

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.

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.

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.8.5'

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

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], ...]' = ()) -> 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

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) -> '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) -> None

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

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) -> 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') -> 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

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'

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'

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

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') -> None

source_digest(spec: 'RuleSpec', *, limits: 'RuleSpecLimits | None' = None) -> 'str'

SOURCE_FEATURES = frozenset({'arrays', 'seed-facts', 'classifications', 'rules-core', 'decisions'})

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'

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)

use(binding: 'str | BindingRef') -> 'BindingRef'

ValidationError

Value = None | bool | int | float | str | bytes | ting_reasoner.nd.NDValue

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