Reference evaluator · Python

"""Offline Fiction-CWR v0.1 token checker. No provider calls or code execution."""
import argparse
import json
import re
import sys
from pathlib import Path

VERSION = "fiction-cwr-token-v0.1"
MAX_JSON_BYTES = 1_048_576


class InputError(ValueError):
    pass


def _pairs(pairs):
    obj = {}
    for key, value in pairs:
        if key in obj:
            raise InputError(f"duplicate JSON key: {key}")
        obj[key] = value
    return obj


def parse_json(text):
    def invalid_constant(value):
        raise InputError(f"non-finite JSON number: {value}")
    try:
        if type(text) is not str or len(text.encode("utf-8")) > MAX_JSON_BYTES:
            raise InputError("JSON text exceeds input limit or is not text")
        return json.loads(text, object_pairs_hook=_pairs,
                          parse_constant=invalid_constant)
    except (ValueError, RecursionError, UnicodeError) as exc:
        raise InputError(str(exc)) from exc


def exact(obj, required, optional=()):
    if type(obj) is not dict:
        raise InputError("expected object")
    keys = set(obj)
    if not set(required) <= keys or not keys <= set(required) | set(optional):
        raise InputError(f"unexpected or missing keys: {list(obj)}")


def integer(value, lower=0, upper=1000):
    if type(value) is not int or not lower <= value <= upper:
        raise InputError(f"expected integer in [{lower}, {upper}]")


def identifier(value):
    return type(value) is str and re.fullmatch(r"[A-Za-z][A-Za-z0-9_-]{0,63}", value)


def counts(obj, entities, complete=False, positive=False):
    if type(obj) is not dict or not set(obj) <= set(entities):
        raise InputError("count map contains unknown entities or is not an object")
    if complete and set(obj) != set(entities):
        raise InputError("complete inventory required")
    for value in obj.values():
        integer(value, int(positive))


def invariant(world, inventory):
    c = world["constraints"]
    return (all(v >= 0 for v in inventory.values())
            and sum(inventory.values()) <= c["inventory_cap"]
            and all(inventory[k] >= v for k, v in c["minimum"].items())
            and all(inventory[k] <= v for k, v in c["maximum"].items()))


def validate_world(world):
    exact(world, ("schema_version", "id", "entities", "initial", "actions",
                  "goal", "constraints", "budget"))
    if world["schema_version"] != VERSION or not identifier(world["id"]):
        raise InputError("unsupported world version or invalid id")
    entities = world["entities"]
    if (type(entities) is not list or not 1 <= len(entities) <= 32
            or not all(identifier(e) for e in entities)
            or len(set(entities)) != len(entities)):
        raise InputError("entities must be distinct identifiers")
    counts(world["initial"], entities, complete=True)
    counts(world["goal"], entities, complete=True)
    ops = world["actions"]
    if (type(ops) is not dict or not 1 <= len(ops) <= 32
            or not all(identifier(a) for a in ops)):
        raise InputError("actions must be named primitive objects")
    for op in ops.values():
        exact(op, ("consume", "require", "produce", "cost"))
        for key in ("consume", "require", "produce"):
            counts(op[key], entities, positive=True)
        if set(op["require"]) & (set(op["consume"]) | set(op["produce"])):
            raise InputError("retained catalysts cannot be consumed or produced")
        if not op["consume"] and not op["produce"]:
            raise InputError("primitive must have an inventory effect")
        integer(op["cost"], 0, 10000)
    c = world["constraints"]
    exact(c, ("inventory_cap", "minimum", "maximum"))
    integer(c["inventory_cap"])
    counts(c["minimum"], entities)
    counts(c["maximum"], entities)
    b = world["budget"]
    exact(b, ("max_actions", "max_cost"))
    integer(b["max_actions"], 0, 64)
    integer(b["max_cost"], 0, 100000)
    if not invariant(world, world["initial"]):
        raise InputError("initial state violates an invariant")
    if not invariant(world, world["goal"]):
        raise InputError("goal violates a static invariant")


def load_world(path):
    world = parse_json(Path(path).read_text(encoding="utf-8"))
    validate_world(world)
    return world


def grade(world, output):
    """Grade raw model JSON without repair; traces are evaluator-only."""
    validate_world(world)
    inventory = dict(world["initial"])
    cost = executed = 0
    def result(passed, failure=None, step=None, detail=None):
        return {"pass": passed, "failure": failure, "step": step,
                "detail": detail, "actions_executed": executed,
                "cost": cost, "final_inventory": dict(inventory)}
    try:
        answer = parse_json(output)
        exact(answer, ("actions",))
        if type(answer["actions"]) is not list:
            raise InputError("actions must be an array")
        if not all(identifier(a) for a in answer["actions"]):
            raise InputError("actions must contain primitive identifiers")
    except InputError as exc:
        return result(False, "schema", detail=str(exc))
    for step, name in enumerate(answer["actions"], 1):
        if name not in world["actions"]:
            return result(False, "entity", step, "unknown primitive")
        op = world["actions"][name]
        if (step > world["budget"]["max_actions"]
                or cost + op["cost"] > world["budget"]["max_cost"]):
            return result(False, "budget", step, "hard budget exceeded")
        if any(inventory[k] < v for k, v in op["require"].items()):
            return result(False, "transition", step, "missing retained catalyst")
        if any(inventory[k] < v for k, v in op["consume"].items()):
            return result(False, "transition", step, "missing consumable")
        successor = dict(inventory)
        for key, value in op["consume"].items():
            successor[key] -= value
        for key, value in op["produce"].items():
            successor[key] += value
        if not invariant(world, successor):
            return result(False, "constraint", step, "successor violates invariant")
        inventory = successor
        cost += op["cost"]
        executed += 1
    if inventory != world["goal"]:
        return result(False, "goal", detail="exact final inventory not achieved")
    return result(True)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    sub = parser.add_subparsers(dest="command", required=True)
    for command in ("inspect", "grade", "solve"):
        item = sub.add_parser(command)
        item.add_argument("world", type=Path)
        if command == "grade":
            item.add_argument("answer", type=Path)
        if command == "solve":
            item.add_argument("--max-nodes", type=int, default=100000)
    sub.add_parser("self-test")
    args = parser.parse_args()
    if args.command == "self-test":
        import unittest
        suite = unittest.defaultTestLoader.discover(
            str(Path(__file__).resolve().parent / "tests"))
        return 0 if unittest.TextTestRunner(verbosity=2).run(suite).wasSuccessful() else 1
    try:
        world = load_world(args.world)
        if args.command == "inspect":
            value = {k: world[k] for k in ("id", "schema_version", "entities", "goal", "budget")}
            value["primitive_names"] = list(world["actions"])
        elif args.command == "grade":
            value = grade(world, args.answer.read_text(encoding="utf-8"))
        else:
            from oracle import solve
            value = solve(world, args.max_nodes)
        print(json.dumps(value, indent=2, sort_keys=True))
        return 0 if args.command != "grade" or value["pass"] else 1
    except (InputError, OSError, UnicodeError) as exc:
        print(json.dumps({"harness_error": str(exc)}), file=sys.stderr)
        return 2


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