loop21:component-factory

#!/usr/bin/env python3
"""loop21-component-factory — declare small logic components as composable data.

A *source* in the composition algebra (∅ → JSONL): you name the logic
primitives you want — a counter, a toggle, a clamp, an accumulator, a
pattern-matcher — give each a small config, and the factory emits one
fully-specified component **spec** per line of JSONL. Nothing is executed
here; the factory's job is to *declare* components as portable data that a
downstream transform, fold, or runner consumes.

    spec  =  {"component": "<kind>", "name": "<id>", "params": {...},
              "port": "transform"|"filter"|..., "spec_version": 1}

WHY IT'S A FACTORY, NOT A LIBRARY

A library hands you a function bound to the process that imported it. This
hands you a *description* of a component — kind, name, parameters, and the
composition port it fills — as a line of JSON. That description travels: you
can pipe it, store it, diff it, hash it, or feed it to any consumer that
knows the five built-in kinds. The factory is the front door to a small,
closed catalog of logic primitives, emitted as a stream.

WHY IT'S HONEST

- **A closed, named catalog.** Five component kinds ship: `counter`,
  `toggle`, `clamp`, `accumulator`, `pattern-match`. Each has a declared
  parameter schema and a declared composition port. Ask for a kind that
  isn't in the catalog and the factory refuses loudly (it never emits a
  spec it can't stand behind).

- **Deterministic by construction.** The emitted stream is a pure function
  of the requested specs. Auto-generated names use a seeded, reproducible
  counter, so `--seed 42` yields byte-identical JSONL on any machine,
  forever. Keys are sorted; there is no wall-clock, no RNG that isn't
  seeded, nothing that drifts between runs.

- **Validated at the door.** Every requested component is checked against
  its kind's parameter schema *before* a spec is emitted. A missing
  required parameter, an out-of-range value, or an unknown parameter is a
  reported error (to stderr, non-zero exit) — never a silently emitted
  spec that a downstream consumer will choke on.

- **The spec is the whole contract.** A consumer needs nothing from this
  tool but the JSONL. The `spec_version` field pins the shape so a consumer
  can reject a spec it doesn't understand rather than mis-read it.

THE HONEST EDGE

The factory *declares* components; it does not *run* them. An emitted spec
is a validated description, not a live object — turning a spec into behavior
is the consumer's job, and this tool makes no claim about whether any
downstream runner implements a kind correctly. It guarantees the spec is
well-formed and catalog-valid, not that anyone honors it.

USAGE
    # one component from a kind + inline params
    python3 loop21-component-factory.py --make counter:start=0,step=2

    # several at once, one JSONL line each
    python3 loop21-component-factory.py \
        --make counter:start=0 --make toggle:initial=false \
        --make clamp:lo=0,hi=100

    # read a batch request from a JSON file ([{kind,name,params}, ...])
    python3 loop21-component-factory.py --batch request.json

    # list the catalog (kinds, params, ports) as JSONL and exit
    python3 loop21-component-factory.py --catalog

    # a short, reproducible demonstration
    python3 loop21-component-factory.py --demo

MIT licensed. Python standard library only. Deterministic under --seed; headless.
"""
# SPDX-License-Identifier: MIT
from __future__ import annotations

import argparse
import json
import sys
from typing import Any

SPEC_VERSION = 1

# ---------------------------------------------------------------------------
# The catalog — a closed, named set of logic-component kinds.
#
# Each kind declares:
#   port    : the composition port the component fills (source/transform/
#             filter/fold/sink) — this is what a MAP tool reads.
#   params  : {name: schema} where schema is
#               {"type": "int"|"number"|"bool"|"string",
#                "required": bool,
#                "default": <value>,        # when not required
#                "min": <n>, "max": <n>}    # optional bounds (numeric)
#   summary : one line, human-facing.
# ---------------------------------------------------------------------------
CATALOG: dict[str, dict[str, Any]] = {
    "counter": {
        "port": "transform",
        "summary": "advances a running integer by a fixed step each item",
        "params": {
            "start": {"type": "int", "required": False, "default": 0},
            "step": {"type": "int", "required": False, "default": 1},
        },
    },
    "toggle": {
        "port": "transform",
        "summary": "flips a boolean on each item, starting from an initial state",
        "params": {
            "initial": {"type": "bool", "required": False, "default": False},
        },
    },
    "clamp": {
        "port": "transform",
        "summary": "constrains a numeric field to an inclusive [lo, hi] range",
        "params": {
            "lo": {"type": "number", "required": True},
            "hi": {"type": "number", "required": True},
        },
    },
    "accumulator": {
        "port": "fold",
        "summary": "reduces a stream to a single running total under an operation",
        "params": {
            "op": {
                "type": "string",
                "required": False,
                "default": "sum",
                "choices": ["sum", "product", "min", "max", "count"],
            },
            "seed": {"type": "number", "required": False, "default": 0},
        },
    },
    "pattern-match": {
        "port": "filter",
        "summary": "passes items whose field matches a fixed literal or set",
        "params": {
            "field": {"type": "string", "required": True},
            "equals": {"type": "string", "required": False, "default": None},
            "in": {"type": "string", "required": False, "default": None},
        },
    },
}


