diff --git a/Makefile b/Makefile index 814f557..8b33f53 100644 --- a/Makefile +++ b/Makefile @@ -10,5 +10,5 @@ test: ctest --test-dir $(BUILD_DIR) --output-on-failure check: test - $(PYTHON) -m py_compile src/pto_asl_model/*.py scripts/pto-asl-run + PYTHONPATH=src $(PYTHON) -m compileall -q src scripts tools git diff --check diff --git a/README.md b/README.md index 5edf9c9..7b5ed97 100644 --- a/README.md +++ b/README.md @@ -6,16 +6,26 @@ consumer boundary for functional runners such as SuperScalarModel `gfrun`. PTO architectural semantics remain owned by [`PTO-ISA/pto-spec`](https://github.com/PTO-ISA/pto-spec). This repository owns -model lifecycle, hosted execution, transport, ABI, ELF loading, and validation. +hosted execution, ABI, ELF loading, and validation. Development changes land through pull requests. The initial ASLRef backend is tracked by the repository issue list. ## Reference runner -The first closure runs consecutive PTO instructions inside one ASLRef process. -It accepts a checked static ELF whose load segments fit the explicit hosted -memory bound, plus an assembled PTO ASL file: +The hosted ELF command is the one model entry used by both the command line +tool and the C ABI. It runs consecutive instructions in one ASLRef process; +ASL owns fetch, instruction length selection, decode, legality, faults, and +architectural state transitions. The model owns ELF validation/loading, the +guest-memory image, stop/result handling, and the process boundary. + +The Python package also exposes strict, passive architectural-state DTOs and +canonical serialization helpers. They do not map memory, authorize accesses, +load images, initialize stacks, reset or restore execution, or provide another +instruction engine. The hosted runner and PTO ASL remain the sole live memory +and execution authority. + +The hosted runner accepts a checked static ELF and an assembled PTO ASL file. ```bash scripts/pto-asl-run \ @@ -78,3 +88,7 @@ promotion gates. The reset contract is in [`docs/model-ndf-v1.md`](docs/model-ndf-v1.md). The snapshot lifecycle, identity, transport, and promotion gates are in [`docs/worker-snapshot-design.md`](docs/worker-snapshot-design.md). + +Run repository checks with `make check`. The C/C++ consumer links +`PTOASLModel::pto_asl_model` and calls the versioned +`pto_model_run_elf` function declared in `include/pto/pto_asl_model.h`. diff --git a/docs/design.md b/docs/design.md new file mode 100644 index 0000000..7050e57 --- /dev/null +++ b/docs/design.md @@ -0,0 +1,50 @@ +# Standalone design + +`pto-asl-model` is an integration layer around the executable PTO ASL +specification. It intentionally keeps the semantic boundary small: + +```text +pto_model_run_elf() → canonical runner → ASLRef → ExecuteNextPTOInstruction() + ↓ + result, manifest +``` + +The package does not import simulator, emulator, benchmark, or compiler +modules. The specification checkout and its generated artifact are explicit +runtime inputs. This makes the model usable from a clean checkout without +creating a second PTO execution API. + +ELF parsing, sidecar validation, artifact identity, and hosted execution stay +behind the single `pto_model_run_elf()` entry. The Python package contains only +passive, strict architectural-state serialization and path helpers alongside +that runner. It has no second image, memory, stack, lifecycle, or snapshot +authority. + +The package owns no duplicate instruction handlers or live memory policy. A +future native backend must implement the same observable state contract and be +admitted by differential tests before it is used for bulk workloads. + +## Specification lifecycle + +The normal lifecycle is: + +1. Edit the modular ASL sources in the specification checkout. +2. Regenerate and type-check `build/pto-spec.asl` there. +3. Invoke `pto_model_run_elf()` or `scripts/pto-asl-run` with the exact lock, + generated ASL artifact, ELF, and sidecar inputs. + +Generated ASL artifacts, ASLRef build outputs, and ELF outputs are not package +source files. + +## Repository split + +This repository owns only the model boundary and its tests. The following +remain external inputs or downstream consumers: + +| Concern | Owner | +| --- | --- | +| ISA semantics, encodings, catalogs | PTO ASL specification | +| ASL interpretation | ASLRef toolchain | +| ELF production | compiler/assembler toolchain | +| high-throughput execution | future native backend | +| timing/performance | timing model | diff --git a/docs/migration.md b/docs/migration.md new file mode 100644 index 0000000..fcecc7f --- /dev/null +++ b/docs/migration.md @@ -0,0 +1,43 @@ +# Standalone checkout guide + +This directory is the complete Python-side ASL model boundary. It can be +installed and tested without importing any other local project. + +## Inputs + +The model deliberately does not vendor the specification or ASLRef build. +Point it at a PTO specification checkout whose generated artifact is ready: + +```bash +export PTO_SPEC_ROOT=/path/to/pto-spec +``` + +The specification checkout must contain `build/pto-spec.asl`, the generated +decoder/source-order files, `scripts/aslref`, and `.aslref-version`. + +## Install and validate + +```bash +python3 -m venv .venv +. .venv/bin/activate +python -m pip install -e . +make test +make check PTO_SPEC=/path/to/pto-spec +``` + +`make test` includes deterministic Python and C/C++ ABI tests. `make check` +adds bytecode compilation and whitespace validation. + +## Python data contract + +The installable Python package contains strict, passive architectural-state +serialization and path helpers. These DTOs can describe observed memory data, +but they do not map storage, enforce access permissions, initialize a stack, +load an image, or reset/restore a live model. Generated ASL and ASLRef remain +explicit inputs to the canonical `pto_model_run_elf()` runner. + +## Extension points + +Keep ISA semantics in ASL. The hosted `pto_model_run_elf()` path alone handles +ELF identity, sidecars, live memory, and execution. Reusable Python code is +limited to passive state serialization and repository-path discovery. diff --git a/docs/step-contract.md b/docs/step-contract.md new file mode 100644 index 0000000..2f93fd0 --- /dev/null +++ b/docs/step-contract.md @@ -0,0 +1,24 @@ +# Canonical next-instruction contract + +PTO has one hosted model entry and one ASL next-instruction entry: + +```text +pto_model_run_elf() + -> pto_asl_model.runner.run() + -> ExecuteNextPTOInstruction() +``` + +The hosted runner owns ELF validation, sidecar and lock identity, memory-image +initialization, stop/result policy, manifests, and the ASLRef process. PTO-SPEC +ASL owns fetch, instruction-width selection, decode, legality, faults, PC/TPC, +and architectural state transitions. + +There is no supported compatibility decoder, explicit-width step command, +arbitrary instruction-handler call, or alternate runtime commit path. Runtime +image, memory, stack, reset, restore, and snapshot-lifecycle classes are not +part of the package. Architectural-state DTOs are passive serialization data; +they cannot initialize or mutate the live runner. + +Any future accelerated backend must preserve this boundary and prove parity +before admission. It may not introduce a private PTO decoder, opcode table, +termination recognizer, or semantic fallback. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..ac54012 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,29 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "pto-asl-model" +version = "0.1.0" +description = "ASL-backed functional model runtime for PTO instruction validation" +readme = "README.md" +requires-python = ">=3.10" +license = {text = "Apache-2.0"} +authors = [{name = "PTO-ISA contributors"}] +dependencies = [] + +[project.optional-dependencies] +test = [] + +[tool.setuptools] +package-dir = {"" = "src"} + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.setuptools.package-data] +pto_asl_model = ["architecture_state.schema.json"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = "-ra" diff --git a/src/pto_asl_model/__init__.py b/src/pto_asl_model/__init__.py index f48d7f2..79e3762 100644 --- a/src/pto_asl_model/__init__.py +++ b/src/pto_asl_model/__init__.py @@ -1,11 +1,42 @@ -"""PTO ASLRef-backed functional-model tooling.""" +"""ASL-backed functional model for PTO instruction validation.""" from .runner import ElfError, ElfImage, LoadSegment, RunConfiguration, run +from .state import ( + ArchitectureState, + BlockState, + FaultState, + MemoryRegion, + MemoryState, + ScalarState, + SharedState, + SharedTile, + StateEnvelope, + TileDescriptor, + TileState, + TileValue, + canonical_hash, + canonical_json, + state_diff, +) +from .paths import repository_root, resolve_pto_spec + __all__ = [ - "ElfError", - "ElfImage", - "LoadSegment", - "RunConfiguration", - "run", + "ElfError", "ElfImage", "LoadSegment", "RunConfiguration", "run", + "ArchitectureState", + "BlockState", + "FaultState", + "MemoryRegion", + "MemoryState", + "ScalarState", + "SharedState", + "SharedTile", + "StateEnvelope", + "TileDescriptor", + "TileState", + "TileValue", + "canonical_hash", + "canonical_json", + "state_diff", + "repository_root", "resolve_pto_spec", ] diff --git a/src/pto_asl_model/architecture_state.schema.json b/src/pto_asl_model/architecture_state.schema.json new file mode 100644 index 0000000..7ce8dbd --- /dev/null +++ b/src/pto_asl_model/architecture_state.schema.json @@ -0,0 +1,167 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://pto-isa.org/schemas/asl-model/architecture-state-v1.json", + "title": "ASL model architectural state envelope", + "description": "Versioned passive architectural-state serialization data; not live runner storage or lifecycle state.", + "$ref": "#/$defs/stateEnvelope", + "$defs": { + "json": {}, + "extensions": { + "type": "object", + "additionalProperties": true + }, + "descriptor": { + "type": "object", + "required": ["dtype", "layout", "shape", "valid_shape", "strides", "extensions"], + "properties": { + "dtype": { "type": "string" }, + "layout": { "type": "string" }, + "shape": { "type": "array", "items": { "type": "integer", "minimum": 0 } }, + "valid_shape": { "type": "array", "items": { "type": "integer", "minimum": 0 } }, + "strides": { "type": "array", "items": { "type": "integer" } }, + "extensions": { "$ref": "#/$defs/extensions" } + }, + "additionalProperties": false + }, + "scalar": { + "type": "object", + "required": ["registers", "pc", "tpc", "flags", "mode", "extensions"], + "properties": { + "registers": { "type": "object", "additionalProperties": { "type": "integer" } }, + "pc": { "type": "integer" }, + "tpc": { "type": "integer" }, + "flags": { "type": "object", "additionalProperties": { "type": ["integer", "boolean"] } }, + "mode": { "type": "string" }, + "extensions": { "$ref": "#/$defs/extensions" } + }, + "additionalProperties": false + }, + "block": { + "type": "object", + "required": ["active", "block_id", "start_pc", "instruction_count", "attributes", "extensions"], + "properties": { + "active": { "type": "boolean" }, + "block_id": { "type": ["integer", "null"] }, + "start_pc": { "type": ["integer", "null"] }, + "instruction_count": { "type": "integer", "minimum": 0 }, + "attributes": { "type": "object", "additionalProperties": true }, + "extensions": { "$ref": "#/$defs/extensions" } + }, + "additionalProperties": false + }, + "tileValue": { + "type": "object", + "required": ["descriptor", "data", "defined", "generation", "extensions"], + "properties": { + "descriptor": { "$ref": "#/$defs/descriptor" }, + "data": { "type": "array" }, + "defined": { "type": "array", "items": { "type": "boolean" } }, + "generation": { "type": "integer", "minimum": 0 }, + "extensions": { "$ref": "#/$defs/extensions" } + }, + "additionalProperties": false + }, + "tile": { + "type": "object", + "required": ["registers", "extensions"], + "properties": { + "registers": { "type": "object", "additionalProperties": { "$ref": "#/$defs/tileValue" } }, + "extensions": { "$ref": "#/$defs/extensions" } + }, + "additionalProperties": false + }, + "sharedTile": { + "type": "object", + "required": ["descriptor", "data", "defined", "generation", "allocation_mask", "extensions"], + "properties": { + "descriptor": { "$ref": "#/$defs/descriptor" }, + "data": { "type": "array" }, + "defined": { "type": "array", "items": { "type": "boolean" } }, + "generation": { "type": "integer", "minimum": 0 }, + "allocation_mask": { "type": "integer", "minimum": 0 }, + "extensions": { "$ref": "#/$defs/extensions" } + }, + "additionalProperties": false + }, + "shared": { + "type": "object", + "required": ["tiles", "generations", "extensions"], + "properties": { + "tiles": { "type": "object", "additionalProperties": { "$ref": "#/$defs/sharedTile" } }, + "generations": { "type": "object", "additionalProperties": { "type": "integer", "minimum": 0 } }, + "extensions": { "$ref": "#/$defs/extensions" } + }, + "additionalProperties": false + }, + "region": { + "type": "object", + "required": ["base", "size", "permissions", "name", "extensions"], + "properties": { + "base": { "type": "integer", "minimum": 0 }, + "size": { "type": "integer", "minimum": 1 }, + "permissions": { "enum": ["r", "w", "x", "rw", "rx", "wx", "rwx"] }, + "name": { "type": "string" }, + "extensions": { "$ref": "#/$defs/extensions" } + }, + "additionalProperties": false + }, + "memory": { + "type": "object", + "required": ["cells", "regions", "extensions"], + "properties": { + "cells": { + "type": "object", + "propertyNames": { "pattern": "^0x(?:0|[1-9a-f][0-9a-f]*)$" }, + "additionalProperties": { "type": "integer", "minimum": 0, "maximum": 255 } + }, + "regions": { "type": "array", "items": { "$ref": "#/$defs/region" } }, + "extensions": { "$ref": "#/$defs/extensions" } + }, + "additionalProperties": false + }, + "fault": { + "type": "object", + "required": ["pending", "kind", "code", "address", "instruction", "message", "recoverable", "extensions"], + "properties": { + "pending": { "type": "boolean" }, + "kind": { "type": "string" }, + "code": { "type": "string" }, + "address": { "type": ["integer", "null"] }, + "instruction": { "type": "string" }, + "message": { "type": "string" }, + "recoverable": { "type": "boolean" }, + "extensions": { "$ref": "#/$defs/extensions" } + }, + "additionalProperties": false + }, + "state": { + "type": "object", + "required": ["scalar", "block", "tile", "shared", "memory", "fault", "pe_id", "thread_id", "cycle", "extensions"], + "properties": { + "scalar": { "$ref": "#/$defs/scalar" }, + "block": { "$ref": "#/$defs/block" }, + "tile": { "$ref": "#/$defs/tile" }, + "shared": { "$ref": "#/$defs/shared" }, + "memory": { "$ref": "#/$defs/memory" }, + "fault": { "$ref": "#/$defs/fault" }, + "pe_id": { "type": "integer", "minimum": 0 }, + "thread_id": { "type": "integer", "minimum": 0 }, + "cycle": { "type": "integer", "minimum": 0 }, + "extensions": { "$ref": "#/$defs/extensions" } + }, + "additionalProperties": false + }, + "stateEnvelope": { + "type": "object", + "required": ["schema", "kind", "state", "artifact", "metadata"], + "properties": { + "schema": { "const": "pto.asl-model.arch-state.v1" }, + "kind": { "enum": ["initial_state", "state_snapshot"] }, + "state": { "$ref": "#/$defs/state" }, + "artifact": { "type": "object", "additionalProperties": true }, + "metadata": { "type": "object", "additionalProperties": true } + }, + "additionalProperties": false + } + } +} diff --git a/src/pto_asl_model/paths.py b/src/pto_asl_model/paths.py new file mode 100644 index 0000000..5e6739b --- /dev/null +++ b/src/pto_asl_model/paths.py @@ -0,0 +1,42 @@ +"""Repository and dependency path resolution for the standalone model. + +The model is intentionally a separate project from both the ISA +specification and any simulator. This module is the single place where +those runtime inputs are located; callers should not infer paths from their +own source-file location. +""" + +from __future__ import annotations + +import os +from pathlib import Path + + +def repository_root() -> Path: + """Return the checkout root containing ``pyproject.toml``.""" + + return Path(__file__).resolve().parents[2] + + +def resolve_pto_spec(path: Path | str | None = None) -> Path: + """Resolve the PTO specification checkout. + + Precedence is explicit CLI/API argument, ``PTO_SPEC_ROOT``, a pinned + ``vendor/pto-spec`` checkout, then a conventional sibling checkout. The + final candidate is useful for local development but never references a + simulator repository. + """ + + if path is not None: + return Path(path).expanduser().resolve() + configured = os.environ.get("PTO_SPEC_ROOT") + if configured: + return Path(configured).expanduser().resolve() + root = repository_root() + vendored = root / "vendor" / "pto-spec" + if (vendored / ".git").exists() or (vendored / "build" / "pto-spec.asl").is_file(): + return vendored.resolve() + return (root.parent / "pto-spec").resolve() + + +__all__ = ["repository_root", "resolve_pto_spec"] diff --git a/src/pto_asl_model/state.py b/src/pto_asl_model/state.py new file mode 100644 index 0000000..f528023 --- /dev/null +++ b/src/pto_asl_model/state.py @@ -0,0 +1,566 @@ +"""Versioned, passive architectural-state serialization objects. + +These classes validate and serialize observed architectural data. They do not +initialize runner state, authorize memory access, map storage, reset execution, +or restore a live model. In particular, ``ArchitectureState.memory`` is only a +DTO representation of architectural memory data; the canonical runner and PTO +ASL remain the sole live storage and access authority. ``extensions`` is the +explicit additive channel for vendor- or instruction-family-specific fields. +""" + +from __future__ import annotations + +import dataclasses +import hashlib +import json +import math +import re +from dataclasses import dataclass, field +from typing import Any, Mapping + + +SCHEMA_ID = "pto.asl-model.arch-state.v1" +SCHEMA_VERSION = 1 + + +JsonValue = Any + +_CANONICAL_MEMORY_ADDRESS = re.compile(r"^0x(?:0|[1-9a-f][0-9a-f]*)$") +_CANONICAL_MEMORY_PERMISSIONS = frozenset( + {"r", "w", "x", "rw", "rx", "wx", "rwx"} +) + + +def _require_object(value: Any, label: str) -> dict[str, Any]: + if not isinstance(value, dict): + raise ValueError(f"{label} must be an object") + if any(not isinstance(key, str) for key in value): + raise ValueError(f"{label} keys must be text") + return value + + +def _require_exact_object( + value: Any, label: str, fields: tuple[str, ...] +) -> dict[str, Any]: + row = _require_object(value, label) + missing = [name for name in fields if name not in row] + if missing: + raise ValueError(f"{label} is missing required fields: " + ", ".join(missing)) + unexpected = sorted(set(row) - set(fields)) + if unexpected: + raise ValueError(f"{label} has unexpected fields: " + ", ".join(unexpected)) + return row + + +def _require_list(value: Any, label: str) -> list[Any]: + if not isinstance(value, list): + raise ValueError(f"{label} must be an array") + return value + + +def _require_string(value: Any, label: str) -> None: + if not isinstance(value, str): + raise ValueError(f"{label} must be text") + + +def _require_integer(value: Any, label: str, *, minimum: int | None = None) -> None: + if not isinstance(value, int) or isinstance(value, bool): + raise ValueError(f"{label} must be an integer") + if minimum is not None and value < minimum: + raise ValueError(f"{label} must be at least {minimum}") + + +def _require_optional_integer(value: Any, label: str) -> None: + if value is not None: + _require_integer(value, label) + + +def _require_boolean(value: Any, label: str) -> None: + if not isinstance(value, bool): + raise ValueError(f"{label} must be boolean") + + +def _require_json_object(value: Any, label: str) -> None: + _require_object(value, label) + canonical_json(value) + + +def _json_value(value: Any) -> JsonValue: + """Convert supported values to JSON values and reject ambiguous values.""" + + if dataclasses.is_dataclass(value): + return {key: _json_value(item) for key, item in dataclasses.asdict(value).items()} + if isinstance(value, Mapping): + if any(not isinstance(key, str) for key in value): + raise ValueError("mapping keys must be text") + return {key: _json_value(item) for key, item in value.items()} + if isinstance(value, (list, tuple)): + return [_json_value(item) for item in value] + if isinstance(value, (str, int, bool)) or value is None: + return value + if isinstance(value, float): + if not math.isfinite(value): + raise ValueError("canonical state cannot contain NaN or infinity") + return value + raise TypeError(f"value is not JSON serializable: {type(value).__name__}") + + +def canonical_json(value: Any) -> str: + """Return deterministic JSON used for snapshots, cache keys and diffs.""" + + return json.dumps( + _json_value(value), + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ) + + +def canonical_hash(value: Any) -> str: + """Return the SHA-256 of :func:`canonical_json`.""" + + return hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest() + + +@dataclass +class ScalarState: + """Scalar architectural state for one processing element/thread.""" + + registers: dict[str, int] = field(default_factory=dict) + pc: int = 0 + tpc: int = 0 + flags: dict[str, int | bool] = field(default_factory=dict) + mode: str = "" + extensions: dict[str, JsonValue] = field(default_factory=dict) + + def __post_init__(self) -> None: + registers = _require_object(self.registers, "scalar.registers") + for name, value in registers.items(): + _require_integer(value, f"scalar.registers[{name!r}]") + _require_integer(self.pc, "scalar.pc") + _require_integer(self.tpc, "scalar.tpc") + flags = _require_object(self.flags, "scalar.flags") + for name, value in flags.items(): + if not isinstance(value, (bool, int)): + raise ValueError(f"scalar.flags[{name!r}] must be boolean or integer") + _require_string(self.mode, "scalar.mode") + _require_json_object(self.extensions, "scalar.extensions") + + +@dataclass +class BlockState: + """State of the currently collected/executing Block transaction.""" + + active: bool = False + block_id: int | None = None + start_pc: int | None = None + instruction_count: int = 0 + attributes: dict[str, JsonValue] = field(default_factory=dict) + extensions: dict[str, JsonValue] = field(default_factory=dict) + + def __post_init__(self) -> None: + _require_boolean(self.active, "block.active") + _require_optional_integer(self.block_id, "block.block_id") + _require_optional_integer(self.start_pc, "block.start_pc") + _require_integer(self.instruction_count, "block.instruction_count", minimum=0) + _require_json_object(self.attributes, "block.attributes") + _require_json_object(self.extensions, "block.extensions") + + +@dataclass +class TileDescriptor: + """Descriptor needed to interpret a Tile payload.""" + + dtype: str = "" + layout: str = "" + shape: list[int] = field(default_factory=list) + valid_shape: list[int] = field(default_factory=list) + strides: list[int] = field(default_factory=list) + extensions: dict[str, JsonValue] = field(default_factory=dict) + + def __post_init__(self) -> None: + _require_string(self.dtype, "descriptor.dtype") + _require_string(self.layout, "descriptor.layout") + for label, values, minimum in ( + ("descriptor.shape", self.shape, 0), + ("descriptor.valid_shape", self.valid_shape, 0), + ("descriptor.strides", self.strides, None), + ): + for index, value in enumerate(_require_list(values, label)): + _require_integer(value, f"{label}[{index}]", minimum=minimum) + _require_json_object(self.extensions, "descriptor.extensions") + + +@dataclass +class TileValue: + """A Tile register value, including definedness and descriptor.""" + + descriptor: TileDescriptor = field(default_factory=TileDescriptor) + data: list[JsonValue] = field(default_factory=list) + defined: list[bool] = field(default_factory=list) + generation: int = 0 + extensions: dict[str, JsonValue] = field(default_factory=dict) + + def __post_init__(self) -> None: + if not isinstance(self.descriptor, TileDescriptor): + raise ValueError("tile value descriptor must be TileDescriptor") + _require_list(self.data, "tile value data") + canonical_json(self.data) + for index, value in enumerate(_require_list(self.defined, "tile value defined")): + _require_boolean(value, f"tile value defined[{index}]") + _require_integer(self.generation, "tile value generation", minimum=0) + _require_json_object(self.extensions, "tile value extensions") + + +@dataclass +class TileState: + registers: dict[str, TileValue] = field(default_factory=dict) + extensions: dict[str, JsonValue] = field(default_factory=dict) + + def __post_init__(self) -> None: + registers = _require_object(self.registers, "tile.registers") + if any(not isinstance(value, TileValue) for value in registers.values()): + raise ValueError("tile.registers values must be TileValue") + _require_json_object(self.extensions, "tile.extensions") + + +@dataclass +class SharedTile: + """One aggregate Shared Tile and its publication generation.""" + + descriptor: TileDescriptor = field(default_factory=TileDescriptor) + data: list[JsonValue] = field(default_factory=list) + defined: list[bool] = field(default_factory=list) + generation: int = 0 + allocation_mask: int = 0 + extensions: dict[str, JsonValue] = field(default_factory=dict) + + def __post_init__(self) -> None: + if not isinstance(self.descriptor, TileDescriptor): + raise ValueError("shared tile descriptor must be TileDescriptor") + _require_list(self.data, "shared tile data") + canonical_json(self.data) + for index, value in enumerate(_require_list(self.defined, "shared tile defined")): + _require_boolean(value, f"shared tile defined[{index}]") + _require_integer(self.generation, "shared tile generation", minimum=0) + _require_integer(self.allocation_mask, "shared tile allocation_mask", minimum=0) + _require_json_object(self.extensions, "shared tile extensions") + + +@dataclass +class SharedState: + tiles: dict[str, SharedTile] = field(default_factory=dict) + generations: dict[str, int] = field(default_factory=dict) + extensions: dict[str, JsonValue] = field(default_factory=dict) + + def __post_init__(self) -> None: + tiles = _require_object(self.tiles, "shared.tiles") + if any(not isinstance(value, SharedTile) for value in tiles.values()): + raise ValueError("shared.tiles values must be SharedTile") + generations = _require_object(self.generations, "shared.generations") + for name, value in generations.items(): + _require_integer(value, f"shared.generations[{name!r}]", minimum=0) + _require_json_object(self.extensions, "shared.extensions") + + +@dataclass +class MemoryRegion: + """Passive serialization metadata for an observed memory region.""" + + base: int + size: int + permissions: str = "rw" + name: str = "" + extensions: dict[str, JsonValue] = field(default_factory=dict) + + def __post_init__(self) -> None: + _require_integer(self.base, "memory region base", minimum=0) + _require_integer(self.size, "memory region size", minimum=1) + if ( + not isinstance(self.permissions, str) + or self.permissions not in _CANONICAL_MEMORY_PERMISSIONS + ): + raise ValueError("memory region permissions must be a canonical rwx subset") + _require_string(self.name, "memory region name") + _require_json_object(self.extensions, "memory region extensions") + + +@dataclass +class MemoryState: + """Passive serialized memory observations, never live runner storage.""" + + cells: dict[str, int] = field(default_factory=dict) + regions: list[MemoryRegion] = field(default_factory=list) + extensions: dict[str, JsonValue] = field(default_factory=dict) + + def __post_init__(self) -> None: + cells = _require_object(self.cells, "memory.cells") + for address, byte in cells.items(): + if _CANONICAL_MEMORY_ADDRESS.fullmatch(address) is None: + raise ValueError(f"invalid canonical memory cell address: {address!r}") + if not isinstance(byte, int) or isinstance(byte, bool) or not 0 <= byte <= 255: + raise ValueError("memory cells require byte values") + regions = _require_list(self.regions, "memory.regions") + if any(not isinstance(region, MemoryRegion) for region in regions): + raise ValueError("memory regions must contain MemoryRegion values") + ordered = sorted(self.regions, key=lambda region: region.base) + if any(left.base + left.size > right.base for left, right in zip(ordered, ordered[1:])): + raise ValueError("memory regions overlap") + _require_json_object(self.extensions, "memory.extensions") + + +@dataclass +class FaultState: + pending: bool = False + kind: str = "" + code: str = "" + address: int | None = None + instruction: str = "" + message: str = "" + recoverable: bool = False + extensions: dict[str, JsonValue] = field(default_factory=dict) + + def __post_init__(self) -> None: + _require_boolean(self.pending, "fault.pending") + _require_string(self.kind, "fault.kind") + _require_string(self.code, "fault.code") + _require_optional_integer(self.address, "fault.address") + _require_string(self.instruction, "fault.instruction") + _require_string(self.message, "fault.message") + _require_boolean(self.recoverable, "fault.recoverable") + _require_json_object(self.extensions, "fault.extensions") + + +@dataclass +class ArchitectureState: + """Complete passive architectural-state serialization data.""" + + scalar: ScalarState = field(default_factory=ScalarState) + block: BlockState = field(default_factory=BlockState) + tile: TileState = field(default_factory=TileState) + shared: SharedState = field(default_factory=SharedState) + memory: MemoryState = field(default_factory=MemoryState) + fault: FaultState = field(default_factory=FaultState) + pe_id: int = 0 + thread_id: int = 0 + cycle: int = 0 + extensions: dict[str, JsonValue] = field(default_factory=dict) + + def __post_init__(self) -> None: + domains = ( + (self.scalar, ScalarState), + (self.block, BlockState), + (self.tile, TileState), + (self.shared, SharedState), + (self.memory, MemoryState), + (self.fault, FaultState), + ) + if any(not isinstance(value, expected) for value, expected in domains): + raise ValueError("architecture state contains an invalid domain") + for label, value in ( + ("pe_id", self.pe_id), + ("thread_id", self.thread_id), + ("cycle", self.cycle), + ): + if not isinstance(value, int) or isinstance(value, bool) or value < 0: + raise ValueError(f"architecture state {label} must be non-negative integer") + _require_json_object(self.extensions, "architecture state extensions") + canonical_json(self.as_dict()) + + def as_dict(self) -> dict[str, JsonValue]: + return _json_value(self) + + def canonical_json(self) -> str: + return canonical_json(self.as_dict()) + + def sha256(self) -> str: + return canonical_hash(self.as_dict()) + + @classmethod + def from_dict(cls, value: Mapping[str, Any]) -> "ArchitectureState": + if not isinstance(value, Mapping): + raise TypeError("architecture state must be an object") + + def object_value( + item: Any, label: str, required: tuple[str, ...] = () + ) -> dict[str, Any]: + return dict(_require_exact_object( + item, f"architecture state {label}", required + )) + + def descriptor(item: Mapping[str, Any]) -> TileDescriptor: + return TileDescriptor(**object_value( + item, + "descriptor", + ("dtype", "layout", "shape", "valid_shape", "strides", "extensions"), + )) + + def tile_value(item: Mapping[str, Any]) -> TileValue: + row = object_value( + item, + "tile value", + ("descriptor", "data", "defined", "generation", "extensions"), + ) + row["descriptor"] = descriptor(row["descriptor"]) + return TileValue(**row) + + def shared_tile(item: Mapping[str, Any]) -> SharedTile: + row = object_value( + item, + "shared tile", + ( + "descriptor", "data", "defined", "generation", + "allocation_mask", "extensions", + ), + ) + row["descriptor"] = descriptor(row["descriptor"]) + return SharedTile(**row) + + required_state = ( + "scalar", "block", "tile", "shared", "memory", "fault", + "pe_id", "thread_id", "cycle", "extensions", + ) + state_row = object_value(value, "state", required_state) + scalar = ScalarState(**object_value( + state_row["scalar"], + "scalar", + ("registers", "pc", "tpc", "flags", "mode", "extensions"), + )) + block = BlockState(**object_value( + state_row["block"], + "block", + ( + "active", "block_id", "start_pc", "instruction_count", + "attributes", "extensions", + ), + )) + tile_row = object_value( + state_row["tile"], "tile", ("registers", "extensions") + ) + if not isinstance(tile_row.get("registers", {}), Mapping): + raise ValueError("architecture state tile registers must be an object") + tile_row["registers"] = { + key: tile_value(item) for key, item in tile_row["registers"].items() + } + tile = TileState(**tile_row) + shared_row = object_value( + state_row["shared"], + "shared", + ("tiles", "generations", "extensions"), + ) + if not isinstance(shared_row.get("tiles", {}), Mapping): + raise ValueError("architecture state shared tiles must be an object") + shared_row["tiles"] = { + key: shared_tile(item) for key, item in shared_row["tiles"].items() + } + shared = SharedState(**shared_row) + memory_row = object_value( + state_row["memory"], "memory", ("cells", "regions", "extensions") + ) + raw_regions = memory_row["regions"] + if not isinstance(raw_regions, list): + raise ValueError("architecture state memory regions must be an array") + memory_row["regions"] = [ + MemoryRegion(**object_value( + item, + "memory region", + ("base", "size", "permissions", "name", "extensions"), + )) + for item in raw_regions + ] + memory = MemoryState(**memory_row) + fault = FaultState(**object_value( + state_row["fault"], + "fault", + ( + "pending", "kind", "code", "address", "instruction", + "message", "recoverable", "extensions", + ), + )) + return cls(scalar=scalar, block=block, tile=tile, shared=shared, + memory=memory, fault=fault, pe_id=state_row["pe_id"], + thread_id=state_row["thread_id"], cycle=state_row["cycle"], + extensions=_require_object( + state_row["extensions"], "architecture state extensions" + )) + + +@dataclass +class StateEnvelope: + """Versioned envelope for an initial state or standalone snapshot.""" + + kind: str + state: ArchitectureState + artifact: dict[str, JsonValue] = field(default_factory=dict) + metadata: dict[str, JsonValue] = field(default_factory=dict) + schema: str = SCHEMA_ID + + def __post_init__(self) -> None: + if self.schema != SCHEMA_ID: + raise ValueError(f"unsupported state envelope schema: {self.schema!r}") + if self.kind not in {"initial_state", "state_snapshot"}: + raise ValueError(f"unsupported state envelope kind: {self.kind!r}") + if not isinstance(self.state, ArchitectureState): + raise TypeError("state envelope state must be ArchitectureState") + _require_json_object(self.artifact, "state envelope artifact") + _require_json_object(self.metadata, "state envelope metadata") + canonical_json(self.as_dict()) + + def as_dict(self) -> dict[str, JsonValue]: + return _json_value(self) + + def canonical_json(self) -> str: + return canonical_json(self.as_dict()) + + def sha256(self) -> str: + return canonical_hash(self.as_dict()) + + @classmethod + def initial(cls, state: ArchitectureState, **kwargs: Any) -> "StateEnvelope": + return cls(kind="initial_state", state=state, **kwargs) + + @classmethod + def from_dict(cls, value: Mapping[str, Any]) -> "StateEnvelope": + if not isinstance(value, Mapping): + raise TypeError("state envelope must be an object") + required = ("schema", "kind", "state", "artifact", "metadata") + row = _require_exact_object(value, "state envelope", required) + if row["schema"] != SCHEMA_ID: + raise ValueError(f"unsupported state envelope schema: {row['schema']!r}") + kind = row["kind"] + if kind not in {"initial_state", "state_snapshot"}: + raise ValueError(f"unsupported state envelope kind: {kind!r}") + state = row["state"] + if not isinstance(state, Mapping): + raise ValueError("state envelope must contain an object state") + artifact = _require_object(row["artifact"], "state envelope artifact") + metadata = _require_object(row["metadata"], "state envelope metadata") + return cls( + schema=SCHEMA_ID, + kind=kind, + state=ArchitectureState.from_dict(state), + artifact=dict(artifact), + metadata=dict(metadata), + ) + + +def state_diff(before: ArchitectureState, after: ArchitectureState) -> dict[str, JsonValue]: + """Return a compact, deterministic top-level diff for backend diagnostics.""" + + left = before.as_dict() + right = after.as_dict() + return { + key: {"before": left[key], "after": right[key]} + for key in sorted(set(left) | set(right)) + if left.get(key) != right.get(key) + } + + +__all__ = [ + "SCHEMA_ID", "SCHEMA_VERSION", "ArchitectureState", "BlockState", + "FaultState", "MemoryRegion", + "MemoryState", "ScalarState", "SharedState", "SharedTile", "StateEnvelope", + "TileDescriptor", "TileState", "TileValue", "canonical_hash", "canonical_json", + "state_diff", +] diff --git a/tests/test_asl_state.py b/tests/test_asl_state.py new file mode 100644 index 0000000..1057800 --- /dev/null +++ b/tests/test_asl_state.py @@ -0,0 +1,397 @@ +import copy +import json +import math +import re +import unittest +from pathlib import Path + +from pto_asl_model.state import ( + ArchitectureState, + FaultState, + MemoryRegion, + MemoryState, + ScalarState, + SharedState, + SharedTile, + StateEnvelope, + TileDescriptor, + TileState, + TileValue, + canonical_hash, + canonical_json, + state_diff, +) + + +ROOT = Path(__file__).resolve().parents[1] + + +def resolve_local_ref(root, reference): + if not isinstance(reference, str) or not reference.startswith("#/"): + raise AssertionError(f"unsupported schema reference: {reference!r}") + value = root + for token in reference[2:].split("/"): + token = token.replace("~1", "/").replace("~0", "~") + if not isinstance(value, dict) or token not in value: + raise AssertionError(f"unresolved schema reference: {reference}") + value = value[token] + return value + + +def validate_instance(instance, schema, root): + if "$ref" in schema: + validate_instance(instance, resolve_local_ref(root, schema["$ref"]), root) + if "const" in schema: + assert instance == schema["const"] + if "enum" in schema: + assert instance in schema["enum"] + expected = schema.get("type") + if expected is not None: + choices = expected if isinstance(expected, list) else [expected] + matches = { + "object": lambda value: isinstance(value, dict), + "array": lambda value: isinstance(value, list), + "string": lambda value: isinstance(value, str), + "integer": lambda value: isinstance(value, int) and not isinstance(value, bool), + "boolean": lambda value: isinstance(value, bool), + "null": lambda value: value is None, + } + assert any(matches[choice](instance) for choice in choices) + if isinstance(instance, int) and not isinstance(instance, bool): + if "minimum" in schema: + assert instance >= schema["minimum"] + if "maximum" in schema: + assert instance <= schema["maximum"] + if isinstance(instance, str) and "pattern" in schema: + assert re.fullmatch(schema["pattern"], instance) + if isinstance(instance, dict): + required = schema.get("required", []) + assert all(key in instance for key in required) + properties = schema.get("properties", {}) + for key, value in instance.items(): + if isinstance(schema.get("propertyNames"), dict): + validate_instance(key, schema["propertyNames"], root) + if key in properties: + validate_instance(value, properties[key], root) + elif schema.get("additionalProperties") is False: + raise AssertionError(f"unexpected property: {key}") + elif isinstance(schema.get("additionalProperties"), dict): + validate_instance(value, schema["additionalProperties"], root) + if isinstance(instance, list) and isinstance(schema.get("items"), dict): + for value in instance: + validate_instance(value, schema["items"], root) + + +class AslStateSchemaTest(unittest.TestCase): + def test_schema_is_versioned_and_covers_architecture_domains(self): + schema = json.loads( + (ROOT / "src" / "pto_asl_model" / "architecture_state.schema.json").read_text(encoding="utf-8") + ) + self.assertEqual(schema["$schema"], "https://json-schema.org/draft/2020-12/schema") + self.assertEqual(schema["$defs"]["state"]["required"], [ + "scalar", "block", "tile", "shared", "memory", "fault", + "pe_id", "thread_id", "cycle", "extensions", + ]) + for domain in ("scalar", "block", "tile", "shared", "memory", "fault"): + self.assertIn(domain, schema["$defs"]) + + def test_all_local_refs_resolve_and_remaining_envelopes_validate(self): + schema = json.loads( + (ROOT / "src" / "pto_asl_model" / "architecture_state.schema.json").read_text( + encoding="utf-8" + ) + ) + + def walk(value): + if isinstance(value, dict): + if "$ref" in value: + self.assertIsInstance(resolve_local_ref(schema, value["$ref"]), dict) + for child in value.values(): + walk(child) + elif isinstance(value, list): + for child in value: + walk(child) + + walk(schema) + for envelope in ( + StateEnvelope.initial(ArchitectureState()), + StateEnvelope(kind="state_snapshot", state=ArchitectureState()), + StateEnvelope.initial(ArchitectureState( + tile=TileState(registers={"t0": TileValue()}), + shared=SharedState(tiles={"s0": SharedTile()}, generations={"s0": 0}), + memory=MemoryState( + cells={"0x1000": 1}, + regions=[MemoryRegion(0x1000, 4, "rw", "data")], + ), + )), + ): + validate_instance(envelope.as_dict(), schema, schema) + + def test_canonical_serialization_is_order_independent(self): + left = {"z": [2, 1], "a": {"b": 2, "a": 1}} + right = {"a": {"a": 1, "b": 2}, "z": [2, 1]} + self.assertEqual(canonical_json(left), canonical_json(right)) + self.assertEqual(canonical_hash(left), canonical_hash(right)) + with self.assertRaises(ValueError): + canonical_json({"bad": math.nan}) + + def test_non_text_mapping_keys_are_rejected_without_hash_aliasing(self): + collision = {1: "integer-key", "1": "text-key"} + for operation in (canonical_json, canonical_hash): + with self.subTest(operation=operation.__name__): + with self.assertRaisesRegex(ValueError, "mapping keys must be text"): + operation(collision) + for label, construct in ( + ( + "metadata", + lambda: StateEnvelope.initial( + ArchitectureState(), metadata={"nested": collision} + ), + ), + ( + "artifact", + lambda: StateEnvelope.initial( + ArchitectureState(), artifact={"nested": collision} + ), + ), + ( + "extensions", + lambda: ArchitectureState(extensions={"nested": collision}), + ), + ("tile data", lambda: TileValue(data=[collision])), + ): + with self.subTest(channel=label): + with self.assertRaisesRegex(ValueError, "mapping keys must be text"): + construct() + + def test_memory_schema_matches_canonical_python_forms(self): + schema = json.loads( + (ROOT / "src" / "pto_asl_model" / "architecture_state.schema.json").read_text( + encoding="utf-8" + ) + ) + permissions = schema["$defs"]["region"]["properties"]["permissions"] + self.assertEqual( + permissions["enum"], ["r", "w", "x", "rw", "rx", "wx", "rwx"] + ) + property_names = schema["$defs"]["memory"]["properties"]["cells"][ + "propertyNames" + ] + self.assertEqual(property_names["pattern"], "^0x(?:0|[1-9a-f][0-9a-f]*)$") + + valid = StateEnvelope.initial( + ArchitectureState( + memory=MemoryState( + cells={"0x0": 0, "0x10": 255}, + regions=[MemoryRegion(0, 16, "rwx")], + ) + ) + ).as_dict() + validate_instance(valid, schema, schema) + + for invalid in ("", "rr", "wr", "xrw", None, ["r"]): + with self.subTest(permission=invalid): + payload = copy.deepcopy(valid) + payload["state"]["memory"]["regions"][0]["permissions"] = invalid + with self.assertRaises(AssertionError): + validate_instance(payload, schema, schema) + with self.assertRaises(ValueError): + MemoryRegion(0, 1, invalid) + + for invalid in ("0", "1", "01", "0x00", "0x01", "0X1", "-1", "0xg"): + with self.subTest(address=invalid): + payload = copy.deepcopy(valid) + payload["state"]["memory"]["cells"] = {invalid: 1} + with self.assertRaises(AssertionError): + validate_instance(payload, schema, schema) + with self.assertRaises(ValueError): + MemoryState(cells={invalid: 1}) + + for invalid in (-1, 256, True): + with self.subTest(byte=invalid): + payload = copy.deepcopy(valid) + payload["state"]["memory"]["cells"] = {"0x0": invalid} + with self.assertRaises(AssertionError): + validate_instance(payload, schema, schema) + with self.assertRaises(ValueError): + MemoryState(cells={"0x0": invalid}) + + def test_state_round_trip_preserves_tile_memory_and_extensions(self): + state = ArchitectureState( + scalar=ScalarState(registers={"x1": 7}, pc=0x100, tpc=0x200), + tile=TileState( + registers={ + "t0": TileValue( + descriptor=TileDescriptor( + dtype="fp16", layout="row_major", shape=[2, 2], + valid_shape=[1, 2], strides=[2, 1], + ), + data=[1, 2, 0, 0], defined=[True, True, False, False], + generation=3, + ) + } + ), + memory=MemoryState( + cells={"0x1000": 42}, regions=[MemoryRegion(0x1000, 16, "rw", "input")] + ), + fault=FaultState(), + extensions={"future.domain": {"version": 2}}, + ) + restored = ArchitectureState.from_dict(state.as_dict()) + self.assertEqual(restored.as_dict(), state.as_dict()) + self.assertEqual(restored.sha256(), state.sha256()) + + def test_state_envelopes_are_deterministic(self): + initial = StateEnvelope.initial(ArchitectureState(), artifact={"commit": "abc"}) + payload = initial.as_dict() + self.assertEqual(payload["schema"], "pto.asl-model.arch-state.v1") + self.assertEqual(payload["kind"], "initial_state") + self.assertEqual(initial.sha256(), canonical_hash(payload)) + restored = StateEnvelope.from_dict(payload) + self.assertEqual(restored.as_dict(), payload) + + def test_nested_dto_constructors_reject_schema_type_violations(self): + with self.assertRaisesRegex(ValueError, "scalar.flags"): + ScalarState(flags={"bad": "true"}) + with self.assertRaisesRegex(ValueError, "scalar.flags"): + ScalarState(flags={"bad": {}}) + with self.assertRaisesRegex(ValueError, "fault.pending"): + FaultState(pending=1) + with self.assertRaisesRegex(ValueError, "tile value generation"): + TileValue(generation=-1) + + def test_nested_state_round_trip_rejects_schema_violations(self): + state = ArchitectureState( + tile=TileState(registers={"t0": TileValue()}), + shared=SharedState( + tiles={"s0": SharedTile()}, generations={"s0": 0} + ), + memory=MemoryState( + cells={"0x1000": 1}, + regions=[MemoryRegion(0x1000, 4, "rw", "data")], + ), + ) + base = StateEnvelope.initial(state).as_dict() + cases = ( + (("state", "scalar", "flags"), {"bad": "yes"}, "scalar.flags"), + (("state", "scalar", "flags"), {"bad": {}}, "scalar.flags"), + (("state", "scalar", "mode"), 7, "scalar.mode"), + (("state", "block", "active"), 1, "block.active"), + (("state", "block", "block_id"), True, "block.block_id"), + (("state", "block", "instruction_count"), -1, "block.instruction_count"), + (("state", "tile", "registers"), [], "tile registers"), + (("state", "tile", "registers", "t0", "descriptor", "dtype"), [], "dtype"), + (("state", "tile", "registers", "t0", "descriptor", "shape"), [-1], "shape"), + (("state", "tile", "registers", "t0", "descriptor", "strides"), [True], "strides"), + (("state", "tile", "registers", "t0", "data"), {}, "data"), + (("state", "tile", "registers", "t0", "defined"), [1], "defined"), + (("state", "tile", "registers", "t0", "generation"), -1, "generation"), + (("state", "shared", "tiles", "s0", "generation"), -1, "generation"), + (("state", "shared", "tiles", "s0", "allocation_mask"), -1, "allocation_mask"), + (("state", "shared", "tiles"), [], "shared tiles"), + (("state", "shared", "generations", "s0"), -1, "shared.generations"), + (("state", "memory", "cells", "0x1000"), 999, "memory cells"), + (("state", "memory", "regions"), {}, "memory regions"), + (("state", "fault", "pending"), 1, "fault.pending"), + (("state", "fault", "address"), True, "fault.address"), + (("state", "fault", "message"), {}, "fault.message"), + (("state", "fault", "kind"), [], "fault.kind"), + (("state", "fault", "recoverable"), 0, "fault.recoverable"), + (("state", "extensions"), [], "extensions"), + ) + for path, replacement, message in cases: + with self.subTest(path=path): + payload = copy.deepcopy(base) + target = payload + for key in path[:-1]: + target = target[key] + target[path[-1]] = replacement + with self.assertRaisesRegex((TypeError, ValueError), message): + StateEnvelope.from_dict(payload) + + def test_additional_properties_are_rejected_at_every_closed_dto(self): + state = ArchitectureState( + tile=TileState(registers={"t0": TileValue()}), + shared=SharedState(tiles={"s0": SharedTile()}, generations={"s0": 0}), + memory=MemoryState( + cells={"0x1000": 1}, + regions=[MemoryRegion(0x1000, 4, "rw", "data")], + ), + ) + base = StateEnvelope.initial(state).as_dict() + closed_paths = ( + (), + ("state",), + ("state", "scalar"), + ("state", "block"), + ("state", "tile"), + ("state", "tile", "registers", "t0"), + ("state", "tile", "registers", "t0", "descriptor"), + ("state", "shared"), + ("state", "shared", "tiles", "s0"), + ("state", "memory"), + ("state", "memory", "regions", 0), + ("state", "fault"), + ) + for path in closed_paths: + with self.subTest(path=path): + payload = copy.deepcopy(base) + target = payload + for key in path: + target = target[key] + target["unexpected"] = 1 + with self.assertRaisesRegex(ValueError, "unexpected fields: unexpected"): + StateEnvelope.from_dict(payload) + + def test_extensions_are_the_positive_additive_round_trip_channel(self): + descriptor = TileDescriptor(extensions={"descriptor.future": {"v": 1}}) + state = ArchitectureState( + scalar=ScalarState(extensions={"scalar.future": True}), + tile=TileState( + registers={ + "t0": TileValue( + descriptor=descriptor, + extensions={"tile_value.future": [1, 2]}, + ) + }, + extensions={"tile.future": None}, + ), + shared=SharedState( + tiles={ + "s0": SharedTile( + descriptor=descriptor, + extensions={"shared_tile.future": "ok"}, + ) + }, + extensions={"shared.future": 3}, + ), + memory=MemoryState( + regions=[ + MemoryRegion( + 0x1000, + 4, + "rw", + "data", + extensions={"region.future": False}, + ) + ], + extensions={"memory.future": {}}, + ), + fault=FaultState(extensions={"fault.future": 1.5}), + extensions={"state.future": {"enabled": True}}, + ) + envelope = StateEnvelope.initial(state) + restored = StateEnvelope.from_dict(envelope.as_dict()) + self.assertEqual(restored.as_dict(), envelope.as_dict()) + + def test_state_diff_only_reports_changed_domains(self): + before = ArchitectureState() + after = ArchitectureState(scalar=ScalarState(pc=4)) + diff = state_diff(before, after) + self.assertEqual(set(diff), {"scalar"}) + self.assertEqual(diff["scalar"]["before"]["pc"], 0) + self.assertEqual(diff["scalar"]["after"]["pc"], 4) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_paths.py b/tests/test_paths.py new file mode 100644 index 0000000..bd0c908 --- /dev/null +++ b/tests/test_paths.py @@ -0,0 +1,30 @@ +import os +import tempfile +import unittest +from pathlib import Path + +from pto_asl_model.paths import resolve_pto_spec, repository_root + + +class PathResolutionTests(unittest.TestCase): + def test_repository_root_is_model_checkout(self): + root = repository_root() + self.assertTrue((root / "CMakeLists.txt").is_file()) + self.assertTrue((root / "src" / "pto_asl_model").is_dir()) + + def test_explicit_paths_win_over_environment(self): + with tempfile.TemporaryDirectory() as directory: + explicit = Path(directory) / "explicit" + configured = Path(directory) / "configured" + old = os.environ.get("PTO_SPEC_ROOT") + try: + os.environ["PTO_SPEC_ROOT"] = str(configured) + self.assertEqual(resolve_pto_spec(explicit), explicit.resolve()) + finally: + if old is None: + os.environ.pop("PTO_SPEC_ROOT", None) + else: + os.environ["PTO_SPEC_ROOT"] = old + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_standalone_import.py b/tests/test_standalone_import.py new file mode 100644 index 0000000..5f895a0 --- /dev/null +++ b/tests/test_standalone_import.py @@ -0,0 +1,42 @@ +import sys +import tomllib +import unittest +from importlib.util import find_spec +from pathlib import Path + + +PACKAGE_ROOT = Path(__file__).resolve().parents[1] +SRC = PACKAGE_ROOT / "src" +if str(SRC) not in sys.path: + sys.path.insert(0, str(SRC)) + + +class StandaloneImportTests(unittest.TestCase): + def test_package_import_does_not_need_superproject(self): + import pto_asl_model + + for name in ( + "AslSession", + "GuestMemory", + "HostMemoryBridge", + "ProgramImage", + "RuntimeAdapter", + "RuntimeSnapshot", + "RuntimeState", + "InstructionTransaction", + "SemanticBackend", + ): + self.assertFalse(hasattr(pto_asl_model, name)) + self.assertIsNone(find_spec("pto_asl_model.runtime")) + self.assertEqual(Path(pto_asl_model.__file__).resolve().parent.name, "pto_asl_model") + + def test_public_runner_remains_the_hosted_runner(self): + import pto_asl_model + + self.assertEqual(pto_asl_model.run.__module__, "pto_asl_model.runner") + project = tomllib.loads((PACKAGE_ROOT / "pyproject.toml").read_text()) + self.assertNotIn("scripts", project["project"]) + + +if __name__ == "__main__": + unittest.main()