Skip to content

Modules

The public Python API of v2: text in, checked documents out; documents in, text out. All of it runs in the DSL CLI (cli-native, TypeScript): parsing, checking, compiling and decompiling. There is no Python model of a composite, so a Python tool that builds composites targets the DSL text, where the result is checked, reviewed and documented (docs/v2/design.md, section 5.3).

  • compile_text / compile_file: .pblang text -> parse -> check -> a process-bigraph {"schema", "state"} document or a CompositeSpec mapping. An error-level finding stops the compile (CompileError carries every finding); warnings and infos come back on the result.
  • decompile: a document or a CompositeSpec mapping -> canonical .pblang text plus sidecar assets.

Documents cross the CLI boundary as JSON in bigraph_schema.json_codec's encoding (numpy arrays tagged), with tuples tagged as {"__tuple__": true, "data": [...]} so that they survive too.

CompileError

Bases: Exception

The text did not pass the checker; diagnostics holds every finding, errors first.

Source code in process_bigraph_lang/v2/api.py 
56
57
58
59
60
61
62
class CompileError(Exception):
    """The text did not pass the checker; `diagnostics` holds every finding, errors first."""

    def __init__(self, diagnostics: list[Diagnostic]):
        self.diagnostics = diagnostics
        errors = [d for d in diagnostics if d.severity == "error"]
        super().__init__("\n".join(str(d) for d in errors or diagnostics))

Compiled dataclass

What compile_text / compile_file return: the artifact and the findings that did not stop the compile.

Source code in process_bigraph_lang/v2/api.py 
36
37
38
39
40
41
42
43
44
45
@dataclass
class Compiled:
    """What `compile_text` / `compile_file` return: the artifact and the findings that did not stop the compile."""

    document: dict[str, Any]
    diagnostics: list[Diagnostic] = field(default_factory=list)

    @property
    def warnings(self) -> list[Diagnostic]:
        return [d for d in self.diagnostics if d.severity == "warning"]

Printed dataclass

What decompile returns: the .pblang text and its sidecar assets (<name>.assets/...), by path.

Source code in process_bigraph_lang/v2/api.py 
48
49
50
51
52
53
@dataclass
class Printed:
    """What `decompile` returns: the `.pblang` text and its sidecar assets (`<name>.assets/...`), by path."""

    text: str
    assets: dict[str, bytes] = field(default_factory=dict)

compile_file(path, *, registry=None, manifests=None, to='document', strict=False, force=False, addresses='import')

Compile a .pblang file; file(...) references resolve next to it. force compiles despite the checker's errors (parse errors still stop it): the escape for documents that rely on defaults the checker flags.

Source code in process_bigraph_lang/v2/api.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def compile_file(
    path: str | os.PathLike[str],
    *,
    registry: InterfaceRegistry | None = None,
    manifests: list[Path] | None = None,
    to: Target = "document",
    strict: bool = False,
    force: bool = False,
    addresses: AddressForm = "import",
) -> Compiled:
    """
    Compile a `.pblang` file; `file(...)` references resolve next to it. `force` compiles despite the checker's
    errors (parse errors still stop it): the escape for documents that rely on defaults the checker flags.
    """
    source = Path(path).resolve()
    with tempfile.TemporaryDirectory() as tmp:
        manifest_paths = [*(manifests or []), *registry_manifests(registry, Path(tmp))]
        diagnostics = check_file(source, manifest_paths)
        fatal = set() if force else {"error", "warning"} if strict else {"error"}
        if any(d.severity in fatal for d in diagnostics) or any(d.code == "P001" for d in diagnostics):
            raise CompileError(diagnostics)
        out = Path(tmp) / ("compiled.yaml" if to == "spec" else "compiled.json")
        args = ["compile", str(source), "--to", to, "--out", str(out), "--addresses", addresses]
        args += ["--force"] if force else []
        if manifest_paths:
            args += ["--interfaces", *[str(Path(m).resolve()) for m in manifest_paths]]
        run_cli(*args)
        text = out.read_text(encoding="utf-8")
    if to == "spec":
        import yaml

        return Compiled(yaml.safe_load(text), diagnostics)
    return Compiled(loads(text), diagnostics)

