diff --git a/changelog.d/orrery-schema-cli.added.md b/changelog.d/orrery-schema-cli.added.md new file mode 100644 index 000000000..8ed3fd1ef --- /dev/null +++ b/changelog.d/orrery-schema-cli.added.md @@ -0,0 +1 @@ +Add a Microcosm schema adapter and `python -m microcosm.graph.explorer` CLI for the Orrery shared viewer, preserving field identities and exact large integers while refusing invalid or oversized input and existing output files. diff --git a/docs/graph-explorer.md b/docs/graph-explorer.md index 0f4fd9f12..caadf0044 100644 --- a/docs/graph-explorer.md +++ b/docs/graph-explorer.md @@ -1,5 +1,10 @@ # Graph explorer +For declaration and schema inspection with the shared viewer, see the +[shared graph explorer adapter](shared-graph-explorer-adapter.md). The run +renderer described below remains unchanged, including its separate cache and +gate statuses. + The graph explorer is one self-contained HTML file generated from a compiled graph and its run manifest. It contains its own CSS, JavaScript, DAG, and small charts. A reviewer can copy the file to another machine and open it directly in diff --git a/docs/shared-graph-explorer-adapter.md b/docs/shared-graph-explorer-adapter.md new file mode 100644 index 000000000..e244254f2 --- /dev/null +++ b/docs/shared-graph-explorer-adapter.md @@ -0,0 +1,126 @@ +# Shared graph explorer adapter + +`microcosm.graph.explorer` converts a `microcosm.graph.schema.v1` metadata +export into `graph-explorer/v1`, the portable contract consumed by [Orrery](https://github.com/TheAxiomFoundation/orrery), the +shared graph viewer. Microcosm owns the graph's calculation +and schema meanings. The shared package owns navigation, rendering and bundled +offline HTML. + +The adapter adds no JavaScript dependency to Microcosm. It reads no microdata, +executes no population operation, and leaves `explain_html` unchanged. + +## Open a saved schema in Orrery + +From a synced Microcosm checkout, export a saved schema with: + +```sh +uv run python -m microcosm.graph.explorer \ + --input compiled-schema.json --output graph.json --title "Microcosm US" +graph-explorer --input graph.json --output graph.html +``` + +The second command uses the separately installed, accepted shared viewer +0.3.0 CLI, whose package name is still `@axiom-foundation/graph-explorer`. +The forthcoming Orrery package rename does not change the JSON contract. +Microcosm does not install or upgrade the viewer as part of export. + +Open `graph.html` using a local static server or your existing artifact viewer. +The HTML contains its viewer assets and needs no CDN. Direct `file://` opening +is outside the current Microcosm browser acceptance. + +The Python command accepts only saved schema JSON, rejects duplicate keys and +non-finite numbers, bounds input before decoding, and creates a new output file. +It refuses an existing output, including the input path. A saved manifest or +microdata file is not a schema export. Review schema metadata before sharing it: +the entire supplied metadata is retained, and canvas filtering does not redact it. + +The same exporter is available as a Python API: + +```python +import json +from pathlib import Path + +from microcosm.graph.explorer import graph_explorer_json + +schema = json.loads(Path("compiled-schema.json").read_text(encoding="utf-8")) +Path("graph.json").write_text(graph_explorer_json(schema), encoding="utf-8") +``` + +The schema producer is being integrated separately. In a checkout containing +`microcosm.graph.schema.graph_schema`, the input can be produced directly with +`graph_schema(compiled)`. This adapter also accepts previously saved exports; +it does not require that producer to be installed. `graph_explorer_document` +returns the same document as detached Python dictionaries and lists. + +The shared package's built CLI accepts: + +```sh +graph-explorer --input graph.json --output graph.html +``` + +The shared package must supply its built viewer assets. Microcosm does not +download them or launch a build automatically. HTML export and browser +verification are distinct checks; a successfully written HTML file alone does +not establish that its embedded viewer runs. + +## What the graph contains + +The document contains every operation, source declaration, visible field version +and input binding in the supplied schema. A field identity contains its +population, entity, column, producer and nearest declaration. A pre-rewrite +input therefore remains distinct from the final value in the same population. +Population coordinates are retained in `data`; they are not replaced by a +presentation revision or containment parent. + +| Edge kind | Meaning | +| --- | --- | +| `compiled_predecessor` | An operation dependency listed in the supplied compiler metadata | +| `produced` | The provider of a versioned field value, including a structural carrier | +| `declared_read` | An operation's slice, slice mask, output mask or rewrite incumbent; role and row mask are retained | +| `structural_input` | Ancestry from a completed base population's fields into its structural successor | +| `source` | A named external input and its declared codec | +| `artifact` | A producer-to-consumer dependency with its exact alias, artifact name and nominal type/version | + +Structural ancestry does not imply unchanged values. Declared reads describe +operation-level dependencies; they do not infer a separate mathematical formula +for each output. Typed artifact declarations do not establish the existence of +runtime artifact bytes. Source declarations remain domain nodes; citation +references cannot substitute for their codec contract. + +The complete original metadata stays in `metadata.microcosm`, including compiled +owners/order/versions and declaration details such as `mass_partition`. Exact +Python integers outside JavaScript's safe range are transported as +`{"integer_literal": "..."}` before JavaScript parses the document. Large integral +floats use `{"float_literal": "..."}` so their type is not silently changed to +integer. The raw declaration digest is retained separately from presentation +revisions. + +## Scope, identity and evidence + +`complete_supplied_schema` means the complete supplied metadata snapshot. It +does not mean the complete US recipe. Entity IDs and memberships, scientific +units, period semantics, source preparation internals and implicit runtime reads +cannot be inferred when the source schema omits them. Declared ownership is +not evidence that a field contains valid materialized values. + +The converter checks JSON bounds, declaration digest consistency, references, +field/declaration agreement and exact read-role coverage. It does not recompile +or authenticate an imported dictionary. The document revision hashes the whole +supplied schema, including its compiler tables; node and edge revisions hash +their transport records before the revision field is attached. These are +metadata content digests, not execution keys or source-byte attestations. + +No execution, authorship, cache, gate or release status is inferred from schema +metadata. This first adapter supplies no runtime activities or Receipt verdicts. +The existing deterministic run viewer retains its independent cache/gate axes. +A later run adapter must bind actual run evidence and keep these statuses +separate. Receipt assessments must come from a configured host verifier and +bind to the exact exported document bytes, not merely the source declaration's +hash. Custody verification does not establish scientific correctness. + +Exports fail explicitly at their development bounds rather than silently +truncating: 256 operations, 256 sources, 20,000 presentation nodes, 100,000 edges, +32 MiB input and 64 MiB output. Prospective transport bytes are charged as +records are added. These are metadata limits, not microdata size limits or a +claim about total Python process memory. The viewer can focus or collapse a +complete document without changing the underlying export. diff --git a/packages/microcosm-graph/README.md b/packages/microcosm-graph/README.md index cb0d46142..ab0edf67d 100644 --- a/packages/microcosm-graph/README.md +++ b/packages/microcosm-graph/README.md @@ -29,6 +29,11 @@ Module map: | `executor.py` | `run_graph`: projection, patching, ownership enforcement, receipts | | `manifest.py` | `RunManifest`, `NodeReceipt`, human decision records | | `view.py` | `describe(node)`: the one-screen view | +| `explorer.py` | Pure adapter from schema metadata to the shared `graph-explorer/v1` presentation contract | The shard depends on `microcosm-frame` only. Kernels that wrap fit, calibrate, or a rules engine live in those shards and register here. + +See [the shared graph explorer adapter](../../docs/shared-graph-explorer-adapter.md) for +versioned field inspection, offline integration and evidence boundaries. The +existing deterministic `explain_html` export remains available unchanged. diff --git a/packages/microcosm-graph/src/microcosm/graph/explorer.py b/packages/microcosm-graph/src/microcosm/graph/explorer.py new file mode 100644 index 000000000..981f18d23 --- /dev/null +++ b/packages/microcosm-graph/src/microcosm/graph/explorer.py @@ -0,0 +1,492 @@ +"""Pure presentation adapter from compiler metadata to graph-explorer/v1. + +The input is a microcosm.graph.schema.v1 export, not a Frame or RunManifest. +This module checks presentation references, not compilation or authenticity. +It never imports a kernel, opens a source, or infers an execution verdict. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import math +import os +import stat +from collections import Counter +from pathlib import Path + +from .canonical import canonical_json + +__all__ = ["graph_explorer_document", "graph_explorer_json"] + +_PROTOCOL = "microcosm.graph.schema.v1" +_SAFE_INTEGER = 2**53 - 1 +_MAX_INPUT_BYTES = 32 * 1024 * 1024 +_MAX_OUTPUT_BYTES = 64 * 1024 * 1024 +_MAX_ITEMS = 2_000_000 +_MAX_NODES = 20_000 +_MAX_EDGES = 100_000 +_FIELD_KEYS = ("population", "entity", "column", "producer", "declared_in") +_READ_KINDS = {"slice", "slice_mask", "output_mask", "rewrite_incumbent"} + + +def _require(condition: bool, message: str) -> None: + if not condition: + raise ValueError(f"Graph explorer: {message}.") + + +def _copy_json(value: object, *, transport: bool = False) -> object: + """Bound and detach plain JSON; reject custom objects before invoking them.""" + items, charge = 0, 0 + + def walk(child: object, depth: int) -> object: + nonlocal items, charge + items += 1 + charge += 16 + _require(items <= _MAX_ITEMS and depth <= 64, "metadata complexity") + kind = type(child) + if kind is str: + _require(len(child) <= 16_384, "string length") + charge += len(child) * 6 + result = child + elif child is None or kind is bool: + result = child + elif kind is int: + _require(child.bit_length() <= 1024, "integer size") + charge += child.bit_length() + result = ( + {"integer_literal": str(child)} + if transport and abs(child) > _SAFE_INTEGER + else child + ) + elif kind is float: + _require(math.isfinite(child), "finite numbers required") + # JS cannot distinguish a large integral float from an unsafe int. + result = ( + {"float_literal": repr(child)} + if transport and child.is_integer() and abs(child) > _SAFE_INTEGER + else child + ) + elif kind is list: + _require(len(child) <= _MAX_ITEMS - items, "array size") + result = [walk(item, depth + 1) for item in child] + elif kind is dict: + _require(len(child) <= (_MAX_ITEMS - items) // 2, "object size") + _require(all(type(key) is str for key in child), "string keys required") + result = { + walk(key, depth + 1): walk(item, depth + 1) + for key, item in child.items() + } + else: + raise ValueError("Graph explorer: plain JSON values required.") + _require(charge <= _MAX_OUTPUT_BYTES, "prospective metadata size") + return result + + return walk(value, 0) + + +def _object(value: object) -> dict: + _require(type(value) is dict, "object required") + return value + + +def _array(value: object) -> list: + _require(type(value) is list, "array required") + return value + + +def _text(value: object) -> str: + _require(type(value) is str and bool(value.strip()), "nonempty text required") + return value + + +def _index(values: object, key: str) -> dict[str, dict]: + result = {} + for item in _array(values): + name = _text(_object(item).get(key)) + _require(name not in result, f"duplicate {key}") + result[name] = item + return result + + +def _id(*parts: str) -> str: + return json.dumps(parts, ensure_ascii=False, separators=(",", ":")) + + +def _field_id(field: dict) -> str: + return _id("field", *(_text(field.get(key)) for key in _FIELD_KEYS)) + + +def _revision(value: object) -> str: + return "sha256:" + hashlib.sha256(canonical_json(value)).hexdigest() + + +def _bounded_json(value: object, limit: int) -> str: + parts, size = [], 0 + encoder = json.JSONEncoder( + ensure_ascii=False, allow_nan=False, sort_keys=True, separators=(",", ":") + ) + for part in encoder.iterencode(value): + size += len(part.encode("utf-8")) + _require(size <= limit, "serialized metadata size") + parts.append(part) + return "".join(parts) + + +def graph_explorer_document(schema: dict, *, title: str | None = None) -> dict: + """Export the entire supplied schema snapshot; never silently truncate. + + Use ``graph_schema(compiled)`` as the producer where available. Imported + dictionaries remain declarations: reference checks and content digests do + not authenticate them or establish that a compiler or kernel actually ran. + The document revision binds all supplied metadata, including compiled tables; + node/edge revisions bind their presentation records, not runtime cache keys. + + Pre-rewrite inputs have distinct field IDs in the same population. Structural + edges mean ancestry, not equality. Reads apply to an operation, without + claiming that each input mathematically determines each individual output. + """ + doc = _object(_copy_json(schema)) + _bounded_json(doc, _MAX_INPUT_BYTES) + _require(doc.get("protocol") == _PROTOCOL, "unsupported schema protocol") + country = _text(doc.get("country")) + graph = _object(doc.get("graph")) + _require(graph.get("country") == country, "country mismatch") + digest = hashlib.sha256(canonical_json(graph)).hexdigest() + _require(doc.get("graph_sha256") == digest, "declaration digest mismatch") + operations = _index(graph.get("nodes"), "id") + sources = _index(graph.get("sources"), "name") + _require(len(operations) <= 256 and len(sources) <= 256, "declaration count") + compiled = _object(doc.get("compiled")) + order = _array(compiled.get("order")) + _require(all(type(item) is str for item in order), "order IDs") + _require(len(order) == len(operations) and set(order) == set(operations), "order") + predecessors = _object(compiled.get("predecessors")) + versions = _object(compiled.get("versions")) + _require(set(predecessors) == set(operations) == set(versions), "compiled IDs") + positions = {name: index for index, name in enumerate(order)} + populations = { + name for name, op in operations.items() if op.get("structural") != "none" + } + for name, op in operations.items(): + _require( + _text(op.get("structural")) + in {"none", "create", "filter", "expand", "reweight"}, + "structural kind", + ) + _require(_text(versions[name]) in populations, "population reference") + _require( + versions[name] == (name if name in populations else op.get("population")), + "population binding", + ) + base = op.get("base") + _require(base is None or _text(base) in populations, "base reference") + parents = _array(predecessors[name]) + _require(all(type(parent) is str for parent in parents), "predecessor IDs") + _require(len(set(parents)) == len(parents), "duplicate predecessor") + _require( + all( + parent in positions and positions[parent] < positions[name] + for parent in parents + ), + "predecessor order", + ) + + nodes, edges = {}, {} + # Reserve wrapper metadata, then charge each detached transport record before + # retaining it. Counts alone would allow long repeated IDs to over-expand. + transport_doc = _copy_json(doc, transport=True) + used = len(_bounded_json(transport_doc, _MAX_OUTPUT_BYTES).encode("utf-8")) + 65_536 + + def presentation_record(record: dict) -> dict: + nonlocal used + record = _object(_copy_json(record, transport=True)) + record["revision"] = _revision(record) + encoded = _bounded_json(record, _MAX_OUTPUT_BYTES - used) + used += len(encoded.encode("utf-8")) + 1 + return record + + def add_node(node: dict) -> None: + _require(node["id"] not in nodes, "duplicate presentation node") + _require(len(nodes) < _MAX_NODES, "node limit") + nodes[node["id"]] = presentation_record(node) + + def add_edge(source: str, target: str, kind: str, data: dict | None = None) -> None: + facts = {} if data is None else data + identity = _id( + "edge", kind, source, target, _bounded_json(facts, _MAX_INPUT_BYTES) + ) + if identity in edges: + return + _require(len(edges) < _MAX_EDGES, "edge limit") + edge = { + "id": identity, + "source": source, + "target": target, + "kind": kind, + "category": "dependency", + "data": facts, + } + edges[identity] = presentation_record(edge) + + declarations = {} + for name, op in operations.items(): + add_node( + { + "id": _id("operation", name), + "label": name, + "kind": "operation", + "data": {"declaration": op, "population": versions[name]}, + } + ) + for output in _array(op.get("outputs")): + output = _object(output) + key = (name, _text(output.get("entity")), _text(output.get("column"))) + _require(key not in declarations, "duplicate Owned declaration") + # The real serializer omits rewrite=False. Normalize only this + # internal lookup; preserve the authored declaration byte-for-byte. + rewrite = output.get("rewrite", False) + _require(type(rewrite) is bool, "rewrite flag") + declarations[key] = {**output, "rewrite": rewrite} + for name, source in sources.items(): + _text(source.get("codec")) + add_node( + { + "id": _id("source", name), + "label": name, + "kind": "source", + "data": {"declaration": source}, + } + ) + + fields, visible = {}, {} + + def add_field(field: dict, *, visible_field: bool) -> str: + identity = _field_id(field) + key = (field["declared_in"], field["entity"], field["column"]) + _require(key in declarations, "field declaration reference") + owned = declarations[key] + _require( + field["population"] in populations and field["producer"] in operations, + "field reference", + ) + _require( + versions[field["producer"]] == field["population"], + "field provider population", + ) + for prop in ("dtype", "rows", "ownership", "rewrite"): + _require( + field.get(prop) == owned.get(prop), "field declaration disagreement" + ) + if identity not in fields: + fields[identity] = field + add_node( + { + "id": identity, + "label": f"{field['entity']}.{field['column']}", + "kind": "field", + "data": {**field, "visible_in_schema": visible_field}, + } + ) + add_edge(_id("operation", field["producer"]), identity, "produced") + else: + _require(fields[identity] == field, "ambiguous field value") + return identity + + for field in _array(doc.get("schema")): + field = _object(field) + coordinate = tuple(_text(field.get(key)) for key in _FIELD_KEYS[:3]) + _require(coordinate not in visible, "duplicate visible field") + visible[coordinate] = add_field(field, visible_field=True) + + # Check role coverage against declarations, without reimplementing the + # compiler's provider resolution. Preserve repeated declared roles too. + expected_reads = Counter() + for name, op in operations.items(): + if op["structural"] == "create": + continue + population = versions[name] if op["structural"] == "none" else op["base"] + for slice_ in _array(op.get("inputs")): + slice_ = _object(slice_) + entity, rows = _text(slice_.get("entity")), _text(slice_.get("rows")) + for column in _array(slice_.get("columns")): + expected_reads[ + name, population, entity, _text(column), rows, "slice" + ] += 1 + if rows != "all": + expected_reads[name, population, entity, rows, "all", "slice_mask"] += 1 + for output in _array(op.get("outputs")): + entity, column, rows = ( + _text(output.get(key)) for key in ("entity", "column", "rows") + ) + if output.get("rewrite", False): + expected_reads[ + name, population, entity, column, rows, "rewrite_incumbent" + ] += 1 + if rows != "all": + expected_reads[ + name, population, entity, rows, "all", "output_mask" + ] += 1 + reads = _array(doc.get("input_bindings")) + _require(len(reads) <= 100_000, "read count") + actual_reads = Counter( + tuple( + _text(_object(read).get(key)) + for key in ("node", "population", "entity", "column", "rows", "kind") + ) + for read in reads + ) + _require(actual_reads == expected_reads, "declared read coverage") + for read in reads: + read = _object(read) + _require( + read.get("node") in operations and read.get("kind") in _READ_KINDS, + "read reference or kind", + ) + _text(read.get("rows")) + key = tuple(_text(read.get(k)) for k in ("declared_in", "entity", "column")) + _require(key in declarations, "read declaration reference") + owned = declarations[key] + field = {key: read[key] for key in _FIELD_KEYS} + field.update( + {key: owned[key] for key in ("dtype", "rows", "ownership", "rewrite")} + ) + identity = add_field(field, visible_field=False) + # Preserve explicit role labels alongside the supplied compiled edges. + # This adapter does not infer a new compiled dependency from a read. + add_edge( + identity, + _id("operation", read["node"]), + "declared_read", + {"read_kind": read["kind"], "rows": read["rows"]}, + ) + + for name, op in operations.items(): + target = _id("operation", name) + for parent in predecessors[name]: + add_edge(_id("operation", parent), target, "compiled_predecessor") + if op.get("base") is not None: + # Every field in the completed base is carried. Values may change + # during structural materialization; this is not an identity edge. + for (population, _entity, _column), field_id in visible.items(): + if population == op["base"]: + add_edge(field_id, target, "structural_input") + for source in _array(op.get("sources")): + _require(type(source) is str and source in sources, "source reference") + add_edge(_id("source", source), target, "source") + for binding in _array(op.get("artifact_inputs", [])): + binding = _object(binding) + producer = binding.get("producer") + _require( + type(producer) is str and producer in operations, "artifact producer" + ) + outputs = _index(operations[producer].get("artifact_outputs", []), "name") + artifact = _text(binding.get("artifact")) + _require( + artifact in outputs + and outputs[artifact].get("type") == binding.get("type"), + "artifact type or output", + ) + _require(producer in predecessors[name], "artifact predecessor") + add_edge(_id("operation", producer), target, "artifact", binding) + + _require( + all( + edge["source"] in nodes and edge["target"] in nodes + for edge in edges.values() + ), + "dangling edge", + ) + result = { + "schemaVersion": "graph-explorer/v1", + "id": _id("microcosm", country), + "title": _text(title) + if title is not None + else f"{country.upper()} graph schema", + "revision": _revision(doc), + "description": "Complete supplied declaration metadata; no execution or release verdict.", + "nodes": [nodes[key] for key in sorted(nodes)], + "edges": [edges[key] for key in sorted(edges)], + "metadata": { + "adapter": "microcosm.graph.explorer.v1", + "scope": "complete_supplied_schema", + "truncated": False, + "evidence_scope": "Imported declaration metadata; not recompiled or authenticated by this adapter.", + "revision_scope": "Content digests of supplied metadata/presentation records; not runtime cache keys.", + "edge_scope": "Operation-level reads and structural ancestry; not individual-output formulas or value equality.", + "microcosm": transport_doc, + "missing_schema": [ + "entity IDs and memberships", + "units", + "period semantics", + "source preparation internals", + "implicit runtime reads", + ], + }, + } + result = _copy_json(result, transport=True) + _bounded_json(result, _MAX_OUTPUT_BYTES) + return result + + +def graph_explorer_json(schema: dict, *, title: str | None = None) -> str: + """Deterministic UTF-8-ready JSON, with exact large-number transport tags. + + Write these bytes directly for host-side snapshot hashing. HTML bundling and + Receipt verification belong to the shared explorer, not this adapter. + """ + return ( + _bounded_json(graph_explorer_document(schema, title=title), _MAX_OUTPUT_BYTES) + + "\n" + ) + + +def _unique_json_object(pairs: list[tuple[str, object]]) -> dict: + result = {} + for key, value in pairs: + _require(key not in result, "duplicate JSON key") + result[key] = value + return result + + +def _invalid_json_constant(value: str) -> None: + raise ValueError("Graph explorer: finite JSON numbers required.") + + +def main(argv: list[str] | None = None) -> int: + """Convert saved schema metadata for Orrery without loading a saved run.""" + parser = argparse.ArgumentParser(description=main.__doc__) + parser.add_argument("--input", type=Path, required=True, help="saved schema JSON") + parser.add_argument( + "--output", type=Path, required=True, help="new graph-explorer/v1 JSON file" + ) + parser.add_argument("--title", help="viewer document title") + args = parser.parse_args(argv) + try: + # Refuse streams/devices before reading; a FIFO must not block the CLI. + descriptor = os.open(args.input, os.O_RDONLY | getattr(os, "O_NONBLOCK", 0)) + with os.fdopen(descriptor, "rb") as stream: + info = os.fstat(stream.fileno()) + _require(stat.S_ISREG(info.st_mode), "input must be a regular file") + _require(info.st_size <= _MAX_INPUT_BYTES, "input byte limit") + raw = stream.read(_MAX_INPUT_BYTES + 1) + _require(len(raw) <= _MAX_INPUT_BYTES, "input byte limit") + schema = json.loads( + raw.decode("utf-8"), + object_pairs_hook=_unique_json_object, + parse_constant=_invalid_json_constant, + ) + rendered = graph_explorer_json(schema, title=args.title) + args.output.parent.mkdir(parents=True, exist_ok=True) + # Exclusive creation also prevents overwriting the source through an alias. + with args.output.open("x", encoding="utf-8", newline="\n") as stream: + stream.write(rendered) + except (OSError, ValueError, RecursionError) as error: + parser.error(str(error)) + print(f"wrote {args.output}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/packages/microcosm-graph/tests/test_explorer.py b/packages/microcosm-graph/tests/test_explorer.py new file mode 100644 index 000000000..df4c676a4 --- /dev/null +++ b/packages/microcosm-graph/tests/test_explorer.py @@ -0,0 +1,465 @@ +"""Presentation contracts over invented declaration metadata; no population run.""" + +import copy +import hashlib +import json +import os + +import pytest + +from microcosm.graph import ( + Graph, + Node, + Owned, + Ownership, + Slice, + SourceRef, + StructuralDelta, + compile_graph, + explorer, + graph_to_json, +) +from microcosm.graph.canonical import canonical_json + + +def test_cli_exports_exact_metadata_and_large_integer(tmp_path): + schema = snapshot() + schema["audit_integer"] = 2**60 + 1 + source = tmp_path / "schema.json" + output = tmp_path / "report" / "graph.json" + source.write_text(json.dumps(schema), encoding="utf-8") + assert ( + explorer.main( + ["--input", str(source), "--output", str(output), "--title", "Microcosm"] + ) + == 0 + ) + assert output.read_text() == explorer.graph_explorer_json(schema, title="Microcosm") + doc = json.loads(output.read_text()) + assert doc["metadata"]["microcosm"]["audit_integer"] == { + "integer_literal": str(2**60 + 1) + } + + +@pytest.mark.parametrize("raw", ['{"protocol":1,"protocol":2}', '{"x":NaN}', "[]"]) +def test_cli_refuses_invalid_json_without_output(tmp_path, raw): + source, output = tmp_path / "schema.json", tmp_path / "graph.json" + source.write_text(raw) + with pytest.raises(SystemExit) as error: + explorer.main(["--input", str(source), "--output", str(output)]) + assert error.value.code == 2 + assert not output.exists() + + +def test_cli_preserves_existing_output_and_input(tmp_path): + source = tmp_path / "schema.json" + raw = json.dumps(snapshot()) + source.write_text(raw) + with pytest.raises(SystemExit) as error: + explorer.main(["--input", str(source), "--output", str(source)]) + assert error.value.code == 2 + assert source.read_text() == raw + + +def test_cli_refuses_oversized_input_before_decoding(tmp_path, monkeypatch): + source, output = tmp_path / "schema.json", tmp_path / "graph.json" + source.write_bytes(b"!" * 33) + monkeypatch.setattr(explorer, "_MAX_INPUT_BYTES", 32) + with pytest.raises(SystemExit) as error: + explorer.main(["--input", str(source), "--output", str(output)]) + assert error.value.code == 2 + assert not output.exists() + + +@pytest.mark.skipif(not hasattr(os, "mkfifo"), reason="requires POSIX FIFO") +def test_cli_refuses_fifo_without_waiting_for_writer(tmp_path): + source, output = tmp_path / "schema.json", tmp_path / "graph.json" + os.mkfifo(source) + with pytest.raises(SystemExit) as error: + explorer.main(["--input", str(source), "--output", str(output)]) + assert error.value.code == 2 + assert not output.exists() + + +def snapshot(): + graph = Graph( + "invented", + sources=(SourceRef("survey", "unused@1"), SourceRef("unused", "unused@1")), + nodes=( + Node( + "base", + "unused.create@1", + structural=StructuralDelta.CREATE, + sources=("survey",), + outputs=( + Owned("person", "age", "int64"), + Owned("person", "mask", "boolean"), + Owned("person", "missing", "float64", ownership=Ownership.ABSENT), + ), + ), + Node( + "filtered", + "unused.filter@1", + structural=StructuralDelta.FILTER, + base="base", + inputs=(Slice("person", ("age",)),), + ), + Node( + "rewrite", + "unused.rewrite@1", + population="filtered", + inputs=(Slice("person", ("age", "mask"), rows="mask"),), + outputs=(Owned("person", "age", "int64", rows="mask", rewrite=True),), + ), + Node( + "consumer", + "unused.consume@1", + population="filtered", + inputs=(Slice("person", ("age",)),), + ), + ), + ) + compiled = compile_graph(graph) + raw = graph_to_json(graph) + fields = [] + for population in ("base", "filtered"): + for owned in graph.nodes[0].outputs: + rewrite = population == "filtered" and owned.column == "age" + fields.append( + { + "population": population, + "entity": owned.entity, + "column": owned.column, + "dtype": owned.dtype, + "producer": "rewrite" if rewrite else population, + "declared_in": "rewrite" if rewrite else "base", + "rows": "mask" if rewrite else "all", + "ownership": owned.ownership.value, + "rewrite": rewrite, + } + ) + + def read( + node, + column, + *, + population="filtered", + producer="filtered", + declared_in="base", + kind="slice", + rows="all", + ): + return { + "node": node, + "entity": "person", + "column": column, + "population": population, + "producer": producer, + "declared_in": declared_in, + "kind": kind, + "rows": rows, + } + + return { + "protocol": "microcosm.graph.schema.v1", + "country": "invented", + "graph_sha256": hashlib.sha256(raw.encode()).hexdigest(), + "graph": json.loads(raw), + "compiled": { + "order": list(compiled.order), + "versions": dict(compiled.versions), + "predecessors": { + key: list(value) for key, value in compiled.predecessors.items() + }, + "owners": [ + list(key) + [owner] for key, owner in sorted(compiled.owners.items()) + ], + }, + "schema": fields, + "input_bindings": [ + read("filtered", "age", population="base", producer="base"), + read("rewrite", "age", rows="mask"), + read("rewrite", "mask", rows="mask"), + read("rewrite", "mask", kind="slice_mask"), + read("rewrite", "mask", kind="output_mask"), + read("rewrite", "age", kind="rewrite_incumbent", rows="mask"), + read("consumer", "age", producer="rewrite", declared_in="rewrite"), + ], + } + + +def update_graph_digest(doc): + doc["graph_sha256"] = hashlib.sha256(canonical_json(doc["graph"])).hexdigest() + + +def test_full_snapshot_and_no_runtime_claims(): + source = snapshot() + doc = explorer.graph_explorer_document(source) + assert doc["schemaVersion"] == "graph-explorer/v1" + assert doc["metadata"]["microcosm"] == source + assert doc["metadata"]["scope"] == "complete_supplied_schema" + assert doc["metadata"]["truncated"] is False + assert len([node for node in doc["nodes"] if node["kind"] == "operation"]) == 4 + assert len([node for node in doc["nodes"] if node["kind"] == "source"]) == 2 + assert len([node for node in doc["nodes"] if node["kind"] == "field"]) == 7 + assert not {"activities", "receipts", "artifacts", "assessments"} & doc.keys() + assert all( + "statuses" not in node and "parentId" not in node for node in doc["nodes"] + ) + assert len( + [edge for edge in doc["edges"] if edge["kind"] == "compiled_predecessor"] + ) == sum(map(len, source["compiled"]["predecessors"].values())) + assert {edge["source"] for edge in doc["edges"]} | { + edge["target"] for edge in doc["edges"] + } <= {node["id"] for node in doc["nodes"]} + + +def test_rewrite_incumbent_is_distinct_and_masks_retain_roles(): + doc = explorer.graph_explorer_document(snapshot()) + ages = [ + n + for n in doc["nodes"] + if n["kind"] == "field" + and n["data"]["population"] == "filtered" + and n["data"]["column"] == "age" + ] + assert len(ages) == 2 and ages[0]["id"] != ages[1]["id"] + incumbent = next(n for n in ages if n["data"]["producer"] == "filtered") + final = next(n for n in ages if n["data"]["producer"] == "rewrite") + assert incumbent["data"]["visible_in_schema"] is False + assert final["data"]["visible_in_schema"] is True + reads = [e for e in doc["edges"] if e["kind"] == "declared_read"] + assert {e["data"]["read_kind"] for e in reads} == { + "slice", + "slice_mask", + "output_mask", + "rewrite_incumbent", + } + incoming = [ + e + for e in reads + if json.loads(e["target"])[1] == "rewrite" + and e["data"]["read_kind"] == "rewrite_incumbent" + ] + assert len(incoming) == 1 and incoming[0]["source"] == incumbent["id"] + assert incoming[0]["data"]["rows"] == "mask" + assert not any( + e["source"] == final["id"] and json.loads(e["target"])[1] == "rewrite" + for e in reads + ) + structural = [e for e in doc["edges"] if e["kind"] == "structural_input"] + assert len(structural) == 3 + assert all(json.loads(e["source"])[1] == "base" for e in structural) + + +def test_typed_artifact_edge_and_uninterpreted_metadata_survive(): + source = snapshot() + base, _, _, consumer = source["graph"]["nodes"] + type_ = {"name": "invented.summary", "schema_version": 2} + base["artifact_outputs"] = [{"name": "summary", "type": type_}] + consumer["artifact_inputs"] = [ + { + "name": "local_alias", + "producer": "base", + "artifact": "summary", + "type": type_, + } + ] + source["compiled"]["predecessors"]["consumer"].append("base") + source["graph"]["mass_partition"] = ["person", "period"] + update_graph_digest(source) + doc = explorer.graph_explorer_document(source) + edge = next(e for e in doc["edges"] if e["kind"] == "artifact") + assert edge["data"] == consumer["artifact_inputs"][0] + assert edge["category"] == "dependency" + assert doc["metadata"]["microcosm"]["graph"]["mass_partition"] == [ + "person", + "period", + ] + + +def test_deterministic_detached_json_and_exact_large_number_transport(): + source = snapshot() + source["graph"]["nodes"][0]["params"] = { + "large": 2**80 + 1, + "negative": -(2**80 + 1), + "safe": 2**53 - 1, + "float": 1e100, + "flag": True, + } + update_graph_digest(source) + before = copy.deepcopy(source) + text = explorer.graph_explorer_json(source) + assert text == explorer.graph_explorer_json(source) and text.endswith("\n") + doc = json.loads(text) + params = doc["metadata"]["microcosm"]["graph"]["nodes"][0]["params"] + assert params == { + "large": {"integer_literal": str(2**80 + 1)}, + "negative": {"integer_literal": str(-(2**80 + 1))}, + "safe": 2**53 - 1, + "float": {"float_literal": "1e+100"}, + "flag": True, + } + doc["metadata"]["microcosm"]["graph"]["nodes"].clear() + assert source == before + + +def test_document_revision_binds_compiled_metadata_not_only_graph(): + source = snapshot() + left = explorer.graph_explorer_document(source) + source["compiled"]["extra_declaration"] = "caller supplied, not verified" + right = explorer.graph_explorer_document(source) + assert left["revision"] != right["revision"] + assert ( + left["metadata"]["microcosm"]["graph_sha256"] + == right["metadata"]["microcosm"]["graph_sha256"] + ) + assert left["nodes"] == right["nodes"] + + +def test_named_mask_read_is_preserved_separately_from_compiler_predecessors(): + graph = Graph( + "invented", + sources=(SourceRef("survey", "unused@1"),), + nodes=( + Node( + "base", + "unused.create@1", + structural=StructuralDelta.CREATE, + sources=("survey",), + outputs=(Owned("person", "age", "int64"),), + ), + Node( + "a_mask", + "unused.mask@1", + population="base", + inputs=(Slice("person", ("age",)),), + outputs=(Owned("person", "mask", "boolean"),), + ), + Node( + "z_consumer", + "unused.consume@1", + population="base", + inputs=(Slice("person", ("age", "mask"), rows="mask"),), + ), + ), + ) + compiled = compile_graph(graph) + source = snapshot() + source["graph"] = json.loads(graph_to_json(graph)) + update_graph_digest(source) + source["compiled"] = { + "order": list(compiled.order), + "versions": dict(compiled.versions), + "predecessors": { + key: list(value) for key, value in compiled.predecessors.items() + }, + "owners": [list(key) + [value] for key, value in compiled.owners.items()], + } + source["schema"] = [ + { + "population": "base", + "entity": "person", + "column": column, + "dtype": dtype, + "producer": owner, + "declared_in": owner, + "rows": "all", + "ownership": "produced", + "rewrite": False, + } + for column, dtype, owner in ( + ("age", "int64", "base"), + ("mask", "boolean", "a_mask"), + ) + ] + source["input_bindings"] = [ + { + "node": node, + "population": "base", + "entity": "person", + "column": column, + "producer": owner, + "declared_in": owner, + "kind": kind, + "rows": rows, + } + for node, column, owner, kind, rows in ( + ("a_mask", "age", "base", "slice", "all"), + ("z_consumer", "age", "base", "slice", "mask"), + ("z_consumer", "mask", "a_mask", "slice", "mask"), + ("z_consumer", "mask", "a_mask", "slice_mask", "all"), + ) + ] + doc = explorer.graph_explorer_document(source) + target = json.dumps(["operation", "z_consumer"], separators=(",", ":")) + incoming = [edge for edge in doc["edges"] if edge["target"] == target] + compiled_parents = { + json.loads(edge["source"])[1] + for edge in incoming + if edge["kind"] == "compiled_predecessor" + } + assert compiled_parents == set(compiled.predecessors["z_consumer"]) + mask = next( + edge + for edge in incoming + if edge["kind"] == "declared_read" and edge["data"]["read_kind"] == "slice_mask" + ) + assert json.loads(mask["source"])[4] == "a_mask" + + +@pytest.mark.parametrize( + "mutation", + [ + lambda doc: doc.update(protocol="future"), + lambda doc: doc.update(graph_sha256="0" * 64), + lambda doc: doc["compiled"]["order"].reverse(), + lambda doc: doc["schema"].append(copy.deepcopy(doc["schema"][0])), + lambda doc: doc["schema"][0].update(dtype="invented"), + lambda doc: doc["input_bindings"][0].update(producer="unknown"), + lambda doc: doc["input_bindings"][0].update(kind="invented"), + lambda doc: doc["input_bindings"].clear(), + lambda doc: doc["input_bindings"].append( + { + **doc["input_bindings"][-1], + "column": "mask", + "producer": "filtered", + "declared_in": "base", + } + ), + ], +) +def test_invalid_references_fail_without_partial_document(mutation): + source = snapshot() + mutation(source) + with pytest.raises(ValueError): + explorer.graph_explorer_document(source) + + +def test_oversize_or_non_json_metadata_fails(monkeypatch): + source = snapshot() + source["extra"] = float("nan") + with pytest.raises(ValueError, match="finite"): + explorer.graph_explorer_document(source) + source["extra"] = source + with pytest.raises(ValueError, match="complexity"): + explorer.graph_explorer_document(source) + source = snapshot() + monkeypatch.setattr(explorer, "_MAX_NODES", 3) + with pytest.raises(ValueError, match="node limit"): + explorer.graph_explorer_document(source) + + +def test_expanded_output_budget_is_checked_during_construction(monkeypatch): + source = snapshot() + budget = len(canonical_json(source)) + 65_536 + 2_000 + monkeypatch.setattr(explorer, "_MAX_OUTPUT_BYTES", budget) + + # The input fits, but presentation records must stop while being added. + # A late-only final serialization check would visit the field loop first. + def unexpected_field(_field): + pytest.fail("expanded records reached field construction before refusal") + + monkeypatch.setattr(explorer, "_field_id", unexpected_field) + with pytest.raises(ValueError, match="serialized metadata size"): + explorer.graph_explorer_document(source)