scripts.rts

Reference › scripts › rts

scripts/rts.py does AST-based safe regression test selection over a Prolog knowledge base; it is a handover artifact, covered by tests/scripts/test_rts.py. A human or a CI step runs it. It has two positional-argument commands rather than options.

rts.py extract <source-dir> <out.pl> calls extract, which writes Prolog facts for the source tree into the output file, including per-symbol hashes (sym_hash) and per-file hashes (file_hash). rts.py select <old.pl> <new.pl> reads two such fact files with parse_pred, and works out the changed symbols: edited, added or deleted. Any changed, added or removed non-Python file adds a safety_escape fact, which forces a full run. It then writes a temporary Prolog program made of the selection rules, the new fact base, the escapes and the changed/1 facts, prints the changed symbols, and runs it with swipl, which prints the selection. Wrong usage prints the module summary and exits 2.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

No internal imports and no importers.

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/rts.py

#!/usr/bin/env python3
"""AST-based safe regression test selection over a Prolog KB (handover artifact).

    python scripts/rts.py extract <root> <snapshot.pl>   symbol/call/hash facts from AST
    python scripts/rts.py select  <old.pl> <new.pl>      diff hashes -> changed symbols
                                                         -> tabled reverse reachability
                                                         -> minimal safe test set (stdout)

Selection criterion (Rothermel-Harrold safe over-approximation): a test is
selected iff it is changed itself, or a changed symbol is in its transitive
call/import closure. Safety escapes (conftest, unparsed files, unresolved
dynamic calls) force full-run and are printed as such.

Provenance: delivered by the 2026-07-31 orchestrata handover
(~/workspace/orchestrata-handover/rts.py), verified there on 4 behavioral
cases + ground truth (selected set == actually-failing set, zero misses).
Re-verified on this host 2026-08-02. Installed 2026-08-02 per the tooling
wiring plan (plans/tooling-wiring.md).

Relationship to scripts/affected_tests.py (the WIRED selector behind
``run_tests.py --affected``): affected_tests works at MODULE granularity via
the serialized import closure + manifest + full-suite escalation; this tool
works at SYMBOL granularity via per-function AST source hashes (formatting
blind). The handover's Phase 7 merges this precision INTO affected_tests —
until that lands (gated on user authorization), this stays a standalone
second opinion: run it when a module-level selection looks too broad.

Defects fixed on install (2026-08-02), diverging from the handover artifact:
1. Modules were keyed by file STEM, so package-qualified call edges
   ("src.calc.add") never matched defined symbols ("calc.add") — silent
   UNSAFE under-selection on any package layout. Modules are now keyed by
   dotted path from root; relative imports resolve against the importing
   package (unresolvable ones become safety escapes).
2. Only top-level functions were hashed — methods, classes, nested defs,
   decorators, imports, constants, and module-level statements were captured
   by NOTHING (edits there selected zero tests). Capture is now
   BYTE-COMPLETE (user mandate): every def/class at any depth is hashed on
   its exact source segment, and a synthetic ``<mod>.<module>`` symbol
   hashes the residue (everything outside top-level def/class spans), with
   dependency edges symbol→its module and module→imported modules. Any
   single-byte change in any .py file changes at least one hash.
Tests are detected under a ``tests/`` directory (functions/methods named
``test_*``).
"""

import ast
import hashlib
import pathlib
import subprocess
import sys
import tempfile


def q(s: object) -> str:  # quote as Prolog atom
    return "'" + str(s).replace("\\", "\\\\").replace("'", "\\'") + "'"


def extract(root: pathlib.Path) -> str:
    facts: list[str] = []
    escapes: list[str] = []
    for path in sorted(root.rglob("*.py")):
        rel = path.relative_to(root)
        # Key modules by their dotted import path from root (src.calc, not
        # the bare stem "calc"). The original handover artifact used
        # path.stem, so defined symbols ("calc.add") could NEVER match
        # package-qualified call edges ("src.calc.add") — silent UNSAFE
        # under-selection. Found by smoke test on install (2026-08-02).
        mod = ".".join(rel.with_suffix("").parts)
        pkg_parts = rel.parts[:-1]
        try:
            src = path.read_text()
            tree = ast.parse(src)
        except SyntaxError:
            escapes.append(f"unparsed:{rel}")
            continue
        if path.name == "conftest.py":
            escapes.append(f"conftest:{rel}")

        imports: dict[str, str] = {}  # local name -> qualified target
        for node in ast.walk(tree):
            if isinstance(node, ast.Import):
                for a in node.names:
                    imports[a.asname or a.name] = a.name
            elif isinstance(node, ast.ImportFrom):
                if node.level == 0:
                    base = node.module or ""
                else:  # relative import: resolve against this module's package
                    drop = node.level - 1
                    if drop > len(pkg_parts):
                        escapes.append(f"unresolved-relative-import:{rel}")
                        continue
                    kept = pkg_parts[: len(pkg_parts) - drop]
                    base = ".".join((*kept, node.module) if node.module else kept)
                for a in node.names:
                    imports[a.asname or a.name] = f"{base}.{a.name}" if base else a.name

        local_fns = {
            n.name
            for n in tree.body
            if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef))
        }

        # BYTE-COMPLETE CAPTURE (user mandate 2026-08-02: "AST must capture
        # byte-identical code, all of the code"). Every byte of the file lands
        # in exactly one hash bucket:
        #   1. every def/async def/class at ANY depth -> sym_hash of its exact
        #      source segment (qualified name; methods/nested defs included);
        #   2. a synthetic "<mod>.<module>" symbol hashes the RESIDUE — the
        #      file with top-level def/class line-spans blanked (imports,
        #      constants, decorators, module-level statements);
        #   3. a segment the AST cannot source byte-exactly is a safety
        #      escape (FULL-RUN), never a silent empty hash.
        # Propagation: every symbol depends on its module's <module> symbol;
        # a module depends on every imported module's <module> symbol (all
        # dotted prefixes — over-approximation is SAFE, nonexistent prefixes
        # never match a changed fact).
        mod_sym = f"{mod}.<module>"
        facts.append(f"defines({q(mod)}, {q(mod_sym)}).")

        def record_call_edges(qname: str, node: ast.AST) -> None:
            for c in ast.walk(node):
                if not isinstance(c, ast.Call):
                    continue
                f, callee = c.func, None
                if isinstance(f, ast.Name):
                    if f.id in imports:
                        callee = imports[f.id]
                    elif f.id in local_fns:
                        callee = f"{mod}.{f.id}"
                elif (
                    isinstance(f, ast.Attribute)
                    and isinstance(f.value, ast.Name)
                    and f.value.id in imports
                ):
                    callee = f"{imports[f.value.id]}.{f.attr}"
                elif isinstance(f, ast.Attribute):
                    pass  # method call on object: covered by module-level edges
                else:
                    escapes.append(f"dynamic-call:{qname}")  # getattr etc.
                if callee:
                    facts.append(f"calls({q(qname)}, {q(callee)}).")

        def walk_defs(body: list[ast.stmt], prefix: str) -> None:
            for node in body:
                if not isinstance(
                    node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)
                ):
                    continue
                qname = f"{prefix}.{node.name}"
                segment = ast.get_source_segment(src, node)
                if segment is None:
                    escapes.append(f"unhashable:{qname}")
                else:
                    h = hashlib.sha256(segment.encode()).hexdigest()[:16]
                    facts.append(f"defines({q(mod)}, {q(qname)}).")
                    facts.append(f"sym_hash({q(qname)}, {q(h)}).")
                facts.append(f"calls({q(qname)}, {q(mod_sym)}).")
                if "tests" in rel.parts and node.name.startswith("test_"):
                    facts.append(f"is_test({q(qname)}).")
                if isinstance(node, ast.ClassDef):
                    walk_defs(node.body, qname)
                else:
                    record_call_edges(qname, node)
                    walk_defs(node.body, qname)  # nested defs get own hashes

        walk_defs(tree.body, mod)

        # Residue: the file with TOP-LEVEL def/class line-spans blanked.
        # Decorator lines precede node.lineno, so they stay in the residue.
        lines = src.splitlines(keepends=True)
        blanked: set[int] = set()
        for node in tree.body:
            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
                end = node.end_lineno or node.lineno
                blanked.update(range(node.lineno - 1, end))
        residue = "".join(ln for i, ln in enumerate(lines) if i not in blanked)
        rh = hashlib.sha256(residue.encode()).hexdigest()[:16]
        facts.append(f"sym_hash({q(mod_sym)}, {q(rh)}).")
        for stmt in tree.body:  # module-level calls only — defs have their own
            if not isinstance(
                stmt, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)
            ):
                record_call_edges(mod_sym, stmt)

        # Module import edges: <module> -> every dotted prefix's <module>.
        for target in set(imports.values()):
            parts = target.split(".")
            for i in range(1, len(parts) + 1):
                pref = ".".join(parts[:i])
                if pref != mod:
                    facts.append(f"calls({q(mod_sym)}, {q(pref + '.<module>')}).")
    # NON-PYTHON capture ("all of the code, leave nothing behind"): every
    # non-.py file gets a byte-exact file_hash fact. select treats any
    # changed/added/removed non-Python file as a safety escape (FULL-RUN) —
    # .pl rule files, YAML, hooks, templates are code this tool cannot
    # attribute to symbols, so it must never silently ignore them.
    skip_dirs = {
        ".git",
        "__pycache__",
        ".venv",
        "node_modules",
        ".pytest_cache",
        ".ruff_cache",
        ".mypy_cache",
    }
    for path in sorted(root.rglob("*")):
        if not path.is_file() or path.suffix == ".py":
            continue
        rel = path.relative_to(root)
        if skip_dirs.intersection(rel.parts) or rel.parts[:2] == (".claude", "run"):
            continue  # VCS/caches/venvs + volatile runtime dir only
        h = hashlib.sha256(path.read_bytes()).hexdigest()[:16]
        facts.append(f"file_hash({q(rel)}, {q(h)}).")

    facts.extend(f"safety_escape({q(e)})." for e in sorted(set(escapes)))
    return (
        "\n".join(
            f":- discontiguous({p}).\n"
            for p in [
                "defines/2",
                "sym_hash/2",
                "is_test/1",
                "calls/2",
                "safety_escape/1",
                "file_hash/2",
            ]
        )
        + "\n".join(sorted(set(facts)))
        + "\n"
    )


SELECT_PL = r"""
:- dynamic calls/2 as incremental.
:- dynamic changed/1 as incremental.
:- table reaches/2 as incremental.
:- discontiguous(defines/2). :- discontiguous(sym_hash/2).
:- discontiguous(is_test/1). :- discontiguous(safety_escape/1).
:- discontiguous(file_hash/2).
:- dynamic defines/2, sym_hash/2, is_test/1, safety_escape/1, file_hash/2.

reaches(A, B) :- calls(A, B).
reaches(A, B) :- calls(A, M), reaches(M, B).

select_test(T) :- is_test(T), changed(T).
select_test(T) :- is_test(T), changed(S), reaches(T, S).

main :-
    ( safety_escape(E) -> format("FULL-RUN ~w~n", [E])
    ; forall(( is_test(T), ( select_test(T) -> V = 'SELECT' ; V = 'skip' )),
             format("~w ~w~n", [V, T]))
    ).
"""


def parse_pred(text: str, pred: str) -> dict[str, str]:
    """Read binary facts ``pred('Key', 'Val').`` back into a dict."""
    out: dict[str, str] = {}
    prefix = pred + "("
    for line in text.splitlines():
        if line.startswith(prefix):
            key, val = line[len(prefix) : -2].split(", ")
            out[key] = val
    return out


def main() -> int:
    if len(sys.argv) < 4 or sys.argv[1] not in ("extract", "select"):
        print((__doc__ or "").split("\n\n")[0].strip(), file=sys.stderr)
        return 2
    cmd = sys.argv[1]
    if cmd == "extract":
        pathlib.Path(sys.argv[3]).write_text(extract(pathlib.Path(sys.argv[2])))
    else:  # select
        old_text = pathlib.Path(sys.argv[2]).read_text()
        new_text = pathlib.Path(sys.argv[3]).read_text()
        old = parse_pred(old_text, "sym_hash")
        new = parse_pred(new_text, "sym_hash")
        # any changed/added/removed NON-PYTHON file forces FULL-RUN
        fo = parse_pred(old_text, "file_hash")
        fn = parse_pred(new_text, "file_hash")
        file_escapes = "".join(
            f"safety_escape(changed_file({p})).\n"
            for p in sorted(
                {p for p, h in fn.items() if fo.get(p) != h} | (set(fo) - set(fn))
            )
        )
        changed = {s for s, h in new.items() if old.get(s) != h} | (
            set(old) - set(new)
        )  # edited/added | deleted
        script = (
            SELECT_PL
            + new_text
            + file_escapes
            + "".join(f"changed({s}).\n" for s in sorted(changed))
        )
        with tempfile.NamedTemporaryFile(
            "w", suffix=".pl", prefix="rts_select_", delete=False
        ) as fh:
            fh.write(script)
            tmp = fh.name
        print("changed:", ", ".join(sorted(changed)) or "(none)")
        subprocess.run(["swipl", "-g", "main", "-t", "halt", tmp], check=True)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())