compile_text(text, *, registry=None, manifests=None, to='document', strict=False, force=False, assets=None, addresses='import')

Compile v2 text. assets holds the sidecar files file(...) references name, by path.

Source code in process_bigraph_lang/v2/api.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def compile_text(
    text: str,
    *,
    registry: InterfaceRegistry | None = None,
    manifests: list[Path] | None = None,
    to: Target = "document",
    strict: bool = False,
    force: bool = False,
    assets: dict[str, bytes] | None = None,
    addresses: AddressForm = "import",
) -> Compiled:
    """Compile v2 text. `assets` holds the sidecar files `file(...)` references name, by path."""
    with tempfile.TemporaryDirectory() as tmp:
        source = Path(tmp) / "document.pblang"
        source.write_text(text, encoding="utf-8")
        for name, content in (assets or {}).items():
            target = Path(tmp) / name
            target.parent.mkdir(parents=True, exist_ok=True)
            target.write_bytes(content)
        return compile_file(
            source, registry=registry, manifests=manifests, to=to, strict=strict, force=force, addresses=addresses
        )

decompile(mapping, *, name='document', registry=None, manifests=None)

Canonical .pblang text (and sidecar assets) for a document or a CompositeSpec mapping (told apart by keys).

Source code in process_bigraph_lang/v2/api.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def decompile(
    mapping: dict[str, Any],
    *,
    name: str = "document",
    registry: InterfaceRegistry | None = None,
    manifests: list[Path] | None = None,
) -> Printed:
    """Canonical `.pblang` text (and sidecar assets) for a document or a CompositeSpec mapping (told apart by keys)."""
    with tempfile.TemporaryDirectory() as tmp:
        source = Path(tmp) / f"{name}.json"
        source.write_text(dumps(mapping), encoding="utf-8")
        out = Path(tmp) / f"{name}.pblang"
        args = ["decompile", str(source), "--name", name, "--out", str(out)]
        manifest_paths = [*(manifests or []), *registry_manifests(registry, Path(tmp))]
        if manifest_paths:
            args += ["--interfaces", *[str(Path(m).resolve()) for m in manifest_paths]]
        run_cli(*args)
        printed = Printed(out.read_text(encoding="utf-8"))
        assets_dir = Path(tmp) / f"{name}.assets"
        if assets_dir.is_dir():
            for asset in sorted(assets_dir.iterdir()):
                printed.assets[f"{name}.assets/{asset.name}"] = asset.read_bytes()
        return printed

dumps(document)

bigraph_schema.json_codec encoding, plus tuples tagged so that they come back as tuples.

Source code in process_bigraph_lang/v2/api.py 
152
153
154
def dumps(document: Any) -> str:
    """`bigraph_schema.json_codec` encoding, plus tuples tagged so that they come back as tuples."""
    return str(json_codec.dumps(_tag_tuples(document)))

Interface declarations as data: a minimal stage-A describe and the registry the normalizer resolves addresses in.

describe_module imports a Python module once (the publish-time step; authoring never imports) and turns every step/process class it exposes into an InterfaceDecl: the address in portable local:!module.Class form, the config_schema, and the ports obtained by calling inputs()/outputs() with the class standing in for self (ports that depend on an instance are marked dynamic_ports). Types stay in bigraph-schema's own form. The InterfaceRegistry holds declarations and reads and writes the JSON interface manifest (pbg-interfaces/1, a format shared with other tools; the earlier pblang-manifest/0 is still read), which the TypeScript checker, printer and binder consume; render_interfaces has the CLI print the .pblang header file.

ConfigField dataclass

One entry of a config_schema: the type in bigraph-schema form and its default, when it has one.

Source code in process_bigraph_lang/v2/interfaces.py 
37
38
39
40
41
42
@dataclass
class ConfigField:
    """One entry of a `config_schema`: the type in bigraph-schema form and its default, when it has one."""

    type: Any = None
    default: Any = MISSING

InterfaceDecl dataclass

