smartbotic/scripts/characterise-loops.py

#!/usr/bin/env python3
"""Characterisation tests for loop execution.

Runs a set of workflows that exercise the loop walk and writes a normalised
summary of what each one did. The point is NOT that the recorded behaviour is
correct - it is to notice when a change to the engine alters any of it, so
every difference has to be looked at and justified rather than discovered
later in production.

    scripts/characterise-loops.py before.json     # before your change
    ... rebuild the runner and restart it ...
    scripts/characterise-loops.py after.json      # after it
    diff before.json after.json

This exists because the node suite cannot catch this class of change. When the
two graph walks were unified (2026-08-13), four behaviours differed between a
node inside a loop and the same node outside one; three of the four were
invisible to all 92 node tests, which passed green throughout. What caught them
was diffing these baselines.

Needs the webserver and a runner up, and logs in as admin/admin. Every workflow
it creates is named "zz char: ..." and is deleted again, including on failure.
"""
import json, sys, time, urllib.request

API = "http://localhost:8090/api/v1"


def req(method, path, body=None, token=None):
    data = json.dumps(body).encode() if body is not None else None
    r = urllib.request.Request(API + path, data=data, method=method,
                               headers={"Content-Type": "application/json"})
    if token:
        r.add_header("Authorization", "Bearer " + token)
    return json.loads(urllib.request.urlopen(r).read())


TOKEN = req("POST", "/auth/login", {"username": "admin", "password": "admin"})["accessToken"]


def conn(src, tgt, src_out="main", tgt_in="data"):
    return {"sourceNodeId": src, "sourceOutput": src_out,
            "targetNodeId": tgt, "targetInput": tgt_in}


def seed(items, node_id="seed"):
    return {"id": node_id, "type": "code", "name": node_id, "position": {"x": 0, "y": 0},
            "config": {"code": "return { items: %s }" % json.dumps(items)}}


def loop_node(node_id="loop", field="result.items", cont=True, x=240):
    return {"id": node_id, "type": "loop", "name": node_id, "position": {"x": x, "y": 0},
            "config": {"inputField": field, "continueOnError": cont}}


def setf(node_id, name, value, x=480, disabled=False):
    n = {"id": node_id, "type": "set-fields", "name": node_id, "position": {"x": x, "y": 0},
         "config": {"fields": [{"name": name, "value": value, "type": "string"}]}}
    if disabled:
        n["disabled"] = True
    return n


CASES = {}


def case(name):
    def deco(fn):
        CASES[name] = fn
        return fn
    return deco


@case("simple_three_items_two_body_nodes")
def _():
    return {
        "nodes": [
            seed([{"n": "a"}, {"n": "b"}, {"n": "c"}]),
            loop_node(),
            setf("first", "seen", "{{ loop.item.n }}"),
            # The item must still be readable on the SECOND body node.
            setf("second", "alsoSeen", "{{ loop.item.n }}-{{ loop.index }}", x=720),
            setf("after", "done", "yes", x=480),
        ],
        "connections": [
            conn("seed", "loop"), conn("loop", "first", "loop"),
            conn("first", "second"), conn("loop", "after", "done"),
        ],
    }


@case("nested_loop_two_by_two")
def _():
    return {
        "nodes": [
            seed([{"n": "x"}, {"n": "y"}]),
            loop_node("outer"),
            {"id": "inner_seed", "type": "code", "name": "inner_seed",
             "position": {"x": 480, "y": 0},
             "config": {"code": "return { sub: [{m:'1'}, {m:'2'}] }"}},
            loop_node("inner", "result.sub", x=720),
            setf("leaf", "pair", "{{ loop.item.m }}", x=960),
            setf("after", "done", "yes", x=480),
        ],
        "connections": [
            conn("seed", "outer"), conn("outer", "inner_seed", "loop"),
            conn("inner_seed", "inner"), conn("inner", "leaf", "loop"),
            conn("outer", "after", "done"),
        ],
    }


@case("branch_gating_in_body")
def _():
    return {
        "nodes": [
            seed([{"n": "a"}, {"n": "b"}]),
            loop_node(),
            setf("mark", "seen", "{{ loop.item.n }}"),
            {"id": "gate", "type": "if-condition", "name": "gate", "position": {"x": 720, "y": 0},
             "config": {"conditions": [{"field": "data.seen", "operator": "equals", "value": "a"}]}},
            setf("on_true", "took", "true", x=960),
            setf("on_false", "took", "false", x=960),
            setf("after", "done", "yes", x=480),
        ],
        "connections": [
            conn("seed", "loop"), conn("loop", "mark", "loop"), conn("mark", "gate"),
            conn("gate", "on_true", "true"), conn("gate", "on_false", "false"),
            conn("loop", "after", "done"),
        ],
    }