class FactoryError(Exception):
    """A requested component could not be produced (unknown kind or bad params)."""


# ---------------------------------------------------------------------------
# Parameter parsing + coercion
# ---------------------------------------------------------------------------
def _coerce(kind: str, pname: str, raw: Any, schema: dict[str, Any]) -> Any:
    """Coerce a raw parameter value to its declared type, or raise FactoryError."""
    t = schema["type"]
    try:
        if t == "int":
            val = int(raw)
        elif t == "number":
            val = float(raw)
            if val.is_integer():
                val = int(val)  # canonical: 3.0 -> 3, keeps JSONL stable
        elif t == "bool":
            if isinstance(raw, bool):
                val = raw
            else:
                s = str(raw).strip().lower()
                if s in ("true", "1", "yes", "on"):
                    val = True
                elif s in ("false", "0", "no", "off"):
                    val = False
                else:
                    raise ValueError(s)
        elif t == "string":
            val = str(raw)
        else:  # pragma: no cover - guarded by catalog authorship
            raise FactoryError(f"{kind}.{pname}: unknown schema type {t!r}")
    except (ValueError, TypeError):
        raise FactoryError(
            f"{kind}.{pname}: value {raw!r} is not a valid {t}"
        )

    if "choices" in schema and val not in schema["choices"]:
        raise FactoryError(
            f"{kind}.{pname}: {val!r} not in {schema['choices']}"
        )
    for bound, cmp_ok in (("min", lambda v, b: v >= b), ("max", lambda v, b: v <= b)):
        if bound in schema and not cmp_ok(val, schema[bound]):
            raise FactoryError(
                f"{kind}.{pname}: {val!r} violates {bound}={schema[bound]}"
            )
    return val


def build_spec(kind: str, name: str, params: dict[str, Any]) -> dict[str, Any]:
    """Validate a request against the catalog and return a component spec dict.

    Raises FactoryError on an unknown kind, an unknown param, a missing
    required param, or a value that fails coercion/bounds.
    """
    if kind not in CATALOG:
        raise FactoryError(
            f"unknown component kind {kind!r}; catalog: {', '.join(sorted(CATALOG))}"
        )
    entry = CATALOG[kind]
    schema = entry["params"]

    unknown = set(params) - set(schema)
    if unknown:
        raise FactoryError(
            f"{kind}: unknown param(s) {', '.join(sorted(unknown))}; "
            f"allowed: {', '.join(sorted(schema)) or '(none)'}"
        )

    resolved: dict[str, Any] = {}
    for pname, pschema in schema.items():
        if pname in params:
            resolved[pname] = _coerce(kind, pname, params[pname], pschema)
        elif pschema.get("required"):
            raise FactoryError(f"{kind}: missing required param {pname!r}")
        else:
            resolved[pname] = pschema.get("default")

    if kind == "pattern-match" and resolved.get("equals") is None and resolved.get("in") is None:
        raise FactoryError("pattern-match: exactly one of 'equals' or 'in' is required")
    if kind == "pattern-match" and resolved.get("equals") is not None and resolved.get("in") is not None:
        raise FactoryError("pattern-match: give only one of 'equals' or 'in', not both")

    return {
        "component": kind,
        "name": name,
        "params": resolved,
        "port": entry["port"],
        "spec_version": SPEC_VERSION,
    }


# ---------------------------------------------------------------------------
# Request parsing
# ---------------------------------------------------------------------------
def parse_make(expr: str) -> tuple[str, dict[str, Any]]:
    """Parse a --make expression 'kind:k=v,k2=v2' -> (kind, {params})."""
    if ":" in expr:
        kind, _, param_str = expr.partition(":")
    else:
        kind, param_str = expr, ""
    kind = kind.strip()
    params: dict[str, Any] = {}
    if param_str.strip():
        for pair in param_str.split(","):
            if "=" not in pair:
                raise FactoryError(f"bad param {pair!r} in {expr!r} (want k=v)")
            k, _, v = pair.partition("=")
            params[k.strip()] = v.strip()
    return kind, params