A process/step interface as a manifest entry: address, config fields, ports (bigraph-schema forms).

Source code in process_bigraph_lang/v2/interfaces.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
@dataclass
class InterfaceDecl:
    """A process/step interface as a manifest entry: address, config fields, ports (bigraph-schema forms)."""

    name: str
    kind: str  # process | step | adapter
    address: str | None = None
    module: str | None = None
    config: dict[str, ConfigField] = field(default_factory=dict)
    inputs: dict[str, Any] | None = None
    outputs: dict[str, Any] | None = None
    doc: str | None = None
    dynamic_ports: bool = False  # inputs()/outputs() could not be evaluated without an instance

InterfaceRegistry dataclass

Declarations by address: the manifest the CLI reads, and what the normalizer resolves edges in.

Source code in process_bigraph_lang/v2/interfaces.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
@dataclass
class InterfaceRegistry:
    """Declarations by address: the manifest the CLI reads, and what the normalizer resolves edges in."""

    decls: list[InterfaceDecl] = field(default_factory=list)

    @classmethod
    def from_modules(cls, *module_names: str) -> InterfaceRegistry:
        registry = cls()
        for module_name in module_names:
            registry.decls.extend(describe_module(module_name))
        return registry

    @classmethod
    def from_manifest(cls, manifest: dict[str, Any]) -> InterfaceRegistry:
        version = manifest.get("schema_version")
        if version not in (None, MANIFEST_VERSION, LEGACY_MANIFEST_VERSION):
            raise ValueError(
                f"unsupported manifest schema_version {version!r}: this reader understands {MANIFEST_VERSION}"
            )
        return cls([_decl_from_manifest(entry) for entry in manifest.get("components", [])])

    def to_manifest(self) -> dict[str, Any]:
        return {
            "$schema": SCHEMA_URL,
            "schema_version": MANIFEST_VERSION,
            "components": [_decl_to_manifest(d) for d in self.decls],
        }

    def lookup(self, address: str) -> InterfaceDecl | None:
        """By exact address; else by unique short name (`local:Name`, or a class re-exported from another module)."""
        wanted = address if ":" in address else f"local:{address}"
        if not wanted.startswith("local:"):
            return None
        exact = [d for d in self.decls if d.address == wanted]
        if exact:
            return exact[0]
        by_name = [d for d in self.decls if d.name == short_name(wanted)]
        return by_name[0] if len(by_name) == 1 else None

    def write_manifest(self, path: Path) -> Path:
        path.write_text(json.dumps(self.to_manifest(), indent=2, default=str) + "\n", encoding="utf-8")
        return path

lookup(address)

By exact address; else by unique short name (local:Name, or a class re-exported from another module).

Source code in process_bigraph_lang/v2/interfaces.py 
125
126
127
128
129
130
131
132
133
134
def lookup(self, address: str) -> InterfaceDecl | None:
    """By exact address; else by unique short name (`local:Name`, or a class re-exported from another module)."""
    wanted = address if ":" in address else f"local:{address}"
    if not wanted.startswith("local:"):
        return None
    exact = [d for d in self.decls if d.address == wanted]
    if exact:
        return exact[0]
    by_name = [d for d in self.decls if d.name == short_name(wanted)]
    return by_name[0] if len(by_name) == 1 else None

describe_class(cls, module_name=None, name=None)

The declaration of one step/process class.

Source code in process_bigraph_lang/v2/interfaces.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def describe_class(cls: type, module_name: str | None = None, name: str | None = None) -> InterfaceDecl:
    """The declaration of one step/process class."""
    module_name = module_name or cls.__module__
    name = name or cls.__name__
    inputs, outputs, dynamic = ports_of(cls)
    return InterfaceDecl(
        name=name,
        kind=kind_of(cls),
        address=f"local:!{module_name}.{name}",
        module=module_name,
        config=_config_fields(getattr(cls, "config_schema", None) or {}),
        inputs=inputs,
        outputs=outputs,
        doc=doc_of(cls),
        dynamic_ports=dynamic,
    )

describe_module(module_name)

Declarations for the step/process classes a module defines or re-exports (classes of its own package only).