@case("back_edge_true_port_only")
def _():
    return {
        "nodes": [
            seed([{"n": "a"}, {"n": "b"}, {"n": "c"}]),
            loop_node(),
            setf("mark", "seen", "{{ loop.item.n }}"),
            {"id": "gate", "type": "if-condition", "name": "gate", "position": {"x": 720, "y": 0},
             "config": {"conditions": [{"field": "data.seen", "operator": "not_equals", "value": "a"}]}},
            setf("after", "done", "yes", x=480),
        ],
        "connections": [
            conn("seed", "loop"), conn("loop", "mark", "loop"), conn("mark", "gate"),
            conn("gate", "loop", "true"), conn("loop", "after", "done"),
        ],
    }


@case("disabled_node_in_body_is_transparent")
def _():
    return {
        "nodes": [
            seed([{"n": "a"}, {"n": "b"}]),
            loop_node(),
            setf("mark", "seen", "{{ loop.item.n }}"),
            setf("off", "ignored", "x", x=720, disabled=True),
            setf("tail", "reached", "yes", x=960),
            setf("after", "done", "yes", x=480),
        ],
        "connections": [
            conn("seed", "loop"), conn("loop", "mark", "loop"),
            conn("mark", "off"), conn("off", "tail"), conn("loop", "after", "done"),
        ],
    }


@case("body_failure_continue_on_error_true")
def _():
    return {
        "nodes": [
            seed([{"n": "a"}, {"n": "b"}, {"n": "c"}]),
            loop_node(cont=True),
            # NOTE: a code node's body cannot see `loop` - it is available to
            # expressions, not as a variable in the sandbox - so this throws on
            # every item, not only on 'b'. Left as it is because a baseline only
            # has to be stable and this one exercises "every item failed, run
            # completed anyway"; do not read the case name as a promise that
            # exactly one item fails.
            {"id": "boom", "type": "code", "name": "boom", "position": {"x": 480, "y": 0},
             "config": {"code": "if (input.item.n === 'b') { throw new Error('planned'); } return { ok: input.item.n }"}},
            setf("after", "done", "yes", x=480),
        ],
        "connections": [
            conn("seed", "loop"), conn("loop", "boom", "loop"), conn("loop", "after", "done"),
        ],
    }


@case("body_failure_continue_on_error_false")
def _():
    return {
        "nodes": [
            seed([{"n": "a"}, {"n": "b"}, {"n": "c"}]),
            loop_node(cont=False),
            {"id": "boom", "type": "code", "name": "boom", "position": {"x": 480, "y": 0},
             "config": {"code": "if (input.item.n === 'b') { throw new Error('planned'); } return { ok: input.item.n }"}},
            setf("after", "done", "yes", x=480),
        ],
        "connections": [
            conn("seed", "loop"), conn("loop", "boom", "loop"), conn("loop", "after", "done"),
        ],
    }


# continue_on_error is a per-item policy, not a per-run one. A loop that
# tolerates failures should still fail the run when NOTHING succeeded - a dead
# dependency made every item fail for hours while the run reported success.
# A code node reaches the loop item through `input` - `input.item`, `input.index`,
# `input.loop`. There is no bare `loop` global; using one throws "'loop' is not
# defined" on EVERY item, which is how two cases here spent their life asserting
# "one item fails" while actually testing "all of them do".
@case("tolerated_all_items_fail_fails_the_run")
def _():
    return {
        "nodes": [
            seed([{"n": "a"}, {"n": "b"}, {"n": "c"}]),
            loop_node(cont=True),
            {"id": "boom", "type": "code", "name": "boom", "position": {"x": 480, "y": 0},
             "config": {"code": "throw new Error('every item fails');"}},
            setf("after", "done", "yes", x=480),
        ],
        "connections": [
            conn("seed", "loop"), conn("loop", "boom", "loop"), conn("loop", "after", "done"),
        ],
    }


# The other half of the same rule: tolerating SOME failures must be unchanged.
@case("tolerated_some_items_fail_still_succeeds")
def _():
    return {
        "nodes": [
            seed([{"n": "a"}, {"n": "b"}, {"n": "c"}]),
            loop_node(cont=True),
            {"id": "boom", "type": "code", "name": "boom", "position": {"x": 480, "y": 0},
             "config": {"code": "if (input.item.n === 'b') { throw new Error('planned'); } return { ok: input.item.n }"}},
            setf("after", "done", "yes", x=480),
        ],
        "connections": [
            conn("seed", "loop"), conn("loop", "boom", "loop"), conn("loop", "after", "done"),
        ],
    }