def _auto_name(kind: str, ordinal: int) -> str:
    """Deterministic auto-name: kind + zero-padded seeded ordinal."""
    return f"{kind}-{ordinal:04d}"


def produce(
    requests: list[dict[str, Any]],
    *,
    seed: int = 0,
) -> list[dict[str, Any]]:
    """Turn a list of {kind, name?, params} requests into validated specs.

    Auto-generated names are deterministic: they use a per-kind ordinal that
    starts from `seed` and increments in request order, so the same requests
    under the same seed yield byte-identical specs.
    """
    specs: list[dict[str, Any]] = []
    counters: dict[str, int] = {}
    for req in requests:
        kind = req["kind"]
        params = req.get("params", {})
        name = req.get("name")
        if not name:
            ordinal = counters.get(kind, seed)
            counters[kind] = ordinal + 1
            name = _auto_name(kind, ordinal)
        specs.append(build_spec(kind, name, params))
    return specs


def catalog_lines() -> list[dict[str, Any]]:
    """The catalog itself, as JSONL-emittable dicts (one per kind)."""
    out = []
    for kind in sorted(CATALOG):
        entry = CATALOG[kind]
        out.append(
            {
                "component": kind,
                "port": entry["port"],
                "summary": entry["summary"],
                "params": {
                    p: {k: v for k, v in schema.items()}
                    for p, schema in entry["params"].items()
                },
            }
        )
    return out


# ---------------------------------------------------------------------------
# Emit
# ---------------------------------------------------------------------------
def emit(objs: list[dict[str, Any]], stream=None) -> None:
    """Write objects as JSONL — one per line, keys sorted (deterministic)."""
    stream = sys.stdout if stream is None else stream
    for obj in objs:
        stream.write(json.dumps(obj, sort_keys=True, ensure_ascii=False) + "\n")


def demo(*, seed: int = 42) -> str:
    """A short reproducible demonstration — returns the JSONL block as a string."""
    reqs = [
        {"kind": "counter", "params": {"start": "0", "step": "2"}},
        {"kind": "toggle", "params": {"initial": "false"}},
        {"kind": "clamp", "params": {"lo": "0", "hi": "100"}},
        {"kind": "accumulator", "params": {"op": "sum"}},
        {"kind": "pattern-match", "params": {"field": "status", "equals": "open"}},
    ]
    specs = produce(reqs, seed=seed)
    return "\n".join(
        json.dumps(s, sort_keys=True, ensure_ascii=False) for s in specs
    )


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def _build_arg_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        prog="loop21-component-factory",
        description="Declare small logic components as composable JSONL specs.",
    )
    p.add_argument(
        "--make",
        action="append",
        default=[],
        metavar="KIND:k=v,...",
        help="a component to emit (repeatable)",
    )
    p.add_argument(
        "--batch",
        metavar="FILE",
        help="a JSON file: [{kind, name?, params?}, ...]",
    )
    p.add_argument("--catalog", action="store_true", help="emit the catalog as JSONL and exit")
    p.add_argument("--demo", action="store_true", help="print a short reproducible demonstration")
    p.add_argument("--seed", type=int, default=0, help="seed for deterministic auto-names")
    return p


def main(argv: list[str] | None = None) -> int:
    parser = _build_arg_parser()
    args = parser.parse_args(argv)

    if args.catalog:
        emit(catalog_lines())
        return 0

    if args.demo:
        sys.stdout.write(demo() + "\n")
        return 0

    requests: list[dict[str, Any]] = []
    try:
        if args.batch:
            with open(args.batch, encoding="utf-8") as fh:
                loaded = json.load(fh)
            if not isinstance(loaded, list):
                raise FactoryError("--batch file must be a JSON array of requests")
            for item in loaded:
                if not isinstance(item, dict) or "kind" not in item:
                    raise FactoryError("each --batch item needs a 'kind' field")
                requests.append(
                    {
                        "kind": item["kind"],
                        "name": item.get("name"),
                        "params": item.get("params", {}),
                    }
                )
        for expr in args.make:
            kind, params = parse_make(expr)
            requests.append({"kind": kind, "params": params})

        if not requests:
            parser.error("nothing to make: give --make, --batch, --catalog, or --demo")

        specs = produce(requests, seed=args.seed)
    except FactoryError as exc:
        sys.stderr.write(f"loop21-component-factory: {exc}\n")
        return 2
    except (OSError, json.JSONDecodeError) as exc:
        sys.stderr.write(f"loop21-component-factory: {exc}\n")
        return 2

    emit(specs)
    return 0


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