Source code in process_bigraph_lang/v2/interfaces.py 
66
67
68
69
70
71
72
73
74
75
def describe_module(module_name: str) -> list[InterfaceDecl]:
    """Declarations for the step/process classes a module defines or re-exports (classes of its own package only)."""
    module = importlib.import_module(module_name)
    is_package = hasattr(module, "__path__")
    decls: list[InterfaceDecl] = []
    for name, cls in inspect.getmembers(module, inspect.isclass):
        if name.startswith("_") or not is_edge_class(cls) or not _belongs_to(cls, module_name, is_package):
            continue
        decls.append(describe_class(cls, module_name, name))
    return decls

render_interfaces(decls, module_name)

The .pblang header file of a module's declarations, as the CLI prints it.

Source code in process_bigraph_lang/v2/interfaces.py 
141
142
143
144
145
146
147
148
149
150
151
def render_interfaces(decls: list[InterfaceDecl], module_name: str) -> str:
    """The `.pblang` header file of a module's declarations, as the CLI prints it."""
    from process_bigraph_lang.v2.langium import run_cli

    with tempfile.TemporaryDirectory() as tmp:
        manifest = Path(tmp) / "manifest.json"
        entries = [_decl_to_manifest(d) for d in decls]
        for entry in entries:
            entry["module"] = module_name
        manifest.write_text(json.dumps({"schema_version": MANIFEST_VERSION, "components": entries}, default=str))
        return run_cli("headers", str(manifest)).stdout

short_name(address)

Name of local:!pkg.mod.Name, local:Name or Name.

Source code in process_bigraph_lang/v2/interfaces.py 
60
61
62
63
def short_name(address: str) -> str:
    """`Name` of `local:!pkg.mod.Name`, `local:Name` or `Name`."""
    data = address.split(":", 1)[1] if ":" in address else address
    return data.lstrip("!").rsplit(".", 1)[-1]

describe-model for SBML: a model file's interface as data (see docs/v2/design.md, section 5.6).

describe_sbml reads an SBML file with the standard library's XML parser (no libsbml, no process code) and returns a ModelHeader: the model's time unit, compartments, species (as concentrations unless the species has only substance units) and parameters, each with its value and its units composed from the model's unit definitions in a pint-readable spelling, plus the counts a solver cares about (reactions, events, rules) and the SBML level, version and packages. to_dict is the JSON form (pblang-model/0), which the CLI renders as the .pblang model header (print_model_header).

describe_sbml(path)

The model header of an SBML file.