def _halt_in_loop(mode):
    """A loop of three whose middle item trips a stop-and-error."""
    return {
        "nodes": [
            seed([{"n": "a"}, {"n": "b"}, {"n": "c"}]),
            loop_node(),
            setf("mark", "seen", "{{ loop.item.n }}"),
            {"id": "gate", "type": "if-condition", "name": "gate", "position": {"x": 720, "y": 0},
             "config": {"conditions": [{"field": "data.seen", "operator": "equals", "value": "b"}]}},
            {"id": "halt", "type": "stop-and-error", "name": "halt",
             "position": {"x": 960, "y": -80},
             "config": {"mode": mode, "message": "item b says stop"}},
            setf("keep", "kept", "yes", x=960),
            setf("after", "done", "yes", x=480),
        ],
        "connections": [
            conn("seed", "loop"), conn("loop", "mark", "loop"), conn("mark", "gate"),
            conn("gate", "halt", "true"), conn("gate", "keep", "false"),
            conn("loop", "after", "done"),
        ],
    }


# The three ways a body node can end something, which are three different
# things: fail the whole run, end the whole run cleanly, and drop this one item
# and carry on. error mode used to do none of them inside a loop - it threw, and
# the loop's Continue On Error caught it - so all three are pinned here.
@case("halt_in_loop_error_fails_the_run")
def _():
    return _halt_in_loop("error")


@case("halt_in_loop_stop_ends_the_run")
def _():
    return _halt_in_loop("stop")


@case("halt_in_loop_skip_drops_one_item")
def _():
    return _halt_in_loop("skip")


@case("empty_item_list")
def _():
    return {
        "nodes": [
            seed([]),
            loop_node(),
            setf("mark", "seen", "{{ loop.item.n }}"),
            setf("after", "done", "yes", x=480),
        ],
        "connections": [
            conn("seed", "loop"), conn("loop", "mark", "loop"), conn("loop", "after", "done"),
        ],
    }


def run_case(name, spec):
    wf = req("POST", "/workflows", {"name": "zz char: " + name, "active": False,
                                    "nodes": spec["nodes"],
                                    "connections": spec["connections"]}, TOKEN)
    wf_id = wf.get("_id") or wf.get("id")
    try:
        ex = req("POST", f"/workflows/{wf_id}/execute", {}, TOKEN)
        exec_id = ex.get("executionId")
        detail = {}
        for _ in range(40):
            time.sleep(1.5)
            detail = req("GET", "/executions/" + exec_id, token=TOKEN)
            if detail.get("status") not in ("running", "pending"):
                break

        def drop_timings(value):
            """executionTime is how long a node took, not what it did.

            It wobbles between 0 and 1 ms run to run, and it appears nested
            inside collected loop results too - so left in, four of thirteen
            cases differ on every single diff and a real change hides among
            them. Timing belongs in a benchmark, not a characterisation.
            """
            if isinstance(value, dict):
                return {k: drop_timings(v) for k, v in value.items()
                        if k != "executionTime"}
            if isinstance(value, list):
                return [drop_timings(v) for v in value]
            return value

        records = []
        for rec in detail.get("nodeExecutions", []):
            out = rec.get("output")
            # Only the fields a workflow author would see, and drop the engine's
            # internal loop bookkeeping, which is noisy and not behaviour.
            if isinstance(out, dict):
                out = {k: v for k, v in out.items() if not k.startswith("_")}
            out = drop_timings(out)
            records.append({
                "node": rec.get("nodeId"),
                "status": rec.get("status"),
                "iteration": rec.get("loopIteration"),
                "loopNode": rec.get("loopNodeId"),
                "error": (rec.get("error") or "").split("\n")[0][:110],
                "output": out,
            })
        # Sorted so a run-to-run ordering wobble is not mistaken for a change.
        records.sort(key=lambda r: (r["node"], str(r["iteration"])))
        return {"status": detail.get("status"),
                "error": (detail.get("error") or "").split("\n")[0][:110],
                "stopped": detail.get("stopped"),
                "stopReason": detail.get("stopReason"),
                "toleratedErrorCount": detail.get("toleratedErrorCount"),
                "records": records}
    finally:
        urllib.request.urlopen(urllib.request.Request(
            API + "/workflows/" + wf_id, method="DELETE",
            headers={"Authorization": "Bearer " + TOKEN}))


out = {}
for name in sorted(CASES):
    print("running", name, flush=True)
    try:
        out[name] = run_case(name, CASES[name]())
    except Exception as e:
        out[name] = {"harness_error": repr(e)}

path = sys.argv[1] if len(sys.argv) > 1 else "loop_characterisation.json"
with open(path, "w") as f:
    json.dump(out, f, indent=2, sort_keys=True)
print("\nwrote", path)
for name in sorted(out):
    r = out[name]
    print("  %-42s %s" % (name, r.get("status") or r.get("harness_error")))