#!/usr/bin/env python3
"""test_loop21-component-factory.py — the certifying properties of the factory gift.

Run:  python3 test_loop21-component-factory.py    (exit 0 = all pass, 1 = failure)

MUTATION-BITTEN: each test is here because a plausible mutation of the tool makes
it fail loud. The determinism test pins a GOLDEN sha256 of a seeded batch rather
than checking self-equality — a weak self-equality test passes benign reorders
(the Loop MMT sudoku lesson), a pinned golden does not.
"""
import hashlib
import importlib.util
import io
import json
import os
import sys

_HERE = os.path.dirname(os.path.abspath(__file__))
_spec = importlib.util.spec_from_file_location(
    "lcf", os.path.join(_HERE, "loop21-component-factory.py")
)
lcf = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(lcf)

# --- pinned goldens: regenerate ONLY on an intentional change ----------------
DEMO_GOLDEN_SHA256 = "ba304b3fd40715875b8298b68dd6281d64844f641263bc7121ce27c35da00a08"
BATCH_GOLDEN_SHA256 = "566171366bffaa9bcfd9001523666585fafe7547f11e3f2851b0cf36288680fe"
CATALOG_KIND_COUNT = 5

_FAILURES: list[str] = []
_PASSES = 0


def check(cond: bool, msg: str) -> None:
    global _PASSES
    if cond:
        _PASSES += 1
    else:
        _FAILURES.append(msg)


def _sha(block: str) -> str:
    return hashlib.sha256(block.encode()).hexdigest()


# --- determinism -------------------------------------------------------------
def test_seeded_demo_reproduces_pinned_golden():
    """The seeded demo output is byte-identical to a pinned sha256, forever."""
    block = lcf.demo(seed=42)
    check(_sha(block) == DEMO_GOLDEN_SHA256,
          f"demo golden drift: got {_sha(block)}")


def test_seeded_batch_reproduces_pinned_golden():
    """A seeded produce() batch is byte-identical to a pinned sha256."""
    reqs = [
        {"kind": "counter", "params": {"start": "5", "step": "3"}},
        {"kind": "accumulator", "params": {"op": "max", "seed": "0"}},
        {"kind": "clamp", "params": {"lo": "-10", "hi": "10"}},
    ]
    specs = lcf.produce(reqs, seed=100)
    block = "\n".join(json.dumps(s, sort_keys=True, ensure_ascii=False) for s in specs)
    check(_sha(block) == BATCH_GOLDEN_SHA256,
          f"batch golden drift: got {_sha(block)}")


def test_same_seed_twice_is_identical():
    """Two runs with the same seed produce identical specs (self-consistency)."""
    reqs = [{"kind": "counter", "params": {}}, {"kind": "toggle", "params": {}}]
    a = lcf.produce(reqs, seed=7)
    b = lcf.produce(reqs, seed=7)
    check(a == b, "same-seed runs diverged")


# --- catalog is a closed, named set ------------------------------------------
def test_catalog_is_the_declared_five_kinds():
    """The catalog holds exactly the five declared component kinds."""
    check(len(lcf.CATALOG) == CATALOG_KIND_COUNT,
          f"catalog size changed: {sorted(lcf.CATALOG)}")
    check(set(lcf.CATALOG) == {"counter", "toggle", "clamp", "accumulator", "pattern-match"},
          f"catalog kinds changed: {sorted(lcf.CATALOG)}")


def test_every_catalog_kind_declares_a_valid_port():
    """Every kind's port is one of the five composition-algebra verbs."""
    valid = {"source", "transform", "filter", "fold", "sink"}
    for kind, entry in lcf.CATALOG.items():
        check(entry["port"] in valid, f"{kind}: bad port {entry['port']!r}")


# --- validation at the door --------------------------------------------------
def test_unknown_kind_refuses():
    """An unknown component kind raises FactoryError, never emits a spec."""
    try:
        lcf.build_spec("frobnicate", "x", {})
        check(False, "unknown kind did not raise")
    except lcf.FactoryError:
        check(True, "")