Source code in process_bigraph_lang/v2/sbml.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def describe_sbml(path: str | Path) -> ModelHeader:
    """The model header of an SBML file."""
    path = Path(path)
    root = ET.parse(path).getroot()
    model = root.find("{*}model")
    if model is None:
        raise ValueError(f"{path}: no <model> element")
    header = ModelHeader(name=_identifier(path.stem), source=path.name, sbml_id=model.get("id"))
    header.level, header.version = _int(root.get("level")), _int(root.get("version"))
    header.packages = sorted({_package(k) for k in root.attrib if k.endswith(":required")} - {""})
    unit_defs = {d.get("id", ""): _unit_text(d) for d in model.findall("{*}listOfUnitDefinitions/{*}unitDefinition")}

    def units(reference: str | None, default_key: str) -> str | None:
        if reference:
            return unit_defs.get(reference) or _KIND_SYMBOLS.get(reference, reference)
        return unit_defs.get(model.get(f"{default_key}Units") or default_key) or _SBML_DEFAULTS.get(default_key)

    header.time_units = units(None, "time")
    for c in model.findall("{*}listOfCompartments/{*}compartment"):
        dims = _int(c.get("spatialDimensions")) if c.get("spatialDimensions") else 3
        default = {3: "volume", 2: "area", 1: "length"}.get(dims or 3, "volume")
        header.compartments[c.get("id", "")] = Quantity(_float_or_none(c.get("size")), units(c.get("units"), default))
    for sp in model.findall("{*}listOfSpecies/{*}species"):
        name = sp.get("id", "")
        compartment = sp.get("compartment")
        substance = units(sp.get("substanceUnits"), "substance")
        comp_units = (
            header.compartments[compartment].units if compartment in header.compartments else units(None, "volume")
        )
        only_substance = sp.get("hasOnlySubstanceUnits") == "true"
        if only_substance:
            initial, kind, unit_text = _float_or_none(sp.get("initialAmount")), "amount", substance
        elif sp.get("initialConcentration") is not None:
            initial, kind, unit_text = (
                _float_or_none(sp.get("initialConcentration")),
                "concentration",
                _ratio(substance, comp_units),
            )
        else:
            amount = _float_or_none(sp.get("initialAmount"))
            size = header.compartments[compartment].value if compartment in header.compartments else None
            initial = amount / size if amount is not None and size else amount
            kind, unit_text = "concentration", _ratio(substance, comp_units)
            if amount is not None and size:
                header.notes.append(
                    f"species {name}: initial amount {amount:g} converted to concentration by compartment size {size:g}"
                )
        header.species[name] = SpeciesDecl(initial, unit_text, kind, compartment, constant=sp.get("constant") == "true")
    for pr in model.findall("{*}listOfParameters/{*}parameter"):
        header.parameters[pr.get("id", "")] = Quantity(
            _float_or_none(pr.get("value")),
            units(pr.get("units"), "") if pr.get("units") else None,
            pr.get("constant") != "false",
        )
    header.counts = {
        "reactions": len(model.findall("{*}listOfReactions/{*}reaction")),
        "events": len(model.findall("{*}listOfEvents/{*}event")),
        "rules": len(model.findall("{*}listOfRules/*")),
        "function_definitions": len(model.findall("{*}listOfFunctionDefinitions/{*}functionDefinition")),
    }
    return header

print_model_header(header)

The .pblang model header: a generated declaration to import, not a composite (rendered by the CLI).

Source code in process_bigraph_lang/v2/sbml.py 
169
170
171
172
173
174
175
176
def print_model_header(header: ModelHeader) -> str:
    """The `.pblang` model header: a generated declaration to `import`, not a composite (rendered by the CLI)."""
    from process_bigraph_lang.v2.langium import run_cli

    with tempfile.TemporaryDirectory() as tmp:
        source = Path(tmp) / "model.json"
        source.write_text(json.dumps(header.to_dict()), encoding="utf-8")
        return run_cli("model-header", str(source)).stdout

Equivalence of composite documents modulo representation.

Two documents are equivalent when they realize the same composite. normalize_document rewrites a document into one canonical shape: types are pushed through bigraph-schema (core.access + core.render(defaults=True)), addresses become protocol:data strings, wires become lists, tuples and arrays become lists, runtime keys (instance, _contract, the root global_time) are dropped, and the edge entries of the schema half are folded into the state half. equivalent compares two normalized documents; diff_documents explains a difference.

deep_equal(a, b, *, strict=False, rel_tol=1e-09)

Structural equality that tolerates numpy arrays and float noise.

With strict=False (the normalizer's mode) tuples equal lists and ints equal floats; with strict=True container types and numeric types must match exactly (the v1 parity tests use this).

Source code in process_bigraph_lang/v2/normalize.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def deep_equal(a: Any, b: Any, *, strict: bool = False, rel_tol: float = 1e-9) -> bool:
    """
    Structural equality that tolerates numpy arrays and float noise.

    With `strict=False` (the normalizer's mode) tuples equal lists and ints equal floats; with `strict=True`
    container types and numeric types must match exactly (the v1 parity tests use this).
    """
    if isinstance(a, np.ndarray) or isinstance(b, np.ndarray):
        if strict and not (isinstance(a, np.ndarray) and isinstance(b, np.ndarray)):
            return False
        return deep_equal(_to_list(a), _to_list(b), strict=False, rel_tol=rel_tol)
    if isinstance(a, dict) and isinstance(b, dict):
        return set(a) == set(b) and all(deep_equal(a[k], b[k], strict=strict, rel_tol=rel_tol) for k in a)
    if isinstance(a, (list, tuple)) and isinstance(b, (list, tuple)):
        if strict and type(a) is not type(b):
            return False
        return len(a) == len(b) and all(deep_equal(x, y, strict=strict, rel_tol=rel_tol) for x, y in zip(a, b))
    if isinstance(a, bool) or isinstance(b, bool):
        return a is b or (isinstance(a, bool) and isinstance(b, bool) and a == b)
    if _is_number(a) and _is_number(b):
        if strict and (isinstance(a, float) != isinstance(b, float)):
            return False
        return math.isclose(float(a), float(b), rel_tol=rel_tol, abs_tol=0.0)
    return bool(a == b)

diff_documents(a, b, core, limit=200, registry=None)

A readable account of how two documents differ after normalization (empty when equivalent).

Source code in process_bigraph_lang/v2/normalize.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def diff_documents(
    a: dict[str, Any], b: dict[str, Any], core: Any, limit: int = 200, registry: InterfaceRegistry | None = None
) -> str:
    """A readable account of how two documents differ after normalization (empty when equivalent)."""
    left, right = normalize_document(a, core, registry), normalize_document(b, core, registry)
    flat_left, flat_right = _flatten(left), _flatten(right)
    lines: list[str] = []
    for key in sorted(set(flat_left) | set(flat_right)):
        if key not in flat_right:
            lines.append(f"- {key}: {_show(flat_left[key])}")
        elif key not in flat_left:
            lines.append(f"+ {key}: {_show(flat_right[key])}")
        elif not deep_equal(flat_left[key], flat_right[key]):
            lines.append(f"~ {key}: {_show(flat_left[key])} -> {_show(flat_right[key])}")
    if len(lines) > limit:
        lines = [*lines[:limit], f"... {len(lines) - limit} more"]
    if not lines and not deep_equal(left, right):
        dump_left = json.dumps(left, indent=2, sort_keys=True, default=str).splitlines()
        dump_right = json.dumps(right, indent=2, sort_keys=True, default=str).splitlines()
        lines = list(difflib.unified_diff(dump_left, dump_right, "a", "b", lineterm=""))
    return "\n".join(lines)

is_edge_node(value)

A step/process node in the state half (its _type says so, or it carries an address).

Source code in process_bigraph_lang/v2/normalize.py 
41
42
43
def is_edge_node(value: Any) -> bool:
    """A step/process node in the state half (its `_type` says so, or it carries an `address`)."""
    return isinstance(value, dict) and (value.get("_type") in EDGE_KINDS or "address" in value)

is_site(value)

A template site: {"_type": "site", "_sort": ..., "_default"?: ...}.

Source code in process_bigraph_lang/v2/normalize.py 
36
37
38
def is_site(value: Any) -> bool:
    """A template site: `{"_type": "site", "_sort": ..., "_default"?: ...}`."""
    return isinstance(value, dict) and value.get("_type") == "site"

normalize_document(doc, core, registry=None)

The canonical shape of a document (see the module docstring). core is a bigraph-schema core. With a registry, an edge's _inputs/_outputs that merely repeat its interface's face are dropped (process-bigraph merges them with interface(), so they carry no information).

Source code in process_bigraph_lang/v2/normalize.py 
49
50
51
52
53
54
55
56
57
58
59
60
def normalize_document(doc: dict[str, Any], core: Any, registry: InterfaceRegistry | None = None) -> dict[str, Any]:
    """
    The canonical shape of a document (see the module docstring). `core` is a bigraph-schema core. With a
    `registry`, an edge's `_inputs`/`_outputs` that merely repeat its interface's face are dropped (process-bigraph
    merges them with `interface()`, so they carry no information).
    """
    table: _EdgeTable = {}
    raw_schema = doc.get("schema") or {}
    schema = _norm_schema(raw_schema, core, (), table)
    state = _norm_state(doc.get("state") or {}, raw_schema, core, (), table, registry)
    options = {k: _canon(doc[k], core) for k in COMPOSITE_OPTION_KEYS if k in doc}
    return {"schema": schema if isinstance(schema, dict) else {}, "state": state, "options": options}