def test_missing_required_param_refuses():
    """A missing required param (clamp.hi) raises, never emits a spec."""
    try:
        lcf.build_spec("clamp", "c", {"lo": 0})
        check(False, "missing required param did not raise")
    except lcf.FactoryError:
        check(True, "")


def test_unknown_param_refuses():
    """An unknown param raises rather than being silently dropped or emitted."""
    try:
        lcf.build_spec("counter", "c", {"bogus": 5})
        check(False, "unknown param did not raise")
    except lcf.FactoryError:
        check(True, "")


def test_bad_type_refuses():
    """A non-int for an int param raises rather than coercing to garbage."""
    try:
        lcf.build_spec("counter", "c", {"step": "abc"})
        check(False, "bad-type param did not raise")
    except lcf.FactoryError:
        check(True, "")


def test_out_of_choice_refuses():
    """An accumulator op outside the declared choices raises."""
    try:
        lcf.build_spec("accumulator", "a", {"op": "divide"})
        check(False, "out-of-choice op did not raise")
    except lcf.FactoryError:
        check(True, "")


def test_pattern_match_requires_exactly_one_selector():
    """pattern-match with neither equals nor in raises; with both raises."""
    try:
        lcf.build_spec("pattern-match", "p", {"field": "s"})
        check(False, "pattern-match with no selector did not raise")
    except lcf.FactoryError:
        check(True, "")
    try:
        lcf.build_spec("pattern-match", "p", {"field": "s", "equals": "a", "in": "b"})
        check(False, "pattern-match with both selectors did not raise")
    except lcf.FactoryError:
        check(True, "")


# --- spec shape --------------------------------------------------------------
def test_spec_carries_version_and_port():
    """Every emitted spec pins spec_version and declares its port."""
    s = lcf.build_spec("counter", "c", {"start": 0})
    check(s["spec_version"] == lcf.SPEC_VERSION, "spec_version missing/wrong")
    check(s["port"] == "transform", "counter port wrong")
    check(s["component"] == "counter", "component field wrong")


def test_number_type_canonicalizes_integer_floats():
    """A number param given as 3.0 canonicalizes to 3 so JSONL stays stable."""
    s = lcf.build_spec("clamp", "c", {"lo": "0.0", "hi": "100"})
    check(s["params"]["lo"] == 0 and isinstance(s["params"]["lo"], int),
          f"number did not canonicalize: {s['params']['lo']!r}")


def test_bool_coercion_accepts_words_and_refuses_garbage():
    """bool params accept true/false words; garbage raises."""
    s = lcf.build_spec("toggle", "t", {"initial": "true"})
    check(s["params"]["initial"] is True, "bool 'true' not coerced")
    try:
        lcf.build_spec("toggle", "t", {"initial": "maybe"})
        check(False, "garbage bool did not raise")
    except lcf.FactoryError:
        check(True, "")


# --- CLI exit codes ----------------------------------------------------------
def test_cli_bad_request_exits_2():
    """The CLI returns 2 (not 0, not a traceback) on a bad request."""
    rc = lcf.main(["--make", "counter:step=abc"])
    check(rc == 2, f"bad CLI request returned {rc}, want 2")


def test_cli_catalog_and_demo_exit_0():
    """--catalog and --demo both succeed (exit 0)."""
    check(lcf.main(["--catalog"]) == 0, "--catalog did not exit 0")
    check(lcf.main(["--demo"]) == 0, "--demo did not exit 0")


def test_auto_names_are_ordinal_and_seeded():
    """Auto-generated names increment per-kind from the seed, deterministically."""
    specs = lcf.produce(
        [{"kind": "counter", "params": {}}, {"kind": "counter", "params": {}}],
        seed=10,
    )
    check(specs[0]["name"] == "counter-0010", f"first name {specs[0]['name']!r}")
    check(specs[1]["name"] == "counter-0011", f"second name {specs[1]['name']!r}")


def _run() -> int:
    tests = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)]
    for t in tests:
        try:
            t()
        except Exception as exc:  # a test itself throwing is a failure
            _FAILURES.append(f"{t.__name__} raised {exc!r}")
    total = _PASSES + len(_FAILURES)
    if _FAILURES:
        print(f"FAIL — {_PASSES}/{total} checks passed; {len(_FAILURES)} failure(s):")
        for f in _FAILURES:
            print(f"  ✗ {f}")
        return 1
    print(f"OK — {_PASSES}/{total} checks passed across {len(tests)} tests.")
    return 0


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