diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 5fc4b47..c4227b5 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -36,13 +36,13 @@ jobs: uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 with: repository: fruwehq/determa-state-conformance - ref: ffbc65cbce49733803119a7dabf02a9727819ba8 + ref: 707a49ce01c6f57f673c1959cdfe078bc8d0fc9a path: .pinned/determa-state-conformance - name: Check out pinned specification uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 with: repository: fruwehq/determa-state-spec - ref: 4bd4d9588d11b75d376380b6120676a056a4bc45 + ref: 1502a58a780d837e05bfacb37680dfc92e3488b5 path: .pinned/determa-state-spec - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 with: diff --git a/README.md b/README.md index be628d1..6785e70 100644 --- a/README.md +++ b/README.md @@ -4,9 +4,9 @@ Python implementation of [Determa State](https://github.com/fruwehq/determa-stat a language-agnostic statechart engine with a shared normative conformance suite. This pre-release implements Determa State `format: 1` at the approved specification -commit `4bd4d9588d11b75d376380b6120676a056a4bc45`. Correctness is determined by the -88-case core suite at conformance commit -`ffbc65cbce49733803119a7dabf02a9727819ba8`. +commit `1502a58a780d837e05bfacb37680dfc92e3488b5`. Correctness is determined by the +110-case core suite and persistence profiles at conformance commit +`707a49ce01c6f57f673c1959cdfe078bc8d0fc9a`. The package metadata is `0.0.7` for the next synchronized release of the specification, conformance suite, Python engine, and Rust engine. @@ -121,6 +121,30 @@ the exact supplied state object. semantic validation path. Native values must satisfy the same portable Unicode and numeric domain as source documents. +## Persist And Migrate + +`serialize_aggregate` produces the canonical §16 aggregate artifact. Restoration +resolves its exact validated definition by fingerprint and fails closed when the +definition is absent or untrusted: + +```python +resolver = ds.MemoryArtifactResolver(definitions={bundle.fingerprint: bundle}) +encoded = ds.serialize_aggregate(bundle, state) +restored = ds.restore_aggregate(encoded, resolver) +``` + +`restore_aggregate_package` verifies a self-contained transport package and seeds a +mutable resolver without replacing existing content. `migrate_aggregate` applies an +exact trusted descriptor route as a pure operation. `migrate_and_dispatch` returns one +commit-ready migration, audit, dispatch, aggregate, and outbox-intent boundary. Failed +migrations return a deterministic `MigrationFailure` and do not mutate the supplied +artifact or resolver. + +Definition and descriptor resolvers are protocols, so applications can back them with +an immutable registry or a transaction-local cache. Database schemas, broker +acknowledgement, retries, and quarantine remain host responsibilities; the conformance +persistence profile verifies the required transaction ordering. + ## Implemented Core - strict format-1 loading, default materialization, bundle fingerprinting, and exact @@ -133,13 +157,14 @@ numeric domain as source documents. - owned spawn, nominal instance references, binding, cancellation, completion, failure propagation, and cleanup cascades; - atomic RTC rollback, deterministic identities/counters, pure inspection, and - incompatible or malformed prior-state rejection. + incompatible or malformed prior-state rejection; +- canonical aggregate serialization/restoration, portable typed values, package + attachments, exact definition resolution, trusted lazy migration, deterministic + audits, resource limits, and atomic migrate-and-dispatch results. Format 1 deliberately does not define native queues, timers, deferral, dead letters, -stores, snapshot wire encoding, machine hot-swap/migration, package imports, -standardized enabled-event inspection, or a standardized execution CLI. Hosts may -persist the returned logical aggregate in their own transaction, but portable -serialization and definition migration remain separate specification work. +database schemas, package imports, standardized enabled-event inspection, or a +standardized execution CLI. The implementation-local CLI only validates a bundle: diff --git a/conformance/persistence.py b/conformance/persistence.py new file mode 100644 index 0000000..70924f4 --- /dev/null +++ b/conformance/persistence.py @@ -0,0 +1,280 @@ +"""Driver for portable aggregate and migration conformance vectors.""" + +from __future__ import annotations + +import copy +from pathlib import Path +from typing import Any + +import yaml + +from determa.state import ( + ArtifactError, + MemoryArtifactResolver, + MigrationLimits, + aggregate_envelope, + create, + dispatch, + load_bundle, + migrate_aggregate, + migrate_and_dispatch, + restore_aggregate, + restore_aggregate_package, + serialize_aggregate, +) +from determa.state.wire import ( + canonical_bytes, + migration_descriptor_digest, + strict_json, +) + +from .harness import CoreCase + + +def persistence_vector_cases(cases: list[CoreCase]) -> list[CoreCase]: + return [ + case + for case in cases + if (_load_yaml(case.test_file).get("persistence_vectors") or []) + ] + + +def run_persistence_vectors(case: CoreCase) -> None: + test = _load_yaml(case.test_file) + for vector in test.get("persistence_vectors") or []: + _run_vector(case.path, vector) + + +def _load_yaml(path: Path) -> dict[str, Any]: + return yaml.safe_load(path.read_text(encoding="utf-8")) or {} + + +def _json(path: Path) -> Any: + value, _ = strict_json(path.read_bytes()) + return value + + +def _resolver(path: Path, vector: dict[str, Any]) -> MemoryArtifactResolver: + if "artifact_resolver" in vector: + fixture = _json(path / vector["artifact_resolver"]) + definitions = { + entry["validated_bundle_fingerprint"]: (path / entry["bundle_file"]).read_text( + encoding="utf-8" + ) + for entry in fixture["definitions"] + } + descriptors = { + entry["migration_descriptor_digest"]: ( + path / entry["descriptor_file"] + ).read_bytes() + for entry in fixture["migration_descriptors"] + } + return MemoryArtifactResolver( + definitions=definitions, + migration_descriptors=descriptors, + trusted_definitions=[ + entry["validated_bundle_fingerprint"] + for entry in fixture["definitions"] + if entry["trusted"] + ], + trusted_migration_descriptors=[ + entry["migration_descriptor_digest"] + for entry in fixture["migration_descriptors"] + if entry["trusted"] + ], + ) + definitions: dict[str, Any] = {} + for filename in vector.get("definitions") or []: + bundle = load_bundle((path / filename).read_text(encoding="utf-8")) + definitions[bundle.fingerprint] = bundle + descriptors: dict[str, Any] = {} + for filename in vector.get("migration_descriptors") or []: + document = _json(path / filename) + digest = document.get("migration_descriptor_digest") + descriptors[ + digest if isinstance(digest, str) else migration_descriptor_digest(document) + ] = document + return MemoryArtifactResolver( + definitions=definitions, migration_descriptors=descriptors + ) + + +def _limits(path: Path, vector: dict[str, Any]) -> MigrationLimits | None: + filename = vector.get("resource_limits") + return MigrationLimits.from_mapping(_json(path / filename)) if filename else None + + +def _run_vector(path: Path, vector: dict[str, Any]) -> None: + operation = vector["operation"] + expected = vector["expect"] + source_key = ( + "aggregate_state_package" + if operation.startswith("restore_package") + else "aggregate_state" + ) + source = (path / vector[source_key]).read_bytes() if source_key in vector else None + source_snapshot = bytes(source) if source is not None else None + resolver = _resolver(path, vector) + try: + if operation == "serialize_created_aggregate": + bundle = load_bundle((path / vector["source_bundle"]).read_text(encoding="utf-8")) + result = create(bundle, **vector["creation"]) + assert result["state"] is not None + document = aggregate_envelope(bundle, result["state"]) + encoded = serialize_aggregate(bundle, result["state"]) + audits: tuple[dict[str, Any], ...] = () + emissions: tuple[dict[str, Any], ...] = () + disposition = None + elif operation == "restore_and_serialize": + assert source is not None + restored = restore_aggregate(source, resolver) + document = aggregate_envelope(restored.bundle, restored.state) + encoded = serialize_aggregate(restored.bundle, restored.state) + audits = () + emissions = () + disposition = None + elif operation == "restore_and_dispatch": + assert source is not None + restored = restore_aggregate(source, resolver) + core = dispatch( + restored.bundle, + restored.state, + {"input": _json(path / vector["input_envelope"])}, + ) + assert core["state"] is not None + document = aggregate_envelope(restored.bundle, core["state"]) + encoded = canonical_bytes(document) + audits = () + emissions = tuple(core["emissions"]) + disposition = core["disposition"] + elif operation == "restore_package": + assert source is not None + package = restore_aggregate_package(source, resolver) + document = package.aggregate.aggregate_envelope + encoded = package.aggregate.canonical_bytes + audits = () + emissions = () + disposition = None + elif operation == "restore_package_and_migrate": + assert source is not None + package = restore_aggregate_package(source, resolver) + migrated = migrate_aggregate( + package.aggregate.canonical_bytes, + vector["target_validated_bundle_fingerprint"], + vector["migration_route"], + resolver, + maintenance_mode=vector["maintenance_mode"], + resource_limits=_limits(path, vector), + ) + if migrated.failure is not None: + raise ArtifactError(migrated.failure.code) + assert migrated.aggregate_envelope is not None + assert migrated.aggregate_bytes is not None + document = migrated.aggregate_envelope + encoded = migrated.aggregate_bytes + audits = migrated.audit_records + emissions = () + disposition = None + elif operation == "migrate_aggregate": + assert source is not None + request = { + "target_validated_bundle_fingerprint": vector.get( + "target_validated_bundle_fingerprint" + ), + "migration_route": vector.get("migration_route"), + "maintenance_mode": vector.get("maintenance_mode"), + } + if "migration_request" in vector: + request_value = vector["migration_request"] + request = ( + _json(path / request_value) + if isinstance(request_value, str) + else copy.deepcopy(request_value) + ) + migrated = migrate_aggregate( + source, + request.get("target_validated_bundle_fingerprint"), + request.get("migration_route"), + resolver, + maintenance_mode=request.get("maintenance_mode"), + resource_limits=_limits(path, vector), + ) + if migrated.failure is not None: + raise ArtifactError(migrated.failure.code) + assert migrated.aggregate_envelope is not None + assert migrated.aggregate_bytes is not None + document = migrated.aggregate_envelope + encoded = migrated.aggregate_bytes + audits = migrated.audit_records + emissions = () + disposition = None + elif operation == "migrate_and_dispatch": + assert source is not None + migrated_dispatch = migrate_and_dispatch( + source, + vector["target_validated_bundle_fingerprint"], + vector["migration_route"], + resolver, + {"input": _json(path / vector["input_envelope"])}, + maintenance_mode=vector["maintenance_mode"], + resource_limits=_limits(path, vector), + ) + if migrated_dispatch.failure is not None: + raise ArtifactError(migrated_dispatch.failure.code) + assert migrated_dispatch.aggregate_envelope is not None + assert migrated_dispatch.aggregate_bytes is not None + document = migrated_dispatch.aggregate_envelope + encoded = migrated_dispatch.aggregate_bytes + audits = migrated_dispatch.audit_records + emissions = migrated_dispatch.emissions + disposition = migrated_dispatch.disposition + else: + raise AssertionError(f"unsupported persistence operation: {operation}") + except ArtifactError as error: + assert expected["result"] == "failure", (vector["name"], error.code) + assert error.code == expected["code"], (vector["name"], error.code) + if expected.get("caller_still_owns_aggregate"): + assert source == source_snapshot + if vector.get("repeat_count", 1) > 1: + repeated = copy.deepcopy(vector) + repeated["repeat_count"] -= 1 + _run_vector(path, repeated) + return + assert expected["result"] == "success", vector["name"] + assert document == _json(path / expected["aggregate_state_file"]) + assert encoded == (path / expected["exact_bytes_file"]).read_bytes() + if "migration_audit_file" in expected: + assert list(audits) == _json(path / expected["migration_audit_file"]) + if "emissions_file" in expected: + assert list(emissions) == _json(path / expected["emissions_file"]) + if "disposition" in expected: + assert disposition == expected["disposition"] + if "artifact_resolver_file" in expected: + _assert_resolver(path, resolver, expected["artifact_resolver_file"]) + if vector.get("repeat_count", 1) > 1: + repeated = copy.deepcopy(vector) + repeated["repeat_count"] -= 1 + _run_vector(path, repeated) + + +def _assert_resolver( + path: Path, resolver: MemoryArtifactResolver, fixture_name: str +) -> None: + fixture = _json(path / fixture_name) + assert resolver.snapshot() == { + "definitions": sorted( + entry["validated_bundle_fingerprint"] for entry in fixture["definitions"] + ), + "migration_descriptors": sorted( + entry["migration_descriptor_digest"] + for entry in fixture["migration_descriptors"] + ), + } + for entry in fixture["definitions"]: + assert resolver.definition_is_trusted(entry["validated_bundle_fingerprint"]) is entry[ + "trusted" + ] + for entry in fixture["migration_descriptors"]: + assert resolver.migration_descriptor_is_trusted( + entry["migration_descriptor_digest"] + ) is entry["trusted"] diff --git a/conformance/persistence_profiles.py b/conformance/persistence_profiles.py new file mode 100644 index 0000000..11b527e --- /dev/null +++ b/conformance/persistence_profiles.py @@ -0,0 +1,189 @@ +"""In-memory transactional driver for the optional persistence profile.""" + +from __future__ import annotations + +import copy +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import yaml + +from determa.state import MemoryArtifactResolver, load_bundle, migrate_and_dispatch +from determa.state.wire import migration_descriptor_digest, strict_json + +from .harness import conformance_root + +PROFILE_DIR = conformance_root() / "conformance" / "profiles" / "persistence" + +_RESOLVED = [ + "resolve_target_definition", + "resolve_route", + "resolve_descriptors", + "verify_trust", +] +_TRANSACTION = ["begin_transaction", "lock_aggregate", "read_inbox"] +_COMMITTED = [ + *_RESOLVED, + *_TRANSACTION, + "migrate_aggregate", + "dispatch_once", + "stage_aggregate", + "stage_inbox", + "stage_outbox", + "stage_audit", + "commit", +] +_ROLLED_BACK = [*_RESOLVED, *_TRANSACTION, "migrate_aggregate", "rollback"] +_REPLAYED = [*_RESOLVED, *_TRANSACTION, "return_recorded_outcome", "commit"] +_QUARANTINED = [ + *_RESOLVED, + *_TRANSACTION, + "migration_failed_permanently", + "stage_blocked_inbox", + "stage_quarantine", + "stage_failure_audit", + "commit", +] + + +@dataclass(frozen=True) +class PersistenceProfileCase: + name: str + path: Path + + +def persistence_profile_cases() -> list[PersistenceProfileCase]: + if not PROFILE_DIR.exists(): + return [] + return [ + PersistenceProfileCase(path.name, path) + for path in sorted(PROFILE_DIR.iterdir()) + if path.is_dir() and (path / "test.yaml").exists() + ] + + +def run_persistence_profile(case: PersistenceProfileCase) -> None: + test = yaml.safe_load((case.path / "test.yaml").read_text(encoding="utf-8")) + profile = test["persistence_profile"] + store = _json(case.path / profile["initial_store"]) + resolver = _profile_resolver(case.path) + for step in profile["steps"]: + envelope = _json(case.path / step["input_envelope"]) + event_id = envelope["event_id"] + operation = step["operation"] + if operation == "replay": + assert any(item["event_id"] == event_id for item in store["inbox"]) + call_log = list(_REPLAYED) + if event_id not in store["acknowledged_event_ids"]: + store["acknowledged_event_ids"].append(event_id) + call_log.append("acknowledge_input") + elif ( + operation == "inject_failure" + and step["failure_class"] == "permanent" + ): + call_log = list(_QUARANTINED) + store["inbox"].append({"event_id": event_id, "status": "blocked"}) + store["quarantine"].append( + { + "event_id": event_id, + "code": "migration_totality_failure", + "aggregate_state_digest": store["aggregate_state"][ + "aggregate_state_digest" + ], + } + ) + else: + candidate = _process( + store, + envelope, + resolver, + step["target_validated_bundle_fingerprint"], + step["migration_route"], + ) + boundary = step.get("failure_boundary") + if operation == "inject_failure" and boundary != ( + "after_commit_before_acknowledgement" + ): + call_log = list(_ROLLED_BACK) + else: + store = candidate + call_log = list(_COMMITTED) + if boundary != "after_commit_before_acknowledgement": + store["acknowledged_event_ids"].append(event_id) + call_log.append("acknowledge_input") + assert store == _json(case.path / step["expect_store"]), step + assert call_log == _json(case.path / step["expect_call_log"]), step + + +def _process( + store: dict[str, Any], + envelope: dict[str, Any], + resolver: MemoryArtifactResolver, + target_fingerprint: str, + route: list[str], +) -> dict[str, Any]: + target_bundle = resolver.resolve_definition(target_fingerprint) + assert target_bundle is not None + bundle = target_bundle if hasattr(target_bundle, "raw") else load_bundle(target_bundle) + event_declaration = (bundle.raw.get("events") or {}).get(envelope["event"]) + delivery_kind = ( + "input" + if isinstance(event_declaration, dict) + and event_declaration.get("direction") == "input" + else "internal" + ) + result = migrate_and_dispatch( + store["aggregate_state"], + target_fingerprint, + route, + resolver, + {delivery_kind: copy.deepcopy(envelope)}, + maintenance_mode=False, + ) + assert result.failure is None + assert result.aggregate_envelope is not None + disposition = ( + "unhandled" + if result.disposition == "rejected" + and result.rejection == {"code": "invalid_event"} + else result.disposition + ) + candidate = copy.deepcopy(store) + candidate["aggregate_state"] = result.aggregate_envelope + candidate["inbox"] = [ + item for item in candidate["inbox"] if item["event_id"] != envelope["event_id"] + ] + candidate["inbox"].append( + { + "event_id": envelope["event_id"], + "disposition": disposition, + "status": "committed", + } + ) + candidate["outbox"].extend(copy.deepcopy(result.emissions)) + candidate["migration_audit"].extend(copy.deepcopy(result.audit_records)) + candidate["quarantine"] = [ + item + for item in candidate["quarantine"] + if item["event_id"] != envelope["event_id"] + ] + return candidate + + +def _profile_resolver(path: Path) -> MemoryArtifactResolver: + definitions = {} + for filename in ("machine.yaml", "target.yaml"): + bundle = load_bundle((path / filename).read_text(encoding="utf-8")) + definitions[bundle.fingerprint] = bundle + descriptor = _json(path / "migration-descriptor.json") + digest = migration_descriptor_digest(descriptor) + return MemoryArtifactResolver( + definitions=definitions, + migration_descriptors={digest: descriptor}, + ) + + +def _json(path: Path) -> Any: + value, _ = strict_json(path.read_bytes()) + return value diff --git a/conformance/pins.py b/conformance/pins.py index 0ee4e10..0674c31 100644 --- a/conformance/pins.py +++ b/conformance/pins.py @@ -4,8 +4,8 @@ from pathlib import Path -CONFORMANCE_COMMIT = "ffbc65cbce49733803119a7dabf02a9727819ba8" -SPEC_COMMIT = "4bd4d9588d11b75d376380b6120676a056a4bc45" +CONFORMANCE_COMMIT = "707a49ce01c6f57f673c1959cdfe078bc8d0fc9a" +SPEC_COMMIT = "1502a58a780d837e05bfacb37680dfc92e3488b5" ROOT = Path(__file__).resolve().parent.parent CONFORMANCE_CACHE = ROOT / ".cache" / f"determa-state-conformance-{CONFORMANCE_COMMIT[:12]}" diff --git a/conformance/test_conformance.py b/conformance/test_conformance.py index 86b4643..e74e6fe 100644 --- a/conformance/test_conformance.py +++ b/conformance/test_conformance.py @@ -11,8 +11,14 @@ from determa.state import load_bundle from determa.state.validator import schema as bundled_schema +from determa.state.wire import artifact_schema from .harness import CORE_DIR, CoreCase, core_cases, run_case +from .persistence import persistence_vector_cases, run_persistence_vectors +from .persistence_profiles import ( + persistence_profile_cases, + run_persistence_profile, +) def _spec_schema() -> dict | None: @@ -33,7 +39,7 @@ def _spec_root() -> Path | None: def test_suite_present() -> None: assert CORE_DIR.exists(), "pinned conformance suite is unavailable" - assert len(core_cases()) == 88 + assert len(core_cases()) == 110 def test_bundled_schema_matches_pinned_spec() -> None: @@ -42,6 +48,28 @@ def test_bundled_schema_matches_pinned_spec() -> None: assert bundled_schema() == upstream +@pytest.mark.parametrize( + ("name", "kind"), + [ + ("aggregate-state.schema.json", "aggregate_state"), + ("migration-descriptor.schema.json", "migration_descriptor"), + ("aggregate-state-package.schema.json", "aggregate_state_package"), + ], +) +def test_bundled_artifact_schemas_match_pinned_spec(name: str, kind: str) -> None: + root = _spec_root() + assert root is not None, "pinned specification is unavailable" + upstream = json.loads((root / "schema" / name).read_text(encoding="utf-8")) + assert artifact_schema(kind) == upstream + + +@pytest.mark.parametrize( + "kind", ["aggregate_state", "migration_descriptor", "aggregate_state_package"] +) +def test_bundled_artifact_schema_is_valid_draft_2020_12(kind: str) -> None: + Draft202012Validator.check_schema(artifact_schema(kind)) + + def test_bundled_schema_is_valid_draft_2020_12() -> None: Draft202012Validator.check_schema(bundled_schema()) @@ -59,3 +87,17 @@ def test_authoritative_spec_examples_load_semantically(name: str) -> None: @pytest.mark.parametrize("case", core_cases(), ids=lambda case: case.name) def test_core_case(case: CoreCase) -> None: run_case(case) + + +@pytest.mark.parametrize( + "case", persistence_vector_cases(core_cases()), ids=lambda case: case.name +) +def test_persistence_vectors(case: CoreCase) -> None: + run_persistence_vectors(case) + + +@pytest.mark.parametrize( + "case", persistence_profile_cases(), ids=lambda case: case.name +) +def test_persistence_profile(case) -> None: + run_persistence_profile(case) diff --git a/pyproject.toml b/pyproject.toml index 8d10a20..a3fd6e8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,6 +26,7 @@ dependencies = [ "pyyaml>=6", "jsonschema>=4", "cel-python>=0.1", + "rfc8785>=0.1.4", ] [project.urls] @@ -57,6 +58,9 @@ packages = ["src/determa"] [tool.hatch.build.targets.wheel.force-include] "src/determa/state/data/machine.schema.json" = "determa/state/data/machine.schema.json" +"src/determa/state/data/aggregate-state.schema.json" = "determa/state/data/aggregate-state.schema.json" +"src/determa/state/data/migration-descriptor.schema.json" = "determa/state/data/migration-descriptor.schema.json" +"src/determa/state/data/aggregate-state-package.schema.json" = "determa/state/data/aggregate-state-package.schema.json" [tool.ruff] line-length = 100 diff --git a/scripts/sync_schema.py b/scripts/sync_schema.py index a0f93d3..a752ed3 100644 --- a/scripts/sync_schema.py +++ b/scripts/sync_schema.py @@ -1,9 +1,9 @@ #!/usr/bin/env python3 -"""Refresh the bundled JSON Schema from the approved immutable specification commit. +"""Refresh bundled JSON Schemas from the approved immutable specification commit. -Writes ``src/determa/state/data/machine.schema.json`` from Determa State's -``schema/machine.schema.json`` at the format-1 pre-release pin, or from a local checkout -via ``DETERMA_SPEC_DIR``. The schema-drift conformance test guards that they match. +Writes the machine and persistence artifact schemas from Determa State's ``schema/`` +directory, or from a local checkout via ``DETERMA_SPEC_DIR``. Schema-drift conformance +tests guard that all copies match. Usage: ``python scripts/sync_schema.py`` (or ``make sync-schema``). """ @@ -18,17 +18,23 @@ from pathlib import Path ROOT = Path(__file__).resolve().parent.parent -DEST = ROOT / "src" / "determa" / "state" / "data" / "machine.schema.json" -SPEC_COMMIT = "4bd4d9588d11b75d376380b6120676a056a4bc45" +DEST = ROOT / "src" / "determa" / "state" / "data" +SCHEMAS = ( + "machine.schema.json", + "aggregate-state.schema.json", + "migration-descriptor.schema.json", + "aggregate-state-package.schema.json", +) +SPEC_COMMIT = "1502a58a780d837e05bfacb37680dfc92e3488b5" -def _fetch() -> str: +def _fetch(name: str) -> str: override = os.environ.get("DETERMA_SPEC_DIR") if override: - return (Path(override) / "schema" / "machine.schema.json").read_text(encoding="utf-8") + return (Path(override) / "schema" / name).read_text(encoding="utf-8") url = ( "https://raw.githubusercontent.com/fruwehq/determa-state-spec/" - f"{SPEC_COMMIT}/schema/machine.schema.json" + f"{SPEC_COMMIT}/schema/{name}" ) try: with urllib.request.urlopen(url, timeout=10) as response: # noqa: S310 (fixed host) @@ -38,13 +44,15 @@ def _fetch() -> str: def main() -> int: - text = _fetch() - json.loads(text) # sanity check: valid JSON before overwriting - if DEST.read_text(encoding="utf-8") == text: - print(f"{DEST.relative_to(ROOT)} already up to date") - return 0 - DEST.write_text(text, encoding="utf-8") - print(f"updated {DEST.relative_to(ROOT)}") + for name in SCHEMAS: + text = _fetch(name) + json.loads(text) + destination = DEST / name + if destination.read_text(encoding="utf-8") == text: + print(f"{destination.relative_to(ROOT)} already up to date") + continue + destination.write_text(text, encoding="utf-8") + print(f"updated {destination.relative_to(ROOT)}") return 0 diff --git a/src/determa/state/__init__.py b/src/determa/state/__init__.py index eca87b8..11b4d5d 100644 --- a/src/determa/state/__init__.py +++ b/src/determa/state/__init__.py @@ -7,24 +7,70 @@ from .__about__ import __version__ from .definition import Bundle, BundleSource, load_bundle from .engine import Delivery, Result, create, dispatch -from .errors import CelError, DetermaError, ErrorRecord, SchemaError, ValidationError +from .errors import ( + ArtifactError, + CelError, + DetermaError, + ErrorRecord, + SchemaError, + ValidationError, +) +from .migration import ( + MigrationDispatchResult, + MigrationFailure, + MigrationLimits, + MigrationResult, + migrate_aggregate, + migrate_and_dispatch, +) from .validator import collect_errors, validate +from .wire import ( + ArtifactResolver, + DefinitionResolver, + MemoryArtifactResolver, + MigrationDescriptorResolver, + RestoredAggregate, + RestoredAggregatePackage, + aggregate_envelope, + aggregate_shape_fingerprint, + restore_aggregate, + restore_aggregate_package, + serialize_aggregate, +) __all__ = [ + "ArtifactError", + "ArtifactResolver", "Bundle", "BundleSource", "CelError", "DetermaError", + "DefinitionResolver", "Delivery", "ErrorRecord", + "MemoryArtifactResolver", + "MigrationDescriptorResolver", + "MigrationDispatchResult", + "MigrationFailure", + "MigrationLimits", + "MigrationResult", "Result", + "RestoredAggregate", + "RestoredAggregatePackage", "SchemaError", "ValidationError", "__version__", + "aggregate_envelope", + "aggregate_shape_fingerprint", "collect_errors", "create", "dispatch", "load_bundle", + "migrate_aggregate", + "migrate_and_dispatch", + "restore_aggregate", + "restore_aggregate_package", + "serialize_aggregate", "validate", ] diff --git a/src/determa/state/data/aggregate-state-package.schema.json b/src/determa/state/data/aggregate-state-package.schema.json new file mode 100644 index 0000000..75d8b6b --- /dev/null +++ b/src/determa/state/data/aggregate-state-package.schema.json @@ -0,0 +1,61 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://determa.dev/state/schema/aggregate-state-package.schema.json", + "title": "Determa State self-contained aggregate-state package", + "description": "Closed transport package for one aggregate and its verified attachments.", + "type": "object", + "required": [ + "aggregate_state_package_format", + "aggregate_state_package_schema_version", + "aggregate_state", + "normalized_definitions", + "migration_descriptors", + "migration_route" + ], + "additionalProperties": false, + "properties": { + "aggregate_state_package_format": { + "const": "determa.aggregate_state_package" + }, + "aggregate_state_package_schema_version": { + "const": 1 + }, + "aggregate_state": { + "$ref": "aggregate-state.schema.json" + }, + "normalized_definitions": { + "type": "array", + "items": { + "type": "object", + "required": [ + "validated_bundle_fingerprint", + "normalized_bundle" + ], + "additionalProperties": false, + "properties": { + "validated_bundle_fingerprint": { + "type": "string", + "pattern": "^sha256:[0-9a-f]{64}$" + }, + "normalized_bundle": { + "$ref": "aggregate-state.schema.json#/$defs/typedValue" + } + } + } + }, + "migration_descriptors": { + "type": "array", + "items": { + "$ref": "migration-descriptor.schema.json" + } + }, + "migration_route": { + "type": "array", + "uniqueItems": true, + "items": { + "type": "string", + "pattern": "^sha256:[0-9a-f]{64}$" + } + } + } +} diff --git a/src/determa/state/data/aggregate-state.schema.json b/src/determa/state/data/aggregate-state.schema.json new file mode 100644 index 0000000..db57842 --- /dev/null +++ b/src/determa/state/data/aggregate-state.schema.json @@ -0,0 +1,769 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://determa.dev/state/schema/aggregate-state.schema.json", + "title": "Determa State portable aggregate-state envelope", + "description": "Closed wire schema for aggregate_state_schema_version 1.", + "$ref": "#/$defs/aggregateState", + "$defs": { + "nonEmptyString": { + "type": "string", + "minLength": 1 + }, + "identifier": { + "type": "string", + "pattern": "^[A-Za-z_][A-Za-z0-9_]*$" + }, + "namespace": { + "type": "string", + "pattern": "^[A-Za-z_][A-Za-z0-9_]*(\\.[A-Za-z_][A-Za-z0-9_]*)*$" + }, + "canonicalDecimal": { + "type": "string", + "pattern": "^(0|[1-9][0-9]*)$" + }, + "positiveCanonicalDecimal": { + "type": "string", + "pattern": "^[1-9][0-9]*$" + }, + "signedCanonicalDecimal": { + "type": "string", + "pattern": "^(0|-?[1-9][0-9]*)$" + }, + "sha256": { + "type": "string", + "pattern": "^sha256:[0-9a-f]{64}$" + }, + "jsonPointer": { + "type": "string", + "pattern": "^(?:/(?:[^~/]|~[01])*)*$" + }, + "sourceLocator": { + "type": "string", + "pattern": "^(?:(?:/(?:[^~/]|~[01])*)+|system:[A-Za-z_][A-Za-z0-9_]*)$" + }, + "typedValue": { + "oneOf": [ + { + "type": "array", + "prefixItems": [ + { + "const": "null" + } + ], + "minItems": 1, + "maxItems": 1 + }, + { + "type": "array", + "prefixItems": [ + { + "const": "boolean" + }, + { + "type": "boolean" + } + ], + "minItems": 2, + "maxItems": 2 + }, + { + "type": "array", + "prefixItems": [ + { + "const": "string" + }, + { + "type": "string" + } + ], + "minItems": 2, + "maxItems": 2 + }, + { + "type": "array", + "prefixItems": [ + { + "const": "integer" + }, + { + "$ref": "#/$defs/signedCanonicalDecimal" + } + ], + "minItems": 2, + "maxItems": 2 + }, + { + "type": "array", + "prefixItems": [ + { + "const": "float" + }, + { + "type": "string", + "pattern": "^[0-9a-f]{16}$" + } + ], + "minItems": 2, + "maxItems": 2 + }, + { + "type": "array", + "prefixItems": [ + { + "const": "list" + }, + { + "type": "array", + "items": { + "$ref": "#/$defs/typedValue" + } + } + ], + "minItems": 2, + "maxItems": 2 + }, + { + "type": "array", + "prefixItems": [ + { + "const": "map" + }, + { + "type": "array", + "items": { + "type": "array", + "prefixItems": [ + { + "type": "string" + }, + { + "$ref": "#/$defs/typedValue" + } + ], + "minItems": 2, + "maxItems": 2 + } + } + ], + "minItems": 2, + "maxItems": 2 + } + ] + }, + "machineIdentity": { + "type": "object", + "required": [ + "namespace", + "machine_id", + "machine_version", + "root_definition_pointer" + ], + "additionalProperties": false, + "properties": { + "namespace": { + "$ref": "#/$defs/namespace" + }, + "machine_id": { + "$ref": "#/$defs/identifier" + }, + "machine_version": { + "$ref": "#/$defs/positiveCanonicalDecimal" + }, + "root_definition_pointer": { + "$ref": "#/$defs/jsonPointer" + } + } + }, + "definitionBinding": { + "type": "object", + "required": [ + "validated_bundle_fingerprint", + "machine" + ], + "additionalProperties": false, + "properties": { + "validated_bundle_fingerprint": { + "$ref": "#/$defs/sha256" + }, + "machine": { + "$ref": "#/$defs/machineIdentity" + } + } + }, + "rootIdentityOrigin": { + "type": "object", + "required": [ + "kind", + "definition", + "root_instance_id" + ], + "additionalProperties": false, + "properties": { + "kind": { + "const": "root" + }, + "definition": { + "$ref": "#/$defs/definitionBinding" + }, + "root_instance_id": { + "$ref": "#/$defs/nonEmptyString" + } + } + }, + "componentIdentityOrigin": { + "type": "object", + "required": [ + "kind", + "definition", + "owner_runtime_id", + "component_definition_pointer", + "activation_sequence", + "declaration_index" + ], + "additionalProperties": false, + "properties": { + "kind": { + "const": "component" + }, + "definition": { + "$ref": "#/$defs/definitionBinding" + }, + "owner_runtime_id": { + "$ref": "#/$defs/sha256" + }, + "component_definition_pointer": { + "$ref": "#/$defs/jsonPointer" + }, + "activation_sequence": { + "$ref": "#/$defs/canonicalDecimal" + }, + "declaration_index": { + "$ref": "#/$defs/canonicalDecimal" + } + } + }, + "spawnedIdentityOrigin": { + "type": "object", + "required": [ + "kind", + "definition", + "owner_runtime_id", + "spawn_action_pointer", + "spawn_sequence" + ], + "additionalProperties": false, + "properties": { + "kind": { + "const": "owned_spawned_instance" + }, + "definition": { + "$ref": "#/$defs/definitionBinding" + }, + "owner_runtime_id": { + "$ref": "#/$defs/sha256" + }, + "spawn_action_pointer": { + "$ref": "#/$defs/jsonPointer" + }, + "spawn_sequence": { + "$ref": "#/$defs/canonicalDecimal" + } + } + }, + "identityOrigin": { + "oneOf": [ + { + "$ref": "#/$defs/rootIdentityOrigin" + }, + { + "$ref": "#/$defs/componentIdentityOrigin" + }, + { + "$ref": "#/$defs/spawnedIdentityOrigin" + } + ] + }, + "instanceReference": { + "type": "object", + "required": [ + "root_instance_id", + "instance_id", + "machine_id", + "machine_version" + ], + "additionalProperties": false, + "properties": { + "root_instance_id": { + "$ref": "#/$defs/nonEmptyString" + }, + "instance_id": { + "$ref": "#/$defs/nonEmptyString" + }, + "machine_id": { + "$ref": "#/$defs/identifier" + }, + "machine_version": { + "$ref": "#/$defs/positiveCanonicalDecimal" + } + } + }, + "targetIdentity": { + "oneOf": [ + { + "type": "object", + "required": [ + "root" + ], + "additionalProperties": false, + "properties": { + "root": { + "type": "object", + "required": [ + "root_instance_id", + "root_runtime_id" + ], + "additionalProperties": false, + "properties": { + "root_instance_id": { + "$ref": "#/$defs/nonEmptyString" + }, + "root_runtime_id": { + "$ref": "#/$defs/nonEmptyString" + } + } + } + } + }, + { + "type": "object", + "required": [ + "component" + ], + "additionalProperties": false, + "properties": { + "component": { + "type": "object", + "required": [ + "root_instance_id", + "owner_runtime_id", + "component_id", + "component_runtime_id", + "activation_sequence" + ], + "additionalProperties": false, + "properties": { + "root_instance_id": { + "$ref": "#/$defs/nonEmptyString" + }, + "owner_runtime_id": { + "$ref": "#/$defs/nonEmptyString" + }, + "component_id": { + "$ref": "#/$defs/identifier" + }, + "component_runtime_id": { + "$ref": "#/$defs/nonEmptyString" + }, + "activation_sequence": { + "$ref": "#/$defs/canonicalDecimal" + } + } + } + } + }, + { + "type": "object", + "required": [ + "spawned_instance" + ], + "additionalProperties": false, + "properties": { + "spawned_instance": { + "$ref": "#/$defs/instanceReference" + } + } + } + ] + }, + "lifetimeHolder": { + "type": "object", + "required": [ + "holder_runtime_id", + "variable_declaration_pointer", + "holder_state_activation_sequence" + ], + "additionalProperties": false, + "properties": { + "holder_runtime_id": { + "$ref": "#/$defs/sha256" + }, + "variable_declaration_pointer": { + "$ref": "#/$defs/jsonPointer" + }, + "holder_state_activation_sequence": { + "$ref": "#/$defs/canonicalDecimal" + } + } + }, + "relation": { + "oneOf": [ + { + "type": "object", + "required": [ + "kind" + ], + "additionalProperties": false, + "properties": { + "kind": { + "const": "root" + } + } + }, + { + "type": "object", + "required": [ + "kind", + "owner_runtime_id", + "component_id", + "current_component_definition_pointer", + "activation_sequence", + "declaration_index" + ], + "additionalProperties": false, + "properties": { + "kind": { + "const": "component" + }, + "owner_runtime_id": { + "$ref": "#/$defs/sha256" + }, + "component_id": { + "$ref": "#/$defs/identifier" + }, + "current_component_definition_pointer": { + "$ref": "#/$defs/jsonPointer" + }, + "activation_sequence": { + "$ref": "#/$defs/canonicalDecimal" + }, + "declaration_index": { + "$ref": "#/$defs/canonicalDecimal" + } + } + }, + { + "type": "object", + "required": [ + "kind", + "owner_runtime_id", + "current_spawn_action_pointer", + "spawn_sequence", + "lifetime_holder" + ], + "additionalProperties": false, + "properties": { + "kind": { + "const": "owned_spawned_instance" + }, + "owner_runtime_id": { + "$ref": "#/$defs/sha256" + }, + "current_spawn_action_pointer": { + "$ref": "#/$defs/jsonPointer" + }, + "spawn_sequence": { + "$ref": "#/$defs/canonicalDecimal" + }, + "lifetime_holder": { + "oneOf": [ + { + "type": "null" + }, + { + "$ref": "#/$defs/lifetimeHolder" + } + ] + } + } + } + ] + }, + "stateActivation": { + "type": "object", + "required": [ + "state_definition_pointer", + "activation_sequence" + ], + "additionalProperties": false, + "properties": { + "state_definition_pointer": { + "$ref": "#/$defs/jsonPointer" + }, + "activation_sequence": { + "$ref": "#/$defs/canonicalDecimal" + } + } + }, + "variable": { + "type": "object", + "required": [ + "variable_declaration_pointer", + "declaring_state_activation_sequence", + "value" + ], + "additionalProperties": false, + "properties": { + "variable_declaration_pointer": { + "$ref": "#/$defs/jsonPointer" + }, + "declaring_state_activation_sequence": { + "$ref": "#/$defs/canonicalDecimal" + }, + "value": { + "$ref": "#/$defs/typedValue" + } + } + }, + "history": { + "type": "object", + "required": [ + "history_declaration_pointer", + "recorded_state_definition_pointers" + ], + "additionalProperties": false, + "properties": { + "history_declaration_pointer": { + "$ref": "#/$defs/jsonPointer" + }, + "recorded_state_definition_pointers": { + "oneOf": [ + { + "type": "null" + }, + { + "type": "array", + "items": { + "$ref": "#/$defs/jsonPointer" + }, + "uniqueItems": true + } + ] + } + } + }, + "nextCounter": { + "type": "object", + "required": [ + "definition_pointer", + "next_sequence" + ], + "additionalProperties": false, + "properties": { + "definition_pointer": { + "$ref": "#/$defs/jsonPointer" + }, + "next_sequence": { + "$ref": "#/$defs/canonicalDecimal" + } + } + }, + "fault": { + "type": "object", + "required": [ + "definition_fingerprint", + "runtime_id", + "cause_id", + "code", + "step_sequence", + "source_locator" + ], + "additionalProperties": false, + "properties": { + "definition_fingerprint": { + "$ref": "#/$defs/sha256" + }, + "runtime_id": { + "$ref": "#/$defs/sha256" + }, + "cause_id": { + "$ref": "#/$defs/nonEmptyString" + }, + "code": { + "$ref": "#/$defs/identifier" + }, + "step_sequence": { + "$ref": "#/$defs/canonicalDecimal" + }, + "source_locator": { + "$ref": "#/$defs/sourceLocator" + } + } + }, + "runtime": { + "type": "object", + "required": [ + "runtime_id", + "identity_origin", + "target_identity", + "current_definition", + "relation", + "status", + "active_leaf_state_definition_pointers", + "active_state_activations", + "variables", + "history", + "next_spawn_sequence", + "next_state_activation_sequences", + "next_component_activation_sequences", + "fault" + ], + "additionalProperties": false, + "properties": { + "runtime_id": { + "$ref": "#/$defs/sha256" + }, + "identity_origin": { + "$ref": "#/$defs/identityOrigin" + }, + "target_identity": { + "$ref": "#/$defs/targetIdentity" + }, + "current_definition": { + "$ref": "#/$defs/definitionBinding" + }, + "relation": { + "$ref": "#/$defs/relation" + }, + "status": { + "enum": [ + "running", + "completed", + "faulted" + ] + }, + "active_leaf_state_definition_pointers": { + "type": "array", + "items": { + "$ref": "#/$defs/jsonPointer" + }, + "uniqueItems": true + }, + "active_state_activations": { + "type": "array", + "items": { + "$ref": "#/$defs/stateActivation" + } + }, + "variables": { + "type": "array", + "items": { + "$ref": "#/$defs/variable" + } + }, + "history": { + "type": "array", + "items": { + "$ref": "#/$defs/history" + } + }, + "next_spawn_sequence": { + "$ref": "#/$defs/canonicalDecimal" + }, + "next_state_activation_sequences": { + "type": "array", + "items": { + "$ref": "#/$defs/nextCounter" + } + }, + "next_component_activation_sequences": { + "type": "array", + "items": { + "$ref": "#/$defs/nextCounter" + } + }, + "fault": { + "oneOf": [ + { + "type": "null" + }, + { + "$ref": "#/$defs/fault" + } + ] + } + } + }, + "aggregateState": { + "type": "object", + "required": [ + "aggregate_state_format", + "aggregate_state_schema_version", + "machine_format", + "validated_bundle_fingerprint", + "namespace", + "root_machine_id", + "root_machine_version", + "root_instance_id", + "creation_id", + "root_runtime_id", + "migration_sequence", + "next_logical_step_sequence", + "next_output_sequence", + "runtimes", + "aggregate_state_digest" + ], + "additionalProperties": false, + "properties": { + "aggregate_state_format": { + "const": "determa.aggregate_state" + }, + "aggregate_state_schema_version": { + "const": 1 + }, + "machine_format": { + "const": 1 + }, + "validated_bundle_fingerprint": { + "$ref": "#/$defs/sha256" + }, + "namespace": { + "$ref": "#/$defs/namespace" + }, + "root_machine_id": { + "$ref": "#/$defs/identifier" + }, + "root_machine_version": { + "$ref": "#/$defs/positiveCanonicalDecimal" + }, + "root_instance_id": { + "$ref": "#/$defs/nonEmptyString" + }, + "creation_id": { + "$ref": "#/$defs/nonEmptyString" + }, + "root_runtime_id": { + "$ref": "#/$defs/sha256" + }, + "migration_sequence": { + "$ref": "#/$defs/canonicalDecimal" + }, + "next_logical_step_sequence": { + "$ref": "#/$defs/canonicalDecimal" + }, + "next_output_sequence": { + "$ref": "#/$defs/canonicalDecimal" + }, + "runtimes": { + "type": "array", + "minItems": 1, + "items": { + "$ref": "#/$defs/runtime" + } + }, + "aggregate_state_digest": { + "$ref": "#/$defs/sha256" + } + } + } + } +} diff --git a/src/determa/state/data/migration-descriptor.schema.json b/src/determa/state/data/migration-descriptor.schema.json new file mode 100644 index 0000000..e186bab --- /dev/null +++ b/src/determa/state/data/migration-descriptor.schema.json @@ -0,0 +1,588 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://determa.dev/state/schema/migration-descriptor.schema.json", + "title": "Determa State aggregate migration descriptor", + "description": "Closed declarative migration schema for migration_descriptor_schema_version 1.", + "$ref": "#/$defs/migrationDescriptor", + "$defs": { + "identifier": { + "type": "string", + "pattern": "^[A-Za-z_][A-Za-z0-9_]*$" + }, + "nonEmptyString": { + "type": "string", + "minLength": 1 + }, + "canonicalDecimal": { + "type": "string", + "pattern": "^(0|[1-9][0-9]*)$" + }, + "sha256": { + "type": "string", + "pattern": "^sha256:[0-9a-f]{64}$" + }, + "jsonPointer": { + "type": "string", + "pattern": "^(?:/(?:[^~/]|~[01])*)*$" + }, + "pointerMapping": { + "type": "object", + "required": [ + "source_definition_pointer", + "target_definition_pointer" + ], + "additionalProperties": false, + "properties": { + "source_definition_pointer": { + "$ref": "#/$defs/jsonPointer" + }, + "target_definition_pointer": { + "$ref": "#/$defs/jsonPointer" + } + } + }, + "activeStateMapping": { + "type": "object", + "required": [ + "source_leaf_state_definition_pointer", + "target_leaf_state_definition_pointers" + ], + "additionalProperties": false, + "properties": { + "source_leaf_state_definition_pointer": { + "$ref": "#/$defs/jsonPointer" + }, + "target_leaf_state_definition_pointers": { + "type": "array", + "minItems": 1, + "uniqueItems": true, + "items": { + "$ref": "#/$defs/jsonPointer" + } + } + } + }, + "variableRule": { + "oneOf": [ + { + "type": "object", + "required": [ + "operation", + "source_declaration_pointer", + "target_declaration_pointer" + ], + "additionalProperties": false, + "properties": { + "operation": { + "const": "copy" + }, + "source_declaration_pointer": { + "$ref": "#/$defs/jsonPointer" + }, + "target_declaration_pointer": { + "$ref": "#/$defs/jsonPointer" + } + } + }, + { + "type": "object", + "required": [ + "operation", + "source_declaration_pointers", + "target_declaration_pointer", + "expression" + ], + "additionalProperties": false, + "properties": { + "operation": { + "const": "transform" + }, + "source_declaration_pointers": { + "type": "array", + "minItems": 1, + "uniqueItems": true, + "items": { + "$ref": "#/$defs/jsonPointer" + } + }, + "target_declaration_pointer": { + "$ref": "#/$defs/jsonPointer" + }, + "expression": { + "type": "string", + "minLength": 1 + } + } + }, + { + "type": "object", + "required": [ + "operation", + "source_declaration_pointer", + "reason" + ], + "additionalProperties": false, + "properties": { + "operation": { + "const": "drop" + }, + "source_declaration_pointer": { + "$ref": "#/$defs/jsonPointer" + }, + "reason": { + "$ref": "#/$defs/nonEmptyString" + } + } + }, + { + "type": "object", + "required": [ + "operation", + "target_declaration_pointer", + "expression" + ], + "additionalProperties": false, + "properties": { + "operation": { + "const": "initialize" + }, + "target_declaration_pointer": { + "$ref": "#/$defs/jsonPointer" + }, + "expression": { + "type": "string", + "minLength": 1 + } + } + } + ] + }, + "historyRule": { + "oneOf": [ + { + "type": "object", + "required": [ + "operation", + "source_history_declaration_pointer", + "target_history_declaration_pointer", + "recorded_state_mappings" + ], + "additionalProperties": false, + "properties": { + "operation": { + "const": "map" + }, + "source_history_declaration_pointer": { + "$ref": "#/$defs/jsonPointer" + }, + "target_history_declaration_pointer": { + "$ref": "#/$defs/jsonPointer" + }, + "recorded_state_mappings": { + "type": "array", + "items": { + "$ref": "#/$defs/pointerMapping" + } + } + } + }, + { + "type": "object", + "required": [ + "operation", + "source_history_declaration_pointer", + "reason" + ], + "additionalProperties": false, + "properties": { + "operation": { + "const": "drop" + }, + "source_history_declaration_pointer": { + "$ref": "#/$defs/jsonPointer" + }, + "reason": { + "$ref": "#/$defs/nonEmptyString" + } + } + }, + { + "type": "object", + "required": [ + "operation", + "target_history_declaration_pointer" + ], + "additionalProperties": false, + "properties": { + "operation": { + "const": "initialize_null" + }, + "target_history_declaration_pointer": { + "$ref": "#/$defs/jsonPointer" + } + } + } + ] + }, + "componentRule": { + "type": "object", + "required": [ + "source_component_definition_pointer", + "target_component_definition_pointer", + "target_component_id" + ], + "additionalProperties": false, + "properties": { + "source_component_definition_pointer": { + "$ref": "#/$defs/jsonPointer" + }, + "target_component_definition_pointer": { + "$ref": "#/$defs/jsonPointer" + }, + "target_component_id": { + "$ref": "#/$defs/identifier" + } + } + }, + "ownedRuntimeRule": { + "type": "object", + "required": [ + "source_spawn_action_pointer", + "target_spawn_action_pointer", + "source_machine_id", + "target_machine_id" + ], + "additionalProperties": false, + "properties": { + "source_spawn_action_pointer": { + "$ref": "#/$defs/jsonPointer" + }, + "target_spawn_action_pointer": { + "$ref": "#/$defs/jsonPointer" + }, + "source_machine_id": { + "$ref": "#/$defs/identifier" + }, + "target_machine_id": { + "$ref": "#/$defs/identifier" + } + } + }, + "lifetimeHolderRule": { + "type": "object", + "required": [ + "source_variable_declaration_pointer", + "target_variable_declaration_pointer" + ], + "additionalProperties": false, + "properties": { + "source_variable_declaration_pointer": { + "$ref": "#/$defs/jsonPointer" + }, + "target_variable_declaration_pointer": { + "$ref": "#/$defs/jsonPointer" + } + } + }, + "counterRule": { + "oneOf": [ + { + "type": "object", + "required": [ + "operation", + "source_definition_pointer", + "target_definition_pointer" + ], + "additionalProperties": false, + "properties": { + "operation": { + "const": "map" + }, + "source_definition_pointer": { + "$ref": "#/$defs/jsonPointer" + }, + "target_definition_pointer": { + "$ref": "#/$defs/jsonPointer" + } + } + }, + { + "type": "object", + "required": [ + "operation", + "target_definition_pointer" + ], + "additionalProperties": false, + "properties": { + "operation": { + "const": "initialize_zero" + }, + "target_definition_pointer": { + "$ref": "#/$defs/jsonPointer" + } + } + }, + { + "type": "object", + "required": [ + "operation", + "source_definition_pointers", + "target_definition_pointer" + ], + "additionalProperties": false, + "properties": { + "operation": { + "const": "merge_maximum" + }, + "source_definition_pointers": { + "type": "array", + "minItems": 1, + "uniqueItems": true, + "items": { + "$ref": "#/$defs/jsonPointer" + } + }, + "target_definition_pointer": { + "$ref": "#/$defs/jsonPointer" + } + } + } + ] + }, + "terminalPolicy": { + "type": "object", + "required": [ + "completed", + "faulted" + ], + "additionalProperties": false, + "properties": { + "completed": { + "enum": [ + "preserve", + "reject" + ] + }, + "faulted": { + "enum": [ + "preserve", + "reject" + ] + } + } + }, + "resourceRequirements": { + "type": "object", + "required": [ + "maximum_transformed_output_bytes", + "maximum_cel_expression_length", + "maximum_cel_ast_nodes", + "maximum_cel_evaluation_steps" + ], + "additionalProperties": false, + "properties": { + "maximum_transformed_output_bytes": { + "$ref": "#/$defs/canonicalDecimal" + }, + "maximum_cel_expression_length": { + "$ref": "#/$defs/canonicalDecimal" + }, + "maximum_cel_ast_nodes": { + "$ref": "#/$defs/canonicalDecimal" + }, + "maximum_cel_evaluation_steps": { + "$ref": "#/$defs/canonicalDecimal" + } + } + }, + "mappings": { + "type": "object", + "required": [ + "machines", + "active_states", + "variables", + "history", + "components", + "owned_runtimes", + "lifetime_holders", + "counters" + ], + "additionalProperties": false, + "properties": { + "machines": { + "type": "array", + "items": { + "$ref": "#/$defs/pointerMapping" + } + }, + "active_states": { + "type": "array", + "items": { + "$ref": "#/$defs/activeStateMapping" + } + }, + "variables": { + "type": "array", + "items": { + "$ref": "#/$defs/variableRule" + } + }, + "history": { + "type": "array", + "items": { + "$ref": "#/$defs/historyRule" + } + }, + "components": { + "type": "array", + "items": { + "$ref": "#/$defs/componentRule" + } + }, + "owned_runtimes": { + "type": "array", + "items": { + "$ref": "#/$defs/ownedRuntimeRule" + } + }, + "lifetime_holders": { + "type": "array", + "items": { + "$ref": "#/$defs/lifetimeHolderRule" + } + }, + "counters": { + "type": "array", + "items": { + "$ref": "#/$defs/counterRule" + } + } + } + }, + "migrationDescriptor": { + "type": "object", + "required": [ + "migration_descriptor_format", + "migration_descriptor_schema_version", + "source_machine_format", + "target_machine_format", + "source_validated_bundle_fingerprint", + "target_validated_bundle_fingerprint", + "source_aggregate_shape_fingerprint", + "target_aggregate_shape_fingerprint", + "mode", + "mappings", + "terminal_policy", + "resource_requirements", + "migration_descriptor_digest" + ], + "additionalProperties": false, + "properties": { + "migration_descriptor_format": { + "const": "determa.aggregate_migration" + }, + "migration_descriptor_schema_version": { + "const": 1 + }, + "source_machine_format": { + "const": 1 + }, + "target_machine_format": { + "const": 1 + }, + "source_validated_bundle_fingerprint": { + "$ref": "#/$defs/sha256" + }, + "target_validated_bundle_fingerprint": { + "$ref": "#/$defs/sha256" + }, + "source_aggregate_shape_fingerprint": { + "$ref": "#/$defs/sha256" + }, + "target_aggregate_shape_fingerprint": { + "$ref": "#/$defs/sha256" + }, + "mode": { + "enum": [ + "compatible", + "transform" + ] + }, + "mappings": { + "$ref": "#/$defs/mappings" + }, + "terminal_policy": { + "$ref": "#/$defs/terminalPolicy" + }, + "resource_requirements": { + "$ref": "#/$defs/resourceRequirements" + }, + "migration_descriptor_digest": { + "$ref": "#/$defs/sha256" + } + }, + "allOf": [ + { + "if": { + "properties": { + "mode": { + "const": "compatible" + } + }, + "required": [ + "mode" + ] + }, + "then": { + "properties": { + "mappings": { + "properties": { + "machines": { + "maxItems": 0 + }, + "active_states": { + "maxItems": 0 + }, + "variables": { + "maxItems": 0 + }, + "history": { + "maxItems": 0 + }, + "components": { + "maxItems": 0 + }, + "owned_runtimes": { + "maxItems": 0 + }, + "lifetime_holders": { + "maxItems": 0 + }, + "counters": { + "maxItems": 0 + } + } + }, + "resource_requirements": { + "properties": { + "maximum_transformed_output_bytes": { + "const": "0" + }, + "maximum_cel_expression_length": { + "const": "0" + }, + "maximum_cel_ast_nodes": { + "const": "0" + }, + "maximum_cel_evaluation_steps": { + "const": "0" + } + } + } + } + } + } + ] + } + } +} diff --git a/src/determa/state/engine.py b/src/determa/state/engine.py index a84557b..9026dda 100644 --- a/src/determa/state/engine.py +++ b/src/determa/state/engine.py @@ -486,6 +486,7 @@ def _is_prior_counter_path(path: tuple[str | int, ...]) -> bool: ("fault", "step_sequence"), ("holder", "state_activation_sequence"), ("target", "component", "activation_sequence"), + ("_target_identity", "component", "activation_sequence"), }: return True return len(suffix) == 2 and suffix[0] in { @@ -583,17 +584,33 @@ def _validate_prior_state(state: dict[str, Any], bundle: Bundle) -> bool: return False if root.get("owner_runtime_id") is not None or root.get("status") != state["status"]: return False - expected_root_id = _identity( - [ - "determa-root-runtime-identity-2", - "1", - state["validated_bundle_fingerprint"], - state["namespace"], - root.get("machine_id"), - str(root.get("machine_version")), - state["root_instance_id"], - ] - ) + origin = root.get("_identity_origin") + if isinstance(origin, dict) and origin.get("kind") == "root": + definition = origin.get("definition", {}) + machine = definition.get("machine", {}) + expected_root_id = _identity( + [ + "determa-root-runtime-identity-2", + "1", + definition.get("validated_bundle_fingerprint"), + machine.get("namespace"), + machine.get("machine_id"), + str(machine.get("machine_version")), + origin.get("root_instance_id"), + ] + ) + else: + expected_root_id = _identity( + [ + "determa-root-runtime-identity-2", + "1", + state["validated_bundle_fingerprint"], + state["namespace"], + root.get("machine_id"), + str(root.get("machine_version")), + state["root_instance_id"], + ] + ) if root.get("runtime_id") != expected_root_id or root["runtime_id"] != state["root_runtime_id"]: return False if root.get("machine_id") != state["root_machine_id"]: @@ -830,55 +847,111 @@ def _valid_component_identity(state: dict[str, Any], runtime: dict[str, Any]) -> or not _logical_counter(runtime["owning_state_activation_sequence"]) ): return False - expected_id = _identity( - [ - "determa-component-runtime-identity-1", - "1", - state["root_instance_id"], - runtime["owner_runtime_id"], - runtime["component_definition_pointer"], - str(runtime["component_activation_sequence"]), - state["namespace"], - runtime["machine_id"], - str(runtime["machine_version"]), - ] - ) - expected_target = { - "component": { - "root_instance_id": state["root_instance_id"], - "owner_runtime_id": runtime["owner_runtime_id"], - "component_id": runtime["component_id"], - "component_runtime_id": runtime["runtime_id"], - "activation_sequence": runtime["component_activation_sequence"], + origin = runtime.get("_identity_origin") + if isinstance(origin, dict) and origin.get("kind") == "component": + definition = origin.get("definition", {}) + machine = definition.get("machine", {}) + expected_id = _identity( + [ + "determa-component-runtime-identity-1", + "1", + state["root_instance_id"], + origin.get("owner_runtime_id"), + origin.get("component_definition_pointer"), + str(origin.get("activation_sequence")), + machine.get("namespace"), + machine.get("machine_id"), + str(machine.get("machine_version")), + ] + ) + expected_target = runtime.get("_target_identity") + else: + expected_id = _identity( + [ + "determa-component-runtime-identity-1", + "1", + state["root_instance_id"], + runtime["owner_runtime_id"], + runtime["component_definition_pointer"], + str(runtime["component_activation_sequence"]), + state["namespace"], + runtime["machine_id"], + str(runtime["machine_version"]), + ] + ) + expected_target = { + "component": { + "root_instance_id": state["root_instance_id"], + "owner_runtime_id": runtime["owner_runtime_id"], + "component_id": runtime["component_id"], + "component_runtime_id": runtime["runtime_id"], + "activation_sequence": runtime["component_activation_sequence"], + } } - } return bool(runtime["runtime_id"] == expected_id and runtime["target"] == expected_target) def _valid_spawned_identity(state: dict[str, Any], runtime: dict[str, Any]) -> bool: + origin = runtime.get("_identity_origin") + target = runtime.get("_target_identity") + migrated = ( + isinstance(origin, dict) + and origin.get("kind") == "owned_spawned_instance" + and isinstance(target, dict) + and target.get("spawned_instance") == runtime.get("instance_reference") + ) if ( not _logical_counter(runtime.get("spawn_sequence")) or not isinstance(runtime.get("spawn_action_pointer"), str) or not _is_instance_reference(runtime.get("instance_reference")) or runtime["instance_reference"].get("root_instance_id") != state["root_instance_id"] or runtime["instance_reference"].get("instance_id") != runtime["runtime_id"] - or runtime["instance_reference"].get("machine_id") != runtime["machine_id"] - or runtime["instance_reference"].get("machine_version") != runtime["machine_version"] + or ( + not migrated + and runtime["instance_reference"].get("machine_id") != runtime["machine_id"] + ) + or ( + not migrated + and runtime["instance_reference"].get("machine_version") + != runtime["machine_version"] + ) ): return False - expected_id = _identity( - [ - "determa-spawned-runtime-identity-1", - "1", - state["root_instance_id"], - runtime["owner_runtime_id"], - runtime["spawn_action_pointer"], - str(runtime["spawn_sequence"]), - state["namespace"], - runtime["machine_id"], - str(runtime["machine_version"]), - ] - ) + if migrated: + assert isinstance(origin, dict) + definition = origin.get("definition") + if not isinstance(definition, dict): + return False + machine = definition.get("machine") + if not isinstance(machine, dict): + return False + expected_id = _identity( + [ + "determa-spawned-runtime-identity-1", + "1", + state["root_instance_id"], + origin.get("owner_runtime_id"), + origin.get("spawn_action_pointer"), + str(origin.get("spawn_sequence")), + machine.get("namespace"), + machine.get("machine_id"), + str(machine.get("machine_version")), + ] + ) + else: + expected_id = _identity( + [ + "determa-spawned-runtime-identity-1", + "1", + state["root_instance_id"], + runtime["owner_runtime_id"], + runtime["spawn_action_pointer"], + str(runtime["spawn_sequence"]), + state["namespace"], + runtime["machine_id"], + str(runtime["machine_version"]), + ] + ) holder = runtime.get("holder") if holder is not None and ( not isinstance(holder, dict) diff --git a/src/determa/state/errors.py b/src/determa/state/errors.py index 63b7490..a0d0ebd 100644 --- a/src/determa/state/errors.py +++ b/src/determa/state/errors.py @@ -37,6 +37,16 @@ class CelError(DetermaError): """A portable CEL expression failed to compile or evaluate.""" +class ArtifactError(DetermaError): + """A portable persistence artifact is invalid or unsupported.""" + + def __init__(self, code: str, path: str = "", message: str = "") -> None: + self.code = code + self.path = path + self.message = message or code + super().__init__(self.message) + + class StepFault(DetermaError): """Internal control flow for one atomic RTC fault.""" diff --git a/src/determa/state/migration.py b/src/determa/state/migration.py new file mode 100644 index 0000000..ef1fca9 --- /dev/null +++ b/src/determa/state/migration.py @@ -0,0 +1,1049 @@ +"""Deterministic portable aggregate migration and atomic dispatch composition.""" + +from __future__ import annotations + +import copy +from collections.abc import Sequence +from dataclasses import dataclass +from typing import Any, cast + +from . import cel +from .definition import Bundle, _escape_pointer +from .engine import Delivery, dispatch +from .errors import ArtifactError, CelError +from .model import BundleModel, MachineModel, StateNode +from .wire import ( + ArtifactResolver, + ArtifactSource, + _bundle_from_resolver, + _source_bytes, + aggregate_shape_fingerprint, + aggregate_state_digest, + canonical_bytes, + decimal, + decoded_typed_value, + load_json_artifact, + migration_descriptor_digest, + restore_aggregate, + typed_value, +) + + +@dataclass(frozen=True) +class MigrationLimits: + """Configured per-operation resource limits, at least the portable floors.""" + + maximum_aggregate_bytes: int = 1_048_576 + maximum_definition_bytes: int = 1_048_576 + maximum_descriptor_bytes: int = 65_536 + maximum_transformed_output_bytes: int = 65_536 + maximum_json_nesting_depth: int = 64 + maximum_runtimes: int = 256 + maximum_active_states_per_runtime: int = 1_024 + maximum_variables_per_runtime: int = 4_096 + maximum_map_members: int = 4_096 + maximum_list_members: int = 4_096 + maximum_string_utf8_bytes: int = 65_536 + maximum_chain_length: int = 8 + maximum_descriptor_rules: int = 1_024 + maximum_cel_expression_length: int = 65_536 + maximum_cel_ast_nodes: int = 65_536 + maximum_cel_evaluation_steps: int = 1_000_000 + + @classmethod + def from_mapping(cls, value: dict[str, Any]) -> MigrationLimits: + expected = set(cls.__dataclass_fields__) + if set(value) != expected: + raise ArtifactError("invalid_migration_request") + try: + parsed = {name: decimal(value[name]) for name in expected} + except ArtifactError as exc: + raise ArtifactError("invalid_migration_request") from exc + return cls(**parsed) + + +@dataclass(frozen=True) +class MigrationFailure: + code: str + + +@dataclass(frozen=True) +class MigrationResult: + """Pure migration success or one closed deterministic failure.""" + + aggregate_envelope: dict[str, Any] | None + aggregate_bytes: bytes | None + audit_records: tuple[dict[str, Any], ...] + failure: MigrationFailure | None + + @property + def succeeded(self) -> bool: + return self.failure is None + + +@dataclass(frozen=True) +class MigrationDispatchResult: + """One atomic migration plus optional ordinary dispatch result.""" + + aggregate_envelope: dict[str, Any] | None + aggregate_bytes: bytes | None + audit_records: tuple[dict[str, Any], ...] + status: str | None + disposition: str | None + emissions: tuple[dict[str, Any], ...] + fault: dict[str, Any] | None + rejection: dict[str, Any] | None + failure: MigrationFailure | None + + @property + def succeeded(self) -> bool: + return self.failure is None + + +def _failure(code: str) -> MigrationResult: + return MigrationResult(None, None, (), MigrationFailure(code)) + + +def _dispatch_failure(code: str) -> MigrationDispatchResult: + return MigrationDispatchResult( + None, None, (), None, None, (), None, None, MigrationFailure(code) + ) + + +def _resource_metrics(value: Any, depth: int = 0) -> tuple[int, int, int, int]: + maximum_depth = depth + maximum_map_members = 0 + maximum_list_members = 0 + maximum_string_bytes = 0 + if isinstance(value, dict): + maximum_depth = depth + 1 + maximum_map_members = len(value) + for key, child in value.items(): + maximum_string_bytes = max(maximum_string_bytes, len(key.encode("utf-8"))) + child_metrics = _resource_metrics(child, depth + 1) + maximum_depth = max(maximum_depth, child_metrics[0]) + maximum_map_members = max(maximum_map_members, child_metrics[1]) + maximum_list_members = max(maximum_list_members, child_metrics[2]) + maximum_string_bytes = max(maximum_string_bytes, child_metrics[3]) + elif isinstance(value, list): + maximum_depth = depth + 1 + maximum_list_members = len(value) + for child in value: + child_metrics = _resource_metrics(child, depth + 1) + maximum_depth = max(maximum_depth, child_metrics[0]) + maximum_map_members = max(maximum_map_members, child_metrics[1]) + maximum_list_members = max(maximum_list_members, child_metrics[2]) + maximum_string_bytes = max(maximum_string_bytes, child_metrics[3]) + elif isinstance(value, str): + maximum_string_bytes = len(value.encode("utf-8")) + return ( + maximum_depth, + maximum_map_members, + maximum_list_members, + maximum_string_bytes, + ) + + +def _check_shape_limits( + aggregate: dict[str, Any], + definitions: list[Bundle], + descriptors: list[dict[str, Any]], + limits: MigrationLimits, +) -> None: + if len(canonical_bytes(aggregate)) > limits.maximum_aggregate_bytes: + raise ArtifactError("migration_resource_limit_exceeded") + if any( + len(canonical_bytes(typed_value(bundle.raw))) > limits.maximum_definition_bytes + for bundle in definitions + ): + raise ArtifactError("migration_resource_limit_exceeded") + if any( + len(canonical_bytes(descriptor)) > limits.maximum_descriptor_bytes + for descriptor in descriptors + ): + raise ArtifactError("migration_resource_limit_exceeded") + values: list[Any] = [aggregate, *[bundle.raw for bundle in definitions], *descriptors] + metrics = [_resource_metrics(value) for value in values] + if ( + max(item[0] for item in metrics) > limits.maximum_json_nesting_depth + or max(item[1] for item in metrics) > limits.maximum_map_members + or max(item[2] for item in metrics) > limits.maximum_list_members + or max(item[3] for item in metrics) > limits.maximum_string_utf8_bytes + or len(aggregate["runtimes"]) > limits.maximum_runtimes + or any( + len(runtime["active_state_activations"]) + > limits.maximum_active_states_per_runtime + for runtime in aggregate["runtimes"] + ) + or any( + len(runtime["variables"]) > limits.maximum_variables_per_runtime + for runtime in aggregate["runtimes"] + ) + ): + raise ArtifactError("migration_resource_limit_exceeded") + + +def _ast_nodes(value: Any) -> int: + children = getattr(value, "children", None) + if children is None: + return 1 + return 1 + sum(_ast_nodes(child) for child in children) + + +def _descriptor_static_requirements( + descriptor: dict[str, Any], limits: MigrationLimits +) -> tuple[int, int]: + rule_count = sum(len(items) for items in descriptor["mappings"].values()) + if rule_count > limits.maximum_descriptor_rules: + raise ArtifactError("migration_resource_limit_exceeded") + expressions = { + rule["expression"] + for rule in descriptor["mappings"]["variables"] + if "expression" in rule + } + expression_bytes = sum(len(expression.encode("utf-8")) for expression in expressions) + ast_nodes = sum(_ast_nodes(cel._tree(expression)) for expression in expressions) + requirements = descriptor["resource_requirements"] + if ( + expression_bytes > limits.maximum_cel_expression_length + or ast_nodes > limits.maximum_cel_ast_nodes + or expression_bytes > decimal(requirements["maximum_cel_expression_length"]) + or ast_nodes > decimal(requirements["maximum_cel_ast_nodes"]) + ): + raise ArtifactError("migration_resource_limit_exceeded") + return expression_bytes, ast_nodes + + +def _pointer_parts(pointer: str) -> list[str]: + if not pointer.startswith("/"): + raise ArtifactError("invalid_migration_descriptor") + return [ + item.replace("~1", "/").replace("~0", "~") + for item in pointer[1:].split("/") + ] + + +def _pointer_get(document: Any, pointer: str) -> Any: + current = document + for part in _pointer_parts(pointer): + if isinstance(current, list): + try: + current = current[int(part)] + except (IndexError, ValueError) as exc: + raise ArtifactError("invalid_migration_descriptor") from exc + elif isinstance(current, dict) and part in current: + current = current[part] + else: + raise ArtifactError("invalid_migration_descriptor") + return current + + +def _machine_identity_for_pointer(bundle: Bundle, root_pointer: str) -> dict[str, Any]: + parts = _pointer_parts(root_pointer) + if len(parts) < 3 or parts[0] != "machines": + raise ArtifactError("migration_totality_failure") + try: + machine = bundle.raw["machines"][int(parts[1])] + except (IndexError, ValueError) as exc: + raise ArtifactError("migration_totality_failure") from exc + return { + "namespace": bundle.namespace, + "machine_id": machine["machine_id"], + "machine_version": str(machine["version"]), + "root_definition_pointer": root_pointer, + } + + +def _definition_binding(bundle: Bundle, root_pointer: str) -> dict[str, Any]: + return { + "validated_bundle_fingerprint": bundle.fingerprint, + "machine": _machine_identity_for_pointer(bundle, root_pointer), + } + + +def _machine_model_for_root(bundle: Bundle, root_pointer: str) -> MachineModel: + models = BundleModel(bundle) + identity = _machine_identity_for_pointer(bundle, root_pointer) + base = models.machine(identity["machine_id"]) + if root_pointer == base.root_pointer: + return base + root = _pointer_get(bundle.raw, root_pointer) + return MachineModel( + bundle, + base.raw, + machine_index=base.machine_index, + root=root, + root_pointer=root_pointer, + identity_machine=base.identity_machine, + ) + + +def _state_nodes(machine: MachineModel) -> dict[str, StateNode]: + return {node.pointer: node for node in machine.states.values()} + + +def _variable_declaration(bundle: Bundle, pointer: str) -> dict[str, Any]: + declaration = _pointer_get(bundle.raw, pointer) + if not isinstance(declaration, dict) or "type" not in declaration: + raise ArtifactError("invalid_migration_descriptor") + return declaration + + +def _state_pointer_for_variable(pointer: str) -> str: + marker = "/variables/" + if marker not in pointer: + raise ArtifactError("invalid_migration_descriptor") + return pointer.split(marker, 1)[0] + + +def _history_pointers(machine: MachineModel) -> list[str]: + return sorted( + ( + f"{node.pointer}/history" + for node in machine.states.values() + if node.type == "composite" and node.raw.get("history", "none") != "none" + ), + key=lambda item: item.encode("utf-8"), + ) + + +def _active_ancestor_pointers(machine: MachineModel, leaves: list[str]) -> list[str]: + result: set[str] = set() + nodes = _state_nodes(machine) + for pointer in leaves: + node = nodes.get(pointer) + if node is None: + raise ArtifactError("migration_totality_failure") + result.update(item.pointer for item in node.ancestors(include_self=True)) + return sorted(result, key=lambda item: item.encode("utf-8")) + + +def _unique_mapping( + rules: list[dict[str, Any]], source_member: str, target_member: str +) -> dict[str, str]: + result: dict[str, str] = {} + targets: set[str] = set() + for rule in rules: + source = rule[source_member] + target = rule[target_member] + if source in result or target in targets: + raise ArtifactError("invalid_migration_descriptor") + result[source] = target + targets.add(target) + return result + + +def _validate_descriptor_semantics( + descriptor: dict[str, Any], + source_bundle: Bundle, + target_bundle: Bundle, + limits: MigrationLimits, +) -> None: + if ( + descriptor["source_validated_bundle_fingerprint"] != source_bundle.fingerprint + or descriptor["target_validated_bundle_fingerprint"] != target_bundle.fingerprint + or descriptor["source_aggregate_shape_fingerprint"] + != aggregate_shape_fingerprint(source_bundle) + or descriptor["target_aggregate_shape_fingerprint"] + != aggregate_shape_fingerprint(target_bundle) + ): + raise ArtifactError("invalid_migration_descriptor") + mappings = descriptor["mappings"] + if descriptor["mode"] == "compatible": + if ( + descriptor["source_aggregate_shape_fingerprint"] + != descriptor["target_aggregate_shape_fingerprint"] + or any(mappings.values()) + ): + raise ArtifactError("invalid_migration_descriptor") + _unique_mapping( + mappings["machines"], "source_definition_pointer", "target_definition_pointer" + ) + _unique_mapping( + mappings["components"], + "source_component_definition_pointer", + "target_component_definition_pointer", + ) + _unique_mapping( + mappings["owned_runtimes"], + "source_spawn_action_pointer", + "target_spawn_action_pointer", + ) + _unique_mapping( + mappings["lifetime_holders"], + "source_variable_declaration_pointer", + "target_variable_declaration_pointer", + ) + active_sources: set[str] = set() + active_targets: set[str] = set() + for rule in mappings["active_states"]: + source = rule["source_leaf_state_definition_pointer"] + targets = rule["target_leaf_state_definition_pointers"] + if source in active_sources or any(target in active_targets for target in targets): + raise ArtifactError("invalid_migration_descriptor") + active_sources.add(source) + active_targets.update(targets) + consumed: set[str] = set() + produced: set[str] = set() + for rule in mappings["variables"]: + sources = ( + rule.get("source_declaration_pointers") + or ( + [rule["source_declaration_pointer"]] + if "source_declaration_pointer" in rule + else [] + ) + ) + target = rule.get("target_declaration_pointer") + if any(source in consumed for source in sources) or ( + target is not None and target in produced + ): + raise ArtifactError("invalid_migration_descriptor") + consumed.update(sources) + if target is not None: + produced.add(target) + _variable_declaration(target_bundle, target) + for source in sources: + _variable_declaration(source_bundle, source) + if "expression" in rule: + scope = { + f"source_{index}": cel.type_from_declaration( + _variable_declaration(source_bundle, source) + ) + for index, source in enumerate(sources) + } + target_declaration = _variable_declaration(target_bundle, cast(str, target)) + if target_declaration["type"] == "instance_reference" or any( + declaration.kind == "instance_reference" for declaration in scope.values() + ): + raise ArtifactError("invalid_migration_descriptor") + try: + cel.check_expression( + rule["expression"], + scope, + expected=cel.type_from_declaration(target_declaration), + event_fields=None, + owner_fields=None, + ) + except CelError as exc: + raise ArtifactError("invalid_migration_descriptor") from exc + _descriptor_static_requirements(descriptor, limits) + + +def _resolve_descriptor( + resolver: ArtifactResolver, digest: str +) -> tuple[dict[str, Any], bytes]: + source = resolver.resolve_migration_descriptor(digest) + if source is None: + raise ArtifactError("migration_route_mismatch") + if not resolver.migration_descriptor_is_trusted(digest): + raise ArtifactError("migration_descriptor_untrusted") + document, _raw = load_json_artifact(source, "migration_descriptor") + encoded = canonical_bytes(document) + if migration_descriptor_digest(document) != digest: + raise ArtifactError("invalid_migration_descriptor") + return document, encoded + + +def _compatible_candidate( + source: dict[str, Any], + target_bundle: Bundle, +) -> dict[str, Any]: + candidate = copy.deepcopy(source) + candidate["validated_bundle_fingerprint"] = target_bundle.fingerprint + candidate["namespace"] = target_bundle.namespace + for runtime in candidate["runtimes"]: + runtime["current_definition"]["validated_bundle_fingerprint"] = ( + target_bundle.fingerprint + ) + runtime["current_definition"]["machine"]["namespace"] = target_bundle.namespace + candidate["migration_sequence"] = str(decimal(candidate["migration_sequence"]) + 1) + candidate["aggregate_state_digest"] = aggregate_state_digest(candidate) + return candidate + + +def _counter_transform( + items: list[dict[str, Any]], rules: list[dict[str, Any]] +) -> list[dict[str, Any]]: + source = {item["definition_pointer"]: decimal(item["next_sequence"]) for item in items} + consumed: set[str] = set() + result: list[dict[str, Any]] = [] + targets: set[str] = set() + for rule in rules: + operation = rule["operation"] + target = rule["target_definition_pointer"] + if target in targets: + raise ArtifactError("invalid_migration_descriptor") + targets.add(target) + if operation == "map": + pointer = rule["source_definition_pointer"] + if pointer not in source or pointer in consumed: + continue + value = source[pointer] + consumed.add(pointer) + elif operation == "initialize_zero": + value = 0 + else: + pointers = rule["source_definition_pointers"] + if any(pointer not in source or pointer in consumed for pointer in pointers): + continue + value = max(source[pointer] for pointer in pointers) + consumed.update(pointers) + result.append({"definition_pointer": target, "next_sequence": str(value)}) + if consumed != set(source): + raise ArtifactError("migration_totality_failure") + result.sort(key=lambda item: item["definition_pointer"].encode("utf-8")) + return result + + +def _mapped_active( + runtime: dict[str, Any], + source_machine: MachineModel, + target_machine: MachineModel, + descriptor: dict[str, Any], +) -> tuple[list[str], list[dict[str, Any]]]: + active_rules = descriptor["mappings"]["active_states"] + targets: list[str] = [] + for leaf in runtime["active_leaf_state_definition_pointers"]: + matching = [ + rule + for rule in active_rules + if rule["source_leaf_state_definition_pointer"] == leaf + ] + if len(matching) != 1: + raise ArtifactError( + "invalid_migration_descriptor" + if len(matching) > 1 + else "migration_totality_failure" + ) + targets.extend(matching[0]["target_leaf_state_definition_pointers"]) + if len(set(targets)) != len(targets): + raise ArtifactError("migration_totality_failure") + targets.sort(key=lambda item: item.encode("utf-8")) + source_activations = { + item["state_definition_pointer"]: item["activation_sequence"] + for item in runtime["active_state_activations"] + } + counter_rules = descriptor["mappings"]["counters"] + pointer_map: dict[str, str] = {} + for rule in counter_rules: + if rule["operation"] == "map": + pointer_map[rule["source_definition_pointer"]] = rule[ + "target_definition_pointer" + ] + target_ancestors = _active_ancestor_pointers(target_machine, targets) + activations: list[dict[str, Any]] = [] + for target in target_ancestors: + sources = [ + source + for source, mapped in pointer_map.items() + if mapped == target and source in source_activations + ] + if len(sources) != 1: + raise ArtifactError("migration_totality_failure") + activations.append( + { + "state_definition_pointer": target, + "activation_sequence": source_activations[sources[0]], + } + ) + return targets, activations + + +def _target_variable_pointers( + target_machine: MachineModel, active_activations: list[dict[str, Any]] +) -> dict[str, str]: + states = _state_nodes(target_machine) + result: dict[str, str] = {} + for activation in active_activations: + state = states[activation["state_definition_pointer"]] + for name in (state.raw.get("variables") or {}): + pointer = f"{state.pointer}/variables/{_escape_pointer(name)}" + result[pointer] = activation["activation_sequence"] + return result + + +def _transform_variables( + runtime: dict[str, Any], + source_bundle: Bundle, + target_bundle: Bundle, + target_machine: MachineModel, + target_activations: list[dict[str, Any]], + descriptor: dict[str, Any], +) -> tuple[list[dict[str, Any]], int, int]: + source_values: dict[str, dict[str, Any]] = {} + for occurrence in runtime["variables"]: + pointer = occurrence["variable_declaration_pointer"] + if pointer in source_values: + raise ArtifactError("migration_totality_failure") + source_values[pointer] = occurrence + target_required = _target_variable_pointers(target_machine, target_activations) + produced: dict[str, dict[str, Any]] = {} + consumed: set[str] = set() + transformed_bytes = 0 + evaluation_steps = 0 + for rule in descriptor["mappings"]["variables"]: + operation = rule["operation"] + target = rule.get("target_declaration_pointer") + sources = ( + rule.get("source_declaration_pointers") + or ( + [rule["source_declaration_pointer"]] + if "source_declaration_pointer" in rule + else [] + ) + ) + applicable_sources = [source_values.get(pointer) for pointer in sources] + if operation == "drop": + if applicable_sources[0] is not None: + consumed.add(sources[0]) + continue + if target not in target_required: + continue + if target in produced: + raise ArtifactError("invalid_migration_descriptor") + if operation in {"copy", "transform"} and any( + value is None for value in applicable_sources + ): + raise ArtifactError("migration_totality_failure") + if operation == "copy": + value = copy.deepcopy(cast(dict[str, Any], applicable_sources[0])["value"]) + else: + activation: dict[str, Any] = {} + if operation == "transform": + for index, occurrence in enumerate(applicable_sources): + activation[f"source_{index}"] = decoded_typed_value( + cast(dict[str, Any], occurrence)["value"] + ) + consumed.update(sources) + expression = rule["expression"] + nodes = _ast_nodes(cel._tree(expression)) + evaluation_steps += nodes + try: + evaluated = cel.evaluate(expression, activation) + except CelError as exc: + raise ArtifactError("migration_transform_fault") from exc + value = typed_value(evaluated) + transformed_bytes += len(canonical_bytes(value)) + if operation == "copy": + consumed.update(sources) + produced[target] = { + "variable_declaration_pointer": target, + "declaring_state_activation_sequence": target_required[target], + "value": value, + } + if consumed != set(source_values) or set(produced) != set(target_required): + raise ArtifactError("migration_totality_failure") + result = sorted( + produced.values(), + key=lambda item: ( + item["variable_declaration_pointer"].encode("utf-8"), + int(item["declaring_state_activation_sequence"]), + ), + ) + return result, transformed_bytes, evaluation_steps + + +def _transform_history( + runtime: dict[str, Any], + target_machine: MachineModel, + descriptor: dict[str, Any], +) -> list[dict[str, Any]]: + source = { + item["history_declaration_pointer"]: item for item in runtime["history"] + } + consumed: set[str] = set() + produced: dict[str, dict[str, Any]] = {} + for rule in descriptor["mappings"]["history"]: + operation = rule["operation"] + if operation == "initialize_null": + target = rule["target_history_declaration_pointer"] + produced[target] = { + "history_declaration_pointer": target, + "recorded_state_definition_pointers": None, + } + continue + source_pointer = rule["source_history_declaration_pointer"] + occurrence = source.get(source_pointer) + if occurrence is None: + continue + consumed.add(source_pointer) + if operation == "drop": + continue + target = rule["target_history_declaration_pointer"] + recorded = occurrence["recorded_state_definition_pointers"] + mapping = { + item["source_definition_pointer"]: item["target_definition_pointer"] + for item in rule["recorded_state_mappings"] + } + if recorded is not None: + if any(pointer not in mapping for pointer in recorded): + raise ArtifactError("migration_totality_failure") + recorded = sorted( + (mapping[pointer] for pointer in recorded), + key=lambda item: item.encode("utf-8"), + ) + produced[target] = { + "history_declaration_pointer": target, + "recorded_state_definition_pointers": recorded, + } + if consumed != set(source) or set(produced) != set(_history_pointers(target_machine)): + raise ArtifactError("migration_totality_failure") + return sorted( + produced.values(), + key=lambda item: item["history_declaration_pointer"].encode("utf-8"), + ) + + +def _component_counter_transform( + items: list[dict[str, Any]], component_rules: list[dict[str, Any]] +) -> list[dict[str, Any]]: + mapping = { + rule["source_component_definition_pointer"]: rule[ + "target_component_definition_pointer" + ] + for rule in component_rules + } + result = [] + for item in items: + source = item["definition_pointer"] + if source not in mapping: + raise ArtifactError("migration_totality_failure") + result.append( + { + "definition_pointer": mapping[source], + "next_sequence": item["next_sequence"], + } + ) + result.sort(key=lambda item: item["definition_pointer"].encode("utf-8")) + return result + + +def _target_root_for_component(bundle: Bundle, pointer: str) -> str: + placement = _pointer_get(bundle.raw, pointer) + if not isinstance(placement, dict): + raise ArtifactError("migration_totality_failure") + if "root" in placement: + return f"{pointer}/root" + machine_id = placement.get("machine_id") + for index, machine in enumerate(bundle.raw["machines"]): + if machine["machine_id"] == machine_id: + return f"/machines/{index}/root" + raise ArtifactError("migration_totality_failure") + + +def _target_root_for_machine(bundle: Bundle, machine_id: str) -> str: + for index, machine in enumerate(bundle.raw["machines"]): + if machine["machine_id"] == machine_id: + return f"/machines/{index}/root" + raise ArtifactError("migration_totality_failure") + + +def _transform_candidate( + source: dict[str, Any], + source_bundle: Bundle, + target_bundle: Bundle, + descriptor: dict[str, Any], + limits: MigrationLimits, +) -> dict[str, Any]: + candidate = copy.deepcopy(source) + mappings = descriptor["mappings"] + machine_mapping = _unique_mapping( + mappings["machines"], "source_definition_pointer", "target_definition_pointer" + ) + component_mapping = { + rule["source_component_definition_pointer"]: rule + for rule in mappings["components"] + } + owned_mapping = { + (rule["source_spawn_action_pointer"], rule["source_machine_id"]): rule + for rule in mappings["owned_runtimes"] + } + holder_mapping = _unique_mapping( + mappings["lifetime_holders"], + "source_variable_declaration_pointer", + "target_variable_declaration_pointer", + ) + total_transformed_bytes = 0 + total_evaluation_steps = 0 + for runtime in candidate["runtimes"]: + relation = runtime["relation"] + source_root = runtime["current_definition"]["machine"]["root_definition_pointer"] + if relation["kind"] == "root": + target_root = machine_mapping.get(source_root) + if target_root is None: + raise ArtifactError("migration_totality_failure") + elif relation["kind"] == "component": + source_component = relation["current_component_definition_pointer"] + rule = component_mapping.get(source_component) + if rule is None: + raise ArtifactError("migration_totality_failure") + target_component = rule["target_component_definition_pointer"] + relation["current_component_definition_pointer"] = target_component + relation["component_id"] = rule["target_component_id"] + parts = _pointer_parts(target_component) + relation["declaration_index"] = str(int(parts[-1])) + target_root = _target_root_for_component(target_bundle, target_component) + else: + source_spawn = relation["current_spawn_action_pointer"] + source_machine = runtime["current_definition"]["machine"]["machine_id"] + rule = owned_mapping.get((source_spawn, source_machine)) + if rule is None: + raise ArtifactError("migration_totality_failure") + relation["current_spawn_action_pointer"] = rule["target_spawn_action_pointer"] + target_root = _target_root_for_machine(target_bundle, rule["target_machine_id"]) + holder = relation["lifetime_holder"] + if holder is not None: + source_holder = holder["variable_declaration_pointer"] + if source_holder not in holder_mapping: + raise ArtifactError("migration_totality_failure") + holder["variable_declaration_pointer"] = holder_mapping[source_holder] + source_machine_model = _machine_model_for_root(source_bundle, source_root) + target_machine_model = _machine_model_for_root(target_bundle, target_root) + leaves, activations = _mapped_active( + runtime, source_machine_model, target_machine_model, descriptor + ) + variables, transformed_bytes, evaluation_steps = _transform_variables( + runtime, + source_bundle, + target_bundle, + target_machine_model, + activations, + descriptor, + ) + runtime["active_leaf_state_definition_pointers"] = leaves + runtime["active_state_activations"] = activations + runtime["variables"] = variables + runtime["history"] = _transform_history( + runtime, target_machine_model, descriptor + ) + runtime["next_state_activation_sequences"] = _counter_transform( + runtime["next_state_activation_sequences"], mappings["counters"] + ) + runtime["next_component_activation_sequences"] = _component_counter_transform( + runtime["next_component_activation_sequences"], mappings["components"] + ) + runtime["current_definition"] = _definition_binding(target_bundle, target_root) + total_transformed_bytes += transformed_bytes + total_evaluation_steps += evaluation_steps + requirements = descriptor["resource_requirements"] + if ( + total_transformed_bytes + > min( + limits.maximum_transformed_output_bytes, + decimal(requirements["maximum_transformed_output_bytes"]), + ) + or total_evaluation_steps + > min( + limits.maximum_cel_evaluation_steps, + decimal(requirements["maximum_cel_evaluation_steps"]), + ) + ): + raise ArtifactError("migration_resource_limit_exceeded") + root = next( + runtime + for runtime in candidate["runtimes"] + if runtime["runtime_id"] == candidate["root_runtime_id"] + ) + candidate["validated_bundle_fingerprint"] = target_bundle.fingerprint + candidate["namespace"] = target_bundle.namespace + candidate["root_machine_id"] = root["current_definition"]["machine"]["machine_id"] + candidate["root_machine_version"] = root["current_definition"]["machine"][ + "machine_version" + ] + candidate["migration_sequence"] = str(decimal(candidate["migration_sequence"]) + 1) + candidate["aggregate_state_digest"] = aggregate_state_digest(candidate) + return candidate + + +def _apply_descriptor( + source: dict[str, Any], + source_bundle: Bundle, + target_bundle: Bundle, + descriptor: dict[str, Any], + artifact_resolver: ArtifactResolver, + limits: MigrationLimits, + maintenance_mode: bool, +) -> dict[str, Any]: + root = next( + runtime + for runtime in source["runtimes"] + if runtime["runtime_id"] == source["root_runtime_id"] + ) + if root["status"] in {"completed", "faulted"}: + if not maintenance_mode: + raise ArtifactError("terminal_migration_requires_maintenance") + if descriptor["terminal_policy"][root["status"]] != "preserve": + raise ArtifactError("terminal_migration_rejected") + _validate_descriptor_semantics(descriptor, source_bundle, target_bundle, limits) + candidate = ( + _compatible_candidate(source, target_bundle) + if descriptor["mode"] == "compatible" + else _transform_candidate( + source, source_bundle, target_bundle, descriptor, limits + ) + ) + restore_aggregate(canonical_bytes(candidate), artifact_resolver) + return candidate + + +def migrate_aggregate( + aggregate: ArtifactSource, + target_validated_bundle_fingerprint: str, + migration_route: Sequence[str], + artifact_resolver: ArtifactResolver, + *, + maintenance_mode: bool, + resource_limits: MigrationLimits | None = None, +) -> MigrationResult: + """Apply one exact trusted descriptor route to an aggregate copy.""" + source_copy = _source_bytes(aggregate) + limits = resource_limits or MigrationLimits() + try: + if ( + not isinstance(target_validated_bundle_fingerprint, str) + or not target_validated_bundle_fingerprint + or not isinstance(migration_route, Sequence) + or isinstance(migration_route, str | bytes) + or not all(isinstance(item, str) and item for item in migration_route) + or not isinstance(maintenance_mode, bool) + ): + raise ArtifactError("invalid_migration_request") + route = list(migration_route) + if len(route) > limits.maximum_chain_length: + raise ArtifactError("migration_resource_limit_exceeded") + restored = restore_aggregate(source_copy, artifact_resolver) + if not route: + if ( + restored.aggregate_envelope["validated_bundle_fingerprint"] + != target_validated_bundle_fingerprint + ): + raise ArtifactError("migration_route_missing") + return MigrationResult( + copy.deepcopy(restored.aggregate_envelope), source_copy, (), None + ) + if len(set(route)) != len(route): + raise ArtifactError("migration_route_mismatch") + descriptors_with_bytes = [ + _resolve_descriptor(artifact_resolver, digest) for digest in route + ] + descriptors = [item[0] for item in descriptors_with_bytes] + if ( + descriptors[0]["source_validated_bundle_fingerprint"] + != restored.bundle.fingerprint + or descriptors[-1]["target_validated_bundle_fingerprint"] + != target_validated_bundle_fingerprint + or any( + left["target_validated_bundle_fingerprint"] + != right["source_validated_bundle_fingerprint"] + for left, right in zip(descriptors, descriptors[1:], strict=False) + ) + ): + raise ArtifactError("migration_route_mismatch") + fingerprints = [descriptors[0]["source_validated_bundle_fingerprint"]] + [ + descriptor["target_validated_bundle_fingerprint"] + for descriptor in descriptors + ] + if len(set(fingerprints)) != len(fingerprints): + raise ArtifactError("migration_route_mismatch") + definitions = [ + _bundle_from_resolver( + artifact_resolver, + fingerprint, + source=index == 0, + require_trust=True, + ) + for index, fingerprint in enumerate(fingerprints) + ] + _check_shape_limits( + restored.aggregate_envelope, definitions, descriptors, limits + ) + candidate = copy.deepcopy(restored.aggregate_envelope) + audits: list[dict[str, Any]] = [] + for index, descriptor in enumerate(descriptors): + before_digest = candidate["aggregate_state_digest"] + candidate = _apply_descriptor( + candidate, + definitions[index], + definitions[index + 1], + descriptor, + artifact_resolver, + limits, + maintenance_mode, + ) + audits.append( + { + "migration_audit_record_schema_version": 1, + "root_instance_id": candidate["root_instance_id"], + "root_runtime_id": candidate["root_runtime_id"], + "migration_sequence": candidate["migration_sequence"], + "source_validated_bundle_fingerprint": descriptor[ + "source_validated_bundle_fingerprint" + ], + "target_validated_bundle_fingerprint": descriptor[ + "target_validated_bundle_fingerprint" + ], + "migration_descriptor_digest": descriptor[ + "migration_descriptor_digest" + ], + "source_aggregate_state_digest": before_digest, + "target_aggregate_state_digest": candidate[ + "aggregate_state_digest" + ], + "result_code": "migration_applied", + } + ) + encoded = canonical_bytes(candidate) + return MigrationResult(candidate, encoded, tuple(audits), None) + except ArtifactError as exc: + return _failure(exc.code) + except (CelError, KeyError, TypeError, ValueError) as exc: + del exc + return _failure("invalid_migration_descriptor") + + +def migrate_and_dispatch( + aggregate: ArtifactSource, + target_validated_bundle_fingerprint: str, + migration_route: Sequence[str], + artifact_resolver: ArtifactResolver, + delivery: Delivery, + *, + maintenance_mode: bool, + resource_limits: MigrationLimits | None = None, +) -> MigrationDispatchResult: + """Migrate and dispatch as one commit-ready pure result boundary.""" + migrated = migrate_aggregate( + aggregate, + target_validated_bundle_fingerprint, + migration_route, + artifact_resolver, + maintenance_mode=maintenance_mode, + resource_limits=resource_limits, + ) + if migrated.failure is not None: + return _dispatch_failure(migrated.failure.code) + assert migrated.aggregate_bytes is not None + try: + restored = restore_aggregate(migrated.aggregate_bytes, artifact_resolver) + core = dispatch(restored.bundle, restored.state, delivery) + state = core["state"] + if state is None: + raise ArtifactError("invalid_aggregate_state") + from .wire import aggregate_envelope + + envelope = aggregate_envelope(restored.bundle, state) + encoded = canonical_bytes(envelope) + return MigrationDispatchResult( + envelope, + encoded, + migrated.audit_records, + core["status"], + core["disposition"], + tuple(copy.deepcopy(core["emissions"])), + copy.deepcopy(core["fault"]), + copy.deepcopy(core["rejection"]), + None, + ) + except ArtifactError as exc: + return _dispatch_failure(exc.code) diff --git a/src/determa/state/wire.py b/src/determa/state/wire.py new file mode 100644 index 0000000..63509d6 --- /dev/null +++ b/src/determa/state/wire.py @@ -0,0 +1,1402 @@ +"""Portable aggregate artifacts, canonical values, and definition resolution.""" + +from __future__ import annotations + +import copy +import hashlib +import json +import math +import struct +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass +from functools import cache, lru_cache +from pathlib import Path +from typing import Any, Protocol, cast + +import rfc8785 + +from .definition import Bundle, BundleSource, _escape_pointer, load_bundle +from .errors import ArtifactError, ValidationError +from .model import BundleModel, MachineModel, StateNode +from .yaml12 import validate_portable_values, validate_unicode + +ArtifactSource = bytes | str | Mapping[str, Any] +SHA256_PREFIX = "sha256:" +_DATA = Path(__file__).parent / "data" +_INT_MIN = -(2**63) +_INT_MAX = 2**63 - 1 + + +class DefinitionResolver(Protocol): + """Resolve one trusted normalized definition by its content fingerprint.""" + + def resolve_definition(self, fingerprint: str) -> Bundle | BundleSource | None: ... + + def definition_is_trusted(self, fingerprint: str) -> bool: ... + + +class MigrationDescriptorResolver(Protocol): + """Resolve one trusted migration descriptor by digest.""" + + def resolve_migration_descriptor(self, digest: str) -> ArtifactSource | None: ... + + def migration_descriptor_is_trusted(self, digest: str) -> bool: ... + + +class ArtifactResolver(DefinitionResolver, MigrationDescriptorResolver, Protocol): + """Resolve definitions and migration descriptors.""" + + +@dataclass(frozen=True) +class RestoredAggregate: + """One verified aggregate and its resolved current definition.""" + + bundle: Bundle + state: dict[str, Any] + aggregate_envelope: dict[str, Any] + canonical_bytes: bytes + source_bytes: bytes + + +@dataclass(frozen=True) +class RestoredAggregatePackage: + """A verified package after idempotently seeding its resolver.""" + + aggregate: RestoredAggregate + migration_route: tuple[str, ...] + package_document: dict[str, Any] + + +class MemoryArtifactResolver: + """Small deterministic resolver suitable for local caches and tests.""" + + def __init__( + self, + *, + definitions: Mapping[str, Bundle | BundleSource] | None = None, + migration_descriptors: Mapping[str, ArtifactSource] | None = None, + trusted_definitions: Sequence[str] | None = None, + trusted_migration_descriptors: Sequence[str] | None = None, + ) -> None: + self._definitions = dict(definitions or {}) + self._migration_descriptors = dict(migration_descriptors or {}) + self._trusted_definitions = set( + self._definitions if trusted_definitions is None else trusted_definitions + ) + self._trusted_migration_descriptors = set( + self._migration_descriptors + if trusted_migration_descriptors is None + else trusted_migration_descriptors + ) + + def resolve_definition(self, fingerprint: str) -> Bundle | BundleSource | None: + return self._definitions.get(fingerprint) + + def definition_is_trusted(self, fingerprint: str) -> bool: + return fingerprint in self._trusted_definitions + + def resolve_migration_descriptor(self, digest: str) -> ArtifactSource | None: + return self._migration_descriptors.get(digest) + + def migration_descriptor_is_trusted(self, digest: str) -> bool: + return digest in self._trusted_migration_descriptors + + def put_definition( + self, fingerprint: str, definition: Bundle | BundleSource, *, trusted: bool = True + ) -> None: + bundle = definition if isinstance(definition, Bundle) else load_bundle(definition) + if bundle.fingerprint != fingerprint: + raise ArtifactError("definition_fingerprint_mismatch") + existing = self._definitions.get(fingerprint) + if existing is not None: + current = existing if isinstance(existing, Bundle) else load_bundle(existing) + if canonical_bytes(typed_value(current.raw)) != canonical_bytes( + typed_value(bundle.raw) + ): + raise ArtifactError("definition_fingerprint_mismatch") + else: + self._definitions[fingerprint] = bundle + if trusted: + self._trusted_definitions.add(fingerprint) + + def put_migration_descriptor( + self, digest: str, descriptor: ArtifactSource, *, trusted: bool = True + ) -> None: + document, _ = load_json_artifact(descriptor, "migration_descriptor") + if migration_descriptor_digest(document) != digest: + raise ArtifactError("invalid_migration_descriptor") + existing = self._migration_descriptors.get(digest) + if existing is not None: + current, _ = load_json_artifact(existing, "migration_descriptor") + if canonical_bytes(current) != canonical_bytes(document): + raise ArtifactError("invalid_migration_descriptor") + else: + self._migration_descriptors[digest] = copy.deepcopy(document) + if trusted: + self._trusted_migration_descriptors.add(digest) + + def snapshot(self) -> dict[str, list[str]]: + return { + "definitions": sorted(self._definitions), + "migration_descriptors": sorted(self._migration_descriptors), + } + + +def _reject_constant(value: str) -> None: + raise ArtifactError("invalid_json_value", message=f"invalid JSON constant {value}") + + +def _object_without_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + raise ArtifactError("duplicate_json_name") + result[key] = value + return result + + +def _source_bytes(source: ArtifactSource) -> bytes: + if isinstance(source, bytes): + return source + if isinstance(source, str): + return source.encode("utf-8", errors="strict") + try: + validate_portable_values(source) + except ValidationError as exc: + raise ArtifactError("invalid_json_value") from exc + if not validate_unicode(source): + raise ArtifactError("invalid_unicode") + return canonical_bytes(copy.deepcopy(dict(source))) + + +def strict_json(source: ArtifactSource) -> tuple[Any, bytes]: + """Parse strict UTF-8 JSON while rejecting duplicate names and invalid values.""" + raw = _source_bytes(source) + try: + text = raw.decode("utf-8", errors="strict") + except UnicodeDecodeError as exc: + raise ArtifactError("invalid_unicode") from exc + try: + value = json.loads( + text, + object_pairs_hook=_object_without_duplicates, + parse_constant=_reject_constant, + ) + except ArtifactError: + raise + except (UnicodeError, json.JSONDecodeError, ValueError) as exc: + raise ArtifactError("invalid_json_value") from exc + try: + validate_portable_values(value) + except ValidationError as exc: + raise ArtifactError("invalid_json_value") from exc + if not validate_unicode(value): + raise ArtifactError("invalid_unicode") + return value, raw + + +def canonical_bytes(value: Any) -> bytes: + """Return exact RFC 8785 bytes for a portable JSON value.""" + try: + return bytes(rfc8785.dumps(value)) + except (rfc8785.CanonicalizationError, UnicodeError, ValueError, TypeError) as exc: + raise ArtifactError("invalid_json_value") from exc + + +def hash_value(value: Any) -> str: + return f"{SHA256_PREFIX}{hashlib.sha256(canonical_bytes(value)).hexdigest()}" + + +def typed_value(value: Any) -> list[Any]: + """Project one Determa value into the exact tagged portable representation.""" + if value is None: + return ["null"] + if isinstance(value, bool): + return ["boolean", value] + if isinstance(value, str): + if not validate_unicode(value): + raise ArtifactError("invalid_aggregate_state") + return ["string", value] + if isinstance(value, int): + if not _INT_MIN <= value <= _INT_MAX: + raise ArtifactError("invalid_aggregate_state") + return ["integer", str(value)] + if isinstance(value, float): + if not math.isfinite(value): + raise ArtifactError("invalid_aggregate_state") + return ["float", struct.pack("!d", 0.0 if value == 0.0 else value).hex()] + if isinstance(value, list): + return ["list", [typed_value(item) for item in value]] + if isinstance(value, dict): + if not all(isinstance(key, str) for key in value): + raise ArtifactError("invalid_aggregate_state") + entries = [ + [key, typed_value(value[key])] + for key in sorted(value, key=lambda item: item.encode("utf-8")) + ] + return ["map", entries] + raise ArtifactError("invalid_aggregate_state") + + +def decoded_typed_value(value: Any) -> Any: + """Decode one exact typed value, rejecting noncanonical representations.""" + if not isinstance(value, list) or not value or not isinstance(value[0], str): + raise ArtifactError("invalid_aggregate_state") + tag = value[0] + if tag == "null" and value == ["null"]: + return None + if len(value) != 2: + raise ArtifactError("invalid_aggregate_state") + payload = value[1] + if tag == "boolean" and isinstance(payload, bool): + return payload + if tag == "string" and isinstance(payload, str) and validate_unicode(payload): + return payload + if tag == "integer" and isinstance(payload, str): + integer = _signed_decimal(payload) + if _INT_MIN <= integer <= _INT_MAX: + return integer + if tag == "float" and isinstance(payload, str) and len(payload) == 16: + try: + number = struct.unpack("!d", bytes.fromhex(payload))[0] + except (ValueError, struct.error): + pass + else: + if payload == payload.lower() and math.isfinite(number): + return 0.0 if number == 0.0 else number + if tag == "list" and isinstance(payload, list): + return [decoded_typed_value(item) for item in payload] + if tag == "map" and isinstance(payload, list): + result: dict[str, Any] = {} + previous: bytes | None = None + for entry in payload: + if ( + not isinstance(entry, list) + or len(entry) != 2 + or not isinstance(entry[0], str) + or not validate_unicode(entry[0]) + ): + raise ArtifactError("invalid_aggregate_state") + encoded = entry[0].encode("utf-8") + if previous is not None and encoded <= previous: + raise ArtifactError("invalid_aggregate_state") + previous = encoded + result[entry[0]] = decoded_typed_value(entry[1]) + return result + raise ArtifactError("invalid_aggregate_state") + + +def _signed_decimal(value: str) -> int: + if value == "0": + return 0 + negative = value.startswith("-") + digits = value[1:] if negative else value + if not digits or not digits.isascii() or not digits.isdigit() or digits.startswith("0"): + raise ArtifactError("invalid_aggregate_state") + return int(value) + + +def decimal(value: Any, *, positive: bool = False) -> int: + if not isinstance(value, str): + raise ArtifactError("invalid_aggregate_state") + number = _signed_decimal(value) + if number < 0 or (positive and number == 0): + raise ArtifactError("invalid_aggregate_state") + return number + + +@cache +def artifact_schema(kind: str) -> dict[str, Any]: + filename = { + "aggregate_state": "aggregate-state.schema.json", + "migration_descriptor": "migration-descriptor.schema.json", + "aggregate_state_package": "aggregate-state-package.schema.json", + }[kind] + return cast( + dict[str, Any], json.loads((_DATA / filename).read_text(encoding="utf-8")) + ) + + +@lru_cache(maxsize=1) +def _schema_registry() -> Any: + from referencing import Registry, Resource + + registry = Registry() + for kind in ( + "aggregate_state", + "migration_descriptor", + "aggregate_state_package", + ): + document = artifact_schema(kind) + registry = registry.with_resource( + document["$id"], Resource.from_contents(document) + ) + return registry + + +def _format_code(document: Any, kind: str) -> str | None: + if not isinstance(document, dict): + return None + definitions = { + "aggregate_state": ( + "aggregate_state_format", + "determa.aggregate_state", + "aggregate_state_schema_version", + 1, + "unsupported_aggregate_state_format", + "unsupported_aggregate_state_schema_version", + ), + "migration_descriptor": ( + "migration_descriptor_format", + "determa.aggregate_migration", + "migration_descriptor_schema_version", + 1, + "unsupported_migration_descriptor_format", + "unsupported_migration_descriptor_schema_version", + ), + "aggregate_state_package": ( + "aggregate_state_package_format", + "determa.aggregate_state_package", + "aggregate_state_package_schema_version", + 1, + "unsupported_aggregate_state_package_format", + "unsupported_aggregate_state_package_schema_version", + ), + } + format_member, expected_format, version_member, expected_version, format_code, version_code = ( + definitions[kind] + ) + if format_member in document and document[format_member] != expected_format: + return format_code + if version_member in document and document[version_member] != expected_version: + return version_code + return None + + +def load_json_artifact( + source: ArtifactSource, kind: str +) -> tuple[dict[str, Any], bytes]: + """Parse and structurally validate one recognized persistence artifact.""" + try: + document, raw = strict_json(source) + except ArtifactError as exc: + code = { + "aggregate_state": "invalid_aggregate_state", + "migration_descriptor": "invalid_migration_descriptor", + "aggregate_state_package": "invalid_aggregate_state_package", + }[kind] + raise ArtifactError(code) from exc + unsupported = _format_code(document, kind) + if unsupported is not None: + raise ArtifactError(unsupported) + import jsonschema + + validator = jsonschema.Draft202012Validator( + artifact_schema(kind), registry=_schema_registry() + ) + if not isinstance(document, dict) or next(validator.iter_errors(document), None) is not None: + code = { + "aggregate_state": "invalid_aggregate_state", + "migration_descriptor": "invalid_migration_descriptor", + "aggregate_state_package": "invalid_aggregate_state_package", + }[kind] + raise ArtifactError(code) + return document, raw + + +def aggregate_state_digest(document: Mapping[str, Any]) -> str: + body = dict(document) + body.pop("aggregate_state_digest", None) + return hash_value(["determa-aggregate-state-digest-1", body]) + + +def migration_descriptor_digest(document: Mapping[str, Any]) -> str: + body = dict(document) + body.pop("migration_descriptor_digest", None) + return hash_value(["determa-migration-descriptor-1", body]) + + +def normalized_definition_attachment(bundle: Bundle) -> dict[str, Any]: + return { + "validated_bundle_fingerprint": bundle.fingerprint, + "normalized_bundle": typed_value(bundle.raw), + } + + +def bundle_from_attachment(attachment: Mapping[str, Any]) -> Bundle: + try: + raw = decoded_typed_value(attachment["normalized_bundle"]) + fingerprint = attachment["validated_bundle_fingerprint"] + except (ArtifactError, KeyError, TypeError) as exc: + raise ArtifactError("invalid_aggregate_state_package") from exc + if not isinstance(raw, dict) or not isinstance(fingerprint, str): + raise ArtifactError("invalid_aggregate_state_package") + bundle = load_bundle(raw) + if bundle.fingerprint != fingerprint: + raise ArtifactError("invalid_aggregate_state_package") + return bundle + + +def _bundle_from_resolver( + resolver: DefinitionResolver, + fingerprint: str, + *, + source: bool, + require_trust: bool = True, +) -> Bundle: + definition = resolver.resolve_definition(fingerprint) + if definition is None: + raise ArtifactError( + "source_definition_unavailable" if source else "target_definition_unavailable" + ) + if require_trust and not resolver.definition_is_trusted(fingerprint): + raise ArtifactError("definition_untrusted") + try: + bundle = definition if isinstance(definition, Bundle) else load_bundle(definition) + except ValidationError as exc: + raise ArtifactError("definition_fingerprint_mismatch") from exc + if bundle.fingerprint != fingerprint: + raise ArtifactError("definition_fingerprint_mismatch") + return bundle + + +def _definition_binding(bundle: Bundle, runtime: Mapping[str, Any]) -> dict[str, Any]: + return { + "validated_bundle_fingerprint": bundle.fingerprint, + "machine": { + "namespace": bundle.namespace, + "machine_id": runtime["machine_id"], + "machine_version": str(runtime["machine_version"]), + "root_definition_pointer": runtime["root_pointer"], + }, + } + + +def _wire_target(target: Mapping[str, Any]) -> dict[str, Any]: + result = copy.deepcopy(dict(target)) + if "component" in result: + result["component"]["activation_sequence"] = str( + result["component"]["activation_sequence"] + ) + elif "spawned_instance" in result: + result["spawned_instance"]["machine_version"] = str( + result["spawned_instance"]["machine_version"] + ) + return result + + +def _runtime_identity_origin( + bundle: Bundle, runtime: Mapping[str, Any] +) -> dict[str, Any]: + stored = runtime.get("_identity_origin") + if isinstance(stored, dict): + return copy.deepcopy(stored) + definition = _definition_binding(bundle, runtime) + role = runtime["role"] + if role == "root": + return { + "kind": "root", + "definition": definition, + "root_instance_id": runtime.get("_root_instance_id"), + } + if role == "component": + return { + "kind": "component", + "definition": definition, + "owner_runtime_id": runtime["owner_runtime_id"], + "component_definition_pointer": runtime["component_definition_pointer"], + "activation_sequence": str(runtime["component_activation_sequence"]), + "declaration_index": str(runtime["component_declaration_index"]), + } + return { + "kind": "owned_spawned_instance", + "definition": definition, + "owner_runtime_id": runtime["owner_runtime_id"], + "spawn_action_pointer": runtime["spawn_action_pointer"], + "spawn_sequence": str(runtime["spawn_sequence"]), + } + + +def _runtime_relation(runtime: Mapping[str, Any]) -> dict[str, Any]: + stored = runtime.get("_relation") + if isinstance(stored, dict): + return copy.deepcopy(stored) + if runtime["role"] == "root": + return {"kind": "root"} + if runtime["role"] == "component": + return { + "kind": "component", + "owner_runtime_id": runtime["owner_runtime_id"], + "component_id": runtime["component_id"], + "current_component_definition_pointer": runtime[ + "component_definition_pointer" + ], + "activation_sequence": str(runtime["component_activation_sequence"]), + "declaration_index": str(runtime["component_declaration_index"]), + } + holder = runtime.get("holder") + lifetime_holder = ( + None + if holder is None + else { + "holder_runtime_id": runtime["owner_runtime_id"], + "variable_declaration_pointer": holder["pointer"], + "holder_state_activation_sequence": str( + holder["state_activation_sequence"] + ), + } + ) + return { + "kind": "owned_spawned_instance", + "owner_runtime_id": runtime["owner_runtime_id"], + "current_spawn_action_pointer": runtime["spawn_action_pointer"], + "spawn_sequence": str(runtime["spawn_sequence"]), + "lifetime_holder": lifetime_holder, + } + + +def _node_for_runtime(machine: MachineModel, path: str) -> StateNode: + try: + return machine.states[path] + except KeyError as exc: + raise ArtifactError("invalid_aggregate_state") from exc + + +def _runtime_wire( + bundle: Bundle, + models: BundleModel, + aggregate: Mapping[str, Any], + runtime: Mapping[str, Any], +) -> dict[str, Any]: + from .engine import _runtime_model + + machine = _runtime_model(bundle, models, cast(dict[str, Any], runtime)) + active_nodes = [_node_for_runtime(machine, path) for path in runtime["active"]] + leaves = ( + [] + if not active_nodes + else [active_nodes[-1].pointer] + ) + activations = sorted( + ( + { + "state_definition_pointer": _node_for_runtime(machine, path).pointer, + "activation_sequence": str(sequence), + } + for path, sequence in runtime["state_activation_sequence"].items() + ), + key=lambda item: ( + item["state_definition_pointer"].encode("utf-8"), + int(item["activation_sequence"]), + ), + ) + variables: list[dict[str, Any]] = [] + for path, values in runtime["scopes"].items(): + node = _node_for_runtime(machine, path) + activation = runtime["state_activation_sequence"][path] + for name, value in values.items(): + variables.append( + { + "variable_declaration_pointer": ( + f"{node.pointer}/variables/{_escape_pointer(name)}" + ), + "declaring_state_activation_sequence": str(activation), + "value": typed_value(value), + } + ) + variables.sort( + key=lambda item: ( + item["variable_declaration_pointer"].encode("utf-8"), + int(item["declaring_state_activation_sequence"]), + ) + ) + history: list[dict[str, Any]] = [] + for path, recorded in runtime["history"].items(): + node = machine.root if path == "$root" else _node_for_runtime(machine, path) + pointers = ( + None + if recorded is None + else sorted( + (_node_for_runtime(machine, item).pointer for item in recorded), + key=lambda item: item.encode("utf-8"), + ) + ) + history.append( + { + "history_declaration_pointer": f"{node.pointer}/history", + "recorded_state_definition_pointers": pointers, + } + ) + history.sort(key=lambda item: item["history_declaration_pointer"].encode("utf-8")) + state_counters = sorted( + ( + { + "definition_pointer": _node_for_runtime(machine, path).pointer, + "next_sequence": str(sequence), + } + for path, sequence in runtime["next_state_activation_sequence"].items() + ), + key=lambda item: item["definition_pointer"].encode("utf-8"), + ) + component_counters = sorted( + ( + {"definition_pointer": pointer, "next_sequence": str(sequence)} + for pointer, sequence in runtime["next_component_activation_sequence"].items() + ), + key=lambda item: item["definition_pointer"].encode("utf-8"), + ) + target = runtime.get("_target_identity") + if not isinstance(target, dict): + if runtime["role"] == "root": + target = { + "root": { + "root_instance_id": aggregate["root_instance_id"], + "root_runtime_id": runtime["runtime_id"], + } + } + elif runtime["role"] == "component": + target = runtime["target"] + else: + target = {"spawned_instance": runtime["instance_reference"]} + current = runtime.get("_current_definition") + if not isinstance(current, dict): + current = _definition_binding(bundle, runtime) + fault = runtime["fault"] + if fault is not None: + fault = copy.deepcopy(fault) + fault["step_sequence"] = str(fault["step_sequence"]) + fault["definition_fingerprint"] = runtime.get( + "_fault_definition_fingerprint", current["validated_bundle_fingerprint"] + ) + origin = _runtime_identity_origin(bundle, runtime) + if origin["kind"] == "root" and origin.get("root_instance_id") is None: + origin["root_instance_id"] = aggregate["root_instance_id"] + return { + "runtime_id": runtime["runtime_id"], + "identity_origin": origin, + "target_identity": _wire_target(target), + "current_definition": copy.deepcopy(current), + "relation": _runtime_relation(runtime), + "status": runtime["status"], + "active_leaf_state_definition_pointers": leaves, + "active_state_activations": activations, + "variables": variables, + "history": history, + "next_spawn_sequence": str(runtime["next_spawn_sequence"]), + "next_state_activation_sequences": state_counters, + "next_component_activation_sequences": component_counters, + "fault": fault, + } + + +def aggregate_envelope(bundle: Bundle | BundleSource, state: dict[str, Any]) -> dict[str, Any]: + """Project one validated engine aggregate into the neutral wire envelope.""" + validated = bundle if isinstance(bundle, Bundle) else load_bundle(bundle) + from .engine import _valid_prior_state + + if not _valid_prior_state(state, validated): + raise ArtifactError("invalid_aggregate_state") + models = BundleModel(validated) + root = state["runtimes"][state["root_runtime_id"]] + restored_order = state.get("_wire_runtime_order") + order_rank = ( + {runtime_id: index for index, runtime_id in enumerate(restored_order)} + if isinstance(restored_order, list) + and set(restored_order) == set(state["runtimes"]) + else None + ) + document: dict[str, Any] = { + "aggregate_state_format": "determa.aggregate_state", + "aggregate_state_schema_version": 1, + "machine_format": 1, + "validated_bundle_fingerprint": validated.fingerprint, + "namespace": validated.namespace, + "root_machine_id": root["machine_id"], + "root_machine_version": str(root["machine_version"]), + "root_instance_id": state["root_instance_id"], + "creation_id": state["creation_id"], + "root_runtime_id": state["root_runtime_id"], + "migration_sequence": str(state.get("migration_sequence", 0)), + "next_logical_step_sequence": str(state["next_logical_step_sequence"]), + "next_output_sequence": str(state["next_output_sequence"]), + "runtimes": sorted( + ( + _runtime_wire(validated, models, state, runtime) + for runtime in state["runtimes"].values() + ), + key=( + (lambda item: order_rank[item["runtime_id"]]) + if order_rank is not None + else (lambda item: item["runtime_id"].encode("utf-8")) + ), + ), + } + document["aggregate_state_digest"] = aggregate_state_digest(document) + return document + + +def serialize_aggregate( + bundle: Bundle | BundleSource, state: dict[str, Any] +) -> bytes: + """Serialize one engine aggregate to exact canonical portable bytes.""" + return canonical_bytes(aggregate_envelope(bundle, state)) + + +def _state_path_for_pointer(machine: MachineModel, pointer: str) -> str: + for path, node in machine.states.items(): + if node.pointer == pointer: + return path + raise ArtifactError("invalid_aggregate_state") + + +def _variable_for_pointer( + machine: MachineModel, pointer: str +) -> tuple[str, str, dict[str, Any]]: + for path, node in machine.states.items(): + prefix = f"{node.pointer}/variables/" + if pointer.startswith(prefix): + for name, declaration in (node.raw.get("variables") or {}).items(): + if pointer == f"{prefix}{_escape_pointer(name)}": + return path, name, declaration + raise ArtifactError("invalid_aggregate_state") + + +def _history_path_for_pointer(machine: MachineModel, pointer: str) -> str: + for path, node in machine.states.items(): + if pointer == f"{node.pointer}/history": + return "$root" if node is machine.root else path + raise ArtifactError("invalid_aggregate_state") + + +def _machine_for_binding( + bundle: Bundle, models: BundleModel, binding: Mapping[str, Any] +) -> MachineModel: + machine_data = binding["machine"] + machine_id = machine_data["machine_id"] + base = models.machine(machine_id) + if ( + machine_data["namespace"] != bundle.namespace + or decimal(machine_data["machine_version"], positive=True) != base.version + ): + raise ArtifactError("invalid_aggregate_state") + root_pointer = machine_data["root_definition_pointer"] + if root_pointer == base.root_pointer: + return base + from .engine import _pointer_get + + root = _pointer_get(bundle.raw, root_pointer) + return MachineModel( + bundle, + base.raw, + machine_index=base.machine_index, + root=root, + root_pointer=root_pointer, + identity_machine=base.identity_machine, + ) + + +def _origin_machine( + resolver: DefinitionResolver, origin: Mapping[str, Any] +) -> tuple[Bundle, MachineModel]: + definition = origin.get("definition") + if not isinstance(definition, Mapping): + raise ArtifactError("invalid_aggregate_state") + fingerprint = definition.get("validated_bundle_fingerprint") + if not isinstance(fingerprint, str): + raise ArtifactError("invalid_aggregate_state") + bundle = _bundle_from_resolver(resolver, fingerprint, source=True, require_trust=True) + return bundle, _machine_for_binding(bundle, BundleModel(bundle), definition) + + +def _validate_immutable_identity( + resolver: DefinitionResolver, + aggregate: Mapping[str, Any], + document: Mapping[str, Any], + role: str, + target: Mapping[str, Any], +) -> None: + origin = document["identity_origin"] + if not isinstance(origin, Mapping) or origin.get("kind") != { + "root": "root", + "component": "component", + "spawned": "owned_spawned_instance", + }[role]: + raise ArtifactError("invalid_aggregate_state") + origin_bundle, origin_machine = _origin_machine(resolver, origin) + root_instance_id = aggregate["root_instance_id"] + runtime_id = document["runtime_id"] + if role == "root": + if ( + origin.get("root_instance_id") != root_instance_id + or target + != { + "root": { + "root_instance_id": root_instance_id, + "root_runtime_id": runtime_id, + } + } + ): + raise ArtifactError("invalid_aggregate_state") + return + if role == "component": + component = target.get("component") + if not isinstance(component, Mapping): + raise ArtifactError("invalid_aggregate_state") + pointer = origin.get("component_definition_pointer") + if not isinstance(pointer, str): + raise ArtifactError("invalid_aggregate_state") + try: + from .engine import _pointer_get + + placement = _pointer_get(origin_bundle.raw, pointer) + declaration_index = int(pointer.rsplit("/", 1)[1]) + except (IndexError, KeyError, TypeError, ValueError): + raise ArtifactError("invalid_aggregate_state") from None + if ( + not isinstance(placement, dict) + or decimal(origin.get("declaration_index")) != declaration_index + or component + != { + "root_instance_id": root_instance_id, + "owner_runtime_id": origin.get("owner_runtime_id"), + "component_id": placement.get("component_id"), + "component_runtime_id": runtime_id, + "activation_sequence": decimal(origin.get("activation_sequence")), + } + ): + raise ArtifactError("invalid_aggregate_state") + return + spawned = target.get("spawned_instance") + if ( + not isinstance(spawned, Mapping) + or spawned + != { + "root_instance_id": root_instance_id, + "instance_id": runtime_id, + "machine_id": origin_machine.machine_id, + "machine_version": origin_machine.version, + } + ): + raise ArtifactError("invalid_aggregate_state") + + +def _runtime_from_wire( + bundle: Bundle, + models: BundleModel, + resolver: DefinitionResolver, + aggregate: Mapping[str, Any], + document: Mapping[str, Any], +) -> dict[str, Any]: + current = document["current_definition"] + if current["validated_bundle_fingerprint"] != bundle.fingerprint: + raise ArtifactError("invalid_aggregate_state") + machine = _machine_for_binding(bundle, models, current) + relation = document["relation"] + role = { + "root": "root", + "component": "component", + "owned_spawned_instance": "spawned", + }[relation["kind"]] + activations: dict[str, int] = {} + for item in document["active_state_activations"]: + path = _state_path_for_pointer(machine, item["state_definition_pointer"]) + if path in activations: + raise ArtifactError("invalid_aggregate_state") + activations[path] = decimal(item["activation_sequence"]) + leaf_pointers = document["active_leaf_state_definition_pointers"] + if len(leaf_pointers) > 1: + raise ArtifactError("invalid_aggregate_state") + active: list[str] = [] + if leaf_pointers: + leaf = machine.states[_state_path_for_pointer(machine, leaf_pointers[0])] + active = [node.path for node in reversed(leaf.ancestors(include_self=True))] + if set(active) != set(activations): + raise ArtifactError("invalid_aggregate_state") + scopes: dict[str, dict[str, Any]] = {path: {} for path in active} + for item in document["variables"]: + path, name, declaration = _variable_for_pointer( + machine, item["variable_declaration_pointer"] + ) + if path not in scopes or name in scopes[path]: + raise ArtifactError("invalid_aggregate_state") + if decimal(item["declaring_state_activation_sequence"]) != activations[path]: + raise ArtifactError("invalid_aggregate_state") + value = decoded_typed_value(item["value"]) + from .engine import _value_matches + + if not _value_matches(value, str(declaration["type"])): + raise ArtifactError("invalid_aggregate_state") + scopes[path][name] = value + for path in active: + declarations = machine.states[path].raw.get("variables") or {} + if set(scopes[path]) != set(declarations): + raise ArtifactError("invalid_aggregate_state") + history: dict[str, list[str] | None] = {} + for item in document["history"]: + path = _history_path_for_pointer(machine, item["history_declaration_pointer"]) + if path in history: + raise ArtifactError("invalid_aggregate_state") + recorded = item["recorded_state_definition_pointers"] + history[path] = ( + None + if recorded is None + else [_state_path_for_pointer(machine, pointer) for pointer in recorded] + ) + state_counters: dict[str, int] = {} + for item in document["next_state_activation_sequences"]: + path = _state_path_for_pointer(machine, item["definition_pointer"]) + if path in state_counters: + raise ArtifactError("invalid_aggregate_state") + state_counters[path] = decimal(item["next_sequence"]) + component_counters: dict[str, int] = {} + for item in document["next_component_activation_sequences"]: + pointer = item["definition_pointer"] + if pointer in component_counters: + raise ArtifactError("invalid_aggregate_state") + component_counters[pointer] = decimal(item["next_sequence"]) + target = copy.deepcopy(document["target_identity"]) + if "component" in target: + target["component"]["activation_sequence"] = decimal( + target["component"]["activation_sequence"] + ) + elif "spawned_instance" in target: + target["spawned_instance"]["machine_version"] = decimal( + target["spawned_instance"]["machine_version"], positive=True + ) + _validate_immutable_identity(resolver, aggregate, document, role, target) + runtime: dict[str, Any] = { + "runtime_id": document["runtime_id"], + "role": role, + "owner_runtime_id": relation.get("owner_runtime_id"), + "machine_id": machine.machine_id, + "machine_version": machine.version, + "root_pointer": machine.root_pointer, + "status": document["status"], + "active": active, + "scopes": scopes, + "history": history, + "fault": None, + "next_spawn_sequence": decimal(document["next_spawn_sequence"]), + "next_state_activation_sequence": state_counters, + "state_activation_sequence": activations, + "next_component_activation_sequence": component_counters, + "components": {}, + "_identity_origin": copy.deepcopy(document["identity_origin"]), + "_target_identity": copy.deepcopy(target), + "_current_definition": copy.deepcopy(current), + "_relation": copy.deepcopy(relation), + } + if role == "component": + runtime.update( + { + "component_id": relation["component_id"], + "component_runtime_id": document["runtime_id"], + "component_definition_pointer": relation[ + "current_component_definition_pointer" + ], + "component_declaration_index": decimal(relation["declaration_index"]), + "component_activation_sequence": decimal(relation["activation_sequence"]), + "owning_state_path": "", + "owning_state_activation_sequence": 0, + "target": target, + } + ) + elif role == "spawned": + holder = relation["lifetime_holder"] + runtime.update( + { + "spawn_sequence": decimal(relation["spawn_sequence"]), + "spawn_action_pointer": relation["current_spawn_action_pointer"], + "instance_reference": target["spawned_instance"], + "holder": ( + None + if holder is None + else { + "pointer": holder["variable_declaration_pointer"], + "state_path": "", + "state_activation_sequence": decimal( + holder["holder_state_activation_sequence"] + ), + } + ), + } + ) + fault = document["fault"] + if fault is not None: + runtime["fault"] = { + key: copy.deepcopy(value) + for key, value in fault.items() + if key != "definition_fingerprint" + } + runtime["fault"]["step_sequence"] = decimal(fault["step_sequence"]) + runtime["_fault_definition_fingerprint"] = fault["definition_fingerprint"] + return runtime + + +def _finish_relationships( + bundle: Bundle, state: dict[str, Any], models: BundleModel +) -> None: + from .engine import _runtime_model + + for runtime in state["runtimes"].values(): + if runtime["role"] == "component": + owner = state["runtimes"].get(runtime["owner_runtime_id"]) + if owner is None: + raise ArtifactError("invalid_aggregate_state") + relation = runtime["_relation"] + owner_machine = _runtime_model(bundle, models, owner) + pointer = relation["current_component_definition_pointer"] + owning = next( + ( + node + for node in owner_machine.states.values() + if pointer.startswith(f"{node.pointer}/components/") + ), + None, + ) + if owning is None or owning.path not in owner["state_activation_sequence"]: + raise ArtifactError("invalid_aggregate_state") + runtime["owning_state_path"] = owning.path + runtime["owning_state_activation_sequence"] = owner[ + "state_activation_sequence" + ][owning.path] + if runtime["component_id"] in owner["components"]: + raise ArtifactError("invalid_aggregate_state") + owner["components"][runtime["component_id"]] = runtime["runtime_id"] + elif runtime["role"] == "spawned" and runtime["holder"] is not None: + owner = state["runtimes"].get(runtime["owner_runtime_id"]) + if owner is None: + raise ArtifactError("invalid_aggregate_state") + owner_machine = _runtime_model(bundle, models, owner) + holder_pointer = runtime["holder"]["pointer"] + path, _name, _declaration = _variable_for_pointer( + owner_machine, holder_pointer + ) + if path not in owner["state_activation_sequence"]: + raise ArtifactError("invalid_aggregate_state") + runtime["holder"]["state_path"] = path + + +def restore_aggregate( + source: ArtifactSource, definition_resolver: DefinitionResolver +) -> RestoredAggregate: + """Verify and restore one portable aggregate without changing the source.""" + document, raw = load_json_artifact(source, "aggregate_state") + if aggregate_state_digest(document) != document["aggregate_state_digest"]: + raise ArtifactError("aggregate_state_digest_mismatch") + fingerprint = document["validated_bundle_fingerprint"] + bundle = _bundle_from_resolver( + definition_resolver, fingerprint, source=True, require_trust=True + ) + if document["namespace"] != bundle.namespace: + raise ArtifactError("invalid_aggregate_state") + models = BundleModel(bundle) + state: dict[str, Any] = { + "validated_bundle_fingerprint": fingerprint, + "namespace": document["namespace"], + "root_instance_id": document["root_instance_id"], + "creation_id": document["creation_id"], + "root_runtime_id": document["root_runtime_id"], + "root_machine_id": document["root_machine_id"], + "status": "running", + "next_logical_step_sequence": decimal(document["next_logical_step_sequence"]), + "next_output_sequence": decimal(document["next_output_sequence"]), + "runtimes": {}, + "fault": None, + "migration_sequence": decimal(document["migration_sequence"]), + "_wire_runtime_order": [ + runtime_document["runtime_id"] for runtime_document in document["runtimes"] + ], + } + for runtime_document in document["runtimes"]: + runtime = _runtime_from_wire( + bundle, models, definition_resolver, document, runtime_document + ) + if runtime["runtime_id"] in state["runtimes"]: + raise ArtifactError("invalid_aggregate_state") + state["runtimes"][runtime["runtime_id"]] = runtime + _finish_relationships(bundle, state, models) + root = state["runtimes"].get(state["root_runtime_id"]) + if root is None or root["role"] != "root": + raise ArtifactError("invalid_aggregate_state") + if ( + document["root_machine_id"] != root["machine_id"] + or decimal(document["root_machine_version"], positive=True) + != root["machine_version"] + or root["_current_definition"]["machine"]["root_definition_pointer"] + != root["root_pointer"] + ): + raise ArtifactError("invalid_aggregate_state") + state["status"] = root["status"] + state["fault"] = copy.deepcopy(root["fault"]) + from .engine import _valid_prior_state + + if not _valid_prior_state(state, bundle): + raise ArtifactError("invalid_aggregate_state") + canonical = canonical_bytes(document) + return RestoredAggregate( + bundle=bundle, + state=state, + aggregate_envelope=copy.deepcopy(document), + canonical_bytes=canonical, + source_bytes=raw, + ) + + +def restore_aggregate_package( + source: ArtifactSource, artifact_resolver: ArtifactResolver +) -> RestoredAggregatePackage: + """Verify a transport package and seed one mutable resolver atomically.""" + document, _raw = load_json_artifact(source, "aggregate_state_package") + definitions: dict[str, Bundle] = {} + descriptors: dict[str, dict[str, Any]] = {} + try: + for attachment in document["normalized_definitions"]: + bundle = bundle_from_attachment(attachment) + if bundle.fingerprint in definitions: + raise ArtifactError("invalid_aggregate_state_package") + definitions[bundle.fingerprint] = bundle + for descriptor in document["migration_descriptors"]: + digest = descriptor["migration_descriptor_digest"] + if digest in descriptors or migration_descriptor_digest(descriptor) != digest: + raise ArtifactError("invalid_aggregate_state_package") + descriptors[digest] = copy.deepcopy(descriptor) + route = tuple(document["migration_route"]) + if len(set(route)) != len(route): + raise ArtifactError("invalid_aggregate_state_package") + for fingerprint, bundle in definitions.items(): + existing_definition = artifact_resolver.resolve_definition(fingerprint) + if existing_definition is not None: + current_bundle = ( + existing_definition + if isinstance(existing_definition, Bundle) + else load_bundle(existing_definition) + ) + if ( + current_bundle.fingerprint != fingerprint + or canonical_bytes(typed_value(current_bundle.raw)) + != canonical_bytes(typed_value(bundle.raw)) + ): + raise ArtifactError("definition_fingerprint_mismatch") + for digest, descriptor in descriptors.items(): + existing_descriptor = artifact_resolver.resolve_migration_descriptor(digest) + if existing_descriptor is not None: + current_descriptor, _ = load_json_artifact( + existing_descriptor, "migration_descriptor" + ) + if canonical_bytes(current_descriptor) != canonical_bytes(descriptor): + raise ArtifactError("invalid_migration_descriptor") + except (ArtifactError, KeyError, TypeError, ValidationError) as exc: + if isinstance(exc, ArtifactError) and ( + exc.code.startswith("unsupported_") + or exc.code + in {"definition_fingerprint_mismatch", "invalid_migration_descriptor"} + ): + raise exc + raise ArtifactError("invalid_aggregate_state_package") from exc + put_definition = getattr(artifact_resolver, "put_definition", None) + put_descriptor = getattr(artifact_resolver, "put_migration_descriptor", None) + if (definitions and not callable(put_definition)) or ( + descriptors and not callable(put_descriptor) + ): + raise ArtifactError("invalid_aggregate_state_package") + overlay = _PackageResolver(artifact_resolver, definitions, descriptors) + try: + aggregate = restore_aggregate(document["aggregate_state"], overlay) + store_definition = cast(Callable[[str, Bundle], None], put_definition) + store_descriptor = cast( + Callable[[str, Mapping[str, Any]], None], put_descriptor + ) + for fingerprint, bundle in definitions.items(): + store_definition(fingerprint, bundle) + for digest, descriptor in descriptors.items(): + store_descriptor(digest, descriptor) + except ArtifactError: + raise + return RestoredAggregatePackage( + aggregate=aggregate, + migration_route=route, + package_document=copy.deepcopy(document), + ) + + +class _PackageResolver: + def __init__( + self, + parent: ArtifactResolver, + definitions: Mapping[str, Bundle], + descriptors: Mapping[str, dict[str, Any]], + ) -> None: + self.parent = parent + self.definitions = definitions + self.descriptors = descriptors + + def resolve_definition(self, fingerprint: str) -> Bundle | BundleSource | None: + return self.definitions.get(fingerprint) or self.parent.resolve_definition( + fingerprint + ) + + def definition_is_trusted(self, fingerprint: str) -> bool: + return fingerprint in self.definitions or self.parent.definition_is_trusted( + fingerprint + ) + + def resolve_migration_descriptor(self, digest: str) -> ArtifactSource | None: + return self.descriptors.get(digest) or self.parent.resolve_migration_descriptor( + digest + ) + + def migration_descriptor_is_trusted(self, digest: str) -> bool: + return digest in self.descriptors or self.parent.migration_descriptor_is_trusted( + digest + ) + + +def aggregate_shape_fingerprint(bundle: Bundle | BundleSource) -> str: + """Compute the exact state-bearing definition fingerprint from SPEC §16.6.""" + validated = bundle if isinstance(bundle, Bundle) else load_bundle(bundle) + + def variable_projection( + declaration: Mapping[str, Any], pointer: str + ) -> dict[str, Any]: + result: dict[str, Any] = { + "declaration_pointer": pointer, + "type": declaration["type"], + "nullable": ( + bool(declaration.get("nullable")) + if declaration["type"] == "instance_reference" + else False + ), + "input": bool(declaration.get("input")), + "external": bool(declaration.get("external")), + } + if declaration.get("machine_id") is not None: + result["machine_id"] = declaration["machine_id"] + return result + + def action_spawn_sites( + machine: MachineModel, + state: StateNode, + actions: Any, + pointer: str, + ) -> list[dict[str, Any]]: + sites: list[dict[str, Any]] = [] + for index, action in enumerate(actions or []): + if "spawn" not in action: + continue + spawn = action["spawn"] + holder_pointer = None + if "bind_to" in spawn: + holder_pointer = _resolve_variable_pointer( + machine, state, spawn["bind_to"] + ) + sites.append( + { + "action_pointer": f"{pointer}/{index}/spawn", + "machine_id": spawn["machine_id"], + "holder_variable_declaration_pointer": holder_pointer, + } + ) + return sites + + def state_projection( + machine: MachineModel, state: StateNode + ) -> dict[str, Any]: + result: dict[str, Any] = { + "definition_pointer": state.pointer, + "type": state.type, + } + if state.type == "composite": + result["history"] = state.raw.get("history", "none") + variables = [ + variable_projection( + declaration, f"{state.pointer}/variables/{_escape_pointer(name)}" + ) + for name, declaration in (state.raw.get("variables") or {}).items() + ] + variables.sort(key=lambda item: item["declaration_pointer"].encode("utf-8")) + if variables: + result["variables"] = variables + children = [ + state_projection(machine, child) + for _name, child in sorted( + state.children.items(), key=lambda item: item[0].encode("utf-8") + ) + ] + if children: + result["states"] = children + components: list[dict[str, Any]] = [] + for index, placement in enumerate(state.raw.get("components") or []): + item: dict[str, Any] = { + "declaration_pointer": f"{state.pointer}/components/{index}", + "declaration_index": index, + "component_id": placement["component_id"], + } + if "machine_id" in placement: + item["machine_id"] = placement["machine_id"] + else: + inline = BundleModel(validated).inline_component( + machine, placement, f"{state.pointer}/components/{index}" + ) + item["inline_root"] = state_projection(inline, inline.root) + components.append(item) + if components: + result["components"] = components + sites = action_spawn_sites( + machine, state, state.raw.get("entry"), f"{state.pointer}/entry" + ) + sites += action_spawn_sites( + machine, state, state.raw.get("exit"), f"{state.pointer}/exit" + ) + for event_name, transition_or_list in (state.raw.get("on_events") or {}).items(): + transitions = ( + transition_or_list + if isinstance(transition_or_list, list) + else [transition_or_list] + ) + for transition_index, transition in enumerate(transitions): + suffix = ( + f"/{transition_index}" if isinstance(transition_or_list, list) else "" + ) + sites += action_spawn_sites( + machine, + state, + transition.get("action"), + ( + f"{state.pointer}/on_events/{_escape_pointer(event_name)}" + f"{suffix}/action" + ), + ) + sites.sort(key=lambda item: item["action_pointer"].encode("utf-8")) + if sites: + result["spawn_sites"] = sites + return result + + models = BundleModel(validated) + machine_values = [ + { + "machine_id": machine.machine_id, + "version": machine.version, + "root": state_projection(machine, machine.root), + } + for machine in models.machines.values() + ] + tree = { + "format": 1, + "namespace": validated.namespace, + "machines": machine_values, + } + return hash_value( + ["determa-aggregate-shape-fingerprint-1", typed_value(tree)] + ) + + +def _resolve_variable_pointer( + machine: MachineModel, state: StateNode, name: str +) -> str: + current: StateNode | None = state + while current is not None: + if name in (current.raw.get("variables") or {}): + return f"{current.pointer}/variables/{_escape_pointer(name)}" + current = current.parent + raise ArtifactError("invalid_migration_descriptor") diff --git a/tests/test_persistence.py b/tests/test_persistence.py new file mode 100644 index 0000000..bd646a3 --- /dev/null +++ b/tests/test_persistence.py @@ -0,0 +1,271 @@ +from __future__ import annotations + +import copy + +import pytest + +from determa.state import ( + ArtifactError, + MemoryArtifactResolver, + aggregate_envelope, + aggregate_shape_fingerprint, + create, + load_bundle, + migrate_aggregate, + restore_aggregate, + serialize_aggregate, +) +from determa.state.wire import ( + aggregate_state_digest, + canonical_bytes, + migration_descriptor_digest, + strict_json, +) + +SOURCE = """ +format: 1 +namespace: tests.persistence +machines: + - machine_id: job + version: 1 + root: + variables: + count: {type: int, init: 3} +""" + +COMPONENT_SOURCE = """ +format: 1 +namespace: tests.persistence_components +machines: + - machine_id: owner + root: + type: parallel + components: + - component_id: worker + machine_id: worker + - component_id: second_worker + machine_id: worker + - machine_id: worker + root: {} +""" + +SPAWN_SOURCE = """ +format: 1 +namespace: tests.persistence_spawn +machines: + - machine_id: owner + root: + variables: + worker_reference: + type: instance_reference + machine_id: worker + nullable: true + init: null + entry: + - spawn: + machine_id: worker + bind_to: worker_reference + - machine_id: worker + version: 1 + root: {} +""" + + +def _resolver_for(bundle): + return MemoryArtifactResolver(definitions={bundle.fingerprint: bundle}) + + +def _redigest(document): + document["aggregate_state_digest"] = aggregate_state_digest(document) + return canonical_bytes(document) + + +def test_aggregate_round_trip_is_canonical_and_does_not_mutate_state() -> None: + bundle = load_bundle(SOURCE) + created = create(bundle, "job", "job-1", "create-1", {}) + state = created["state"] + assert state is not None + snapshot = copy.deepcopy(state) + + encoded = serialize_aggregate(bundle, state) + resolver = MemoryArtifactResolver(definitions={bundle.fingerprint: bundle}) + restored = restore_aggregate(encoded, resolver) + + assert state == snapshot + assert restored.aggregate_envelope == aggregate_envelope(bundle, restored.state) + assert restored.canonical_bytes == encoded + assert canonical_bytes(restored.aggregate_envelope) == encoded + + +def test_restore_rejects_root_header_version_mismatch() -> None: + bundle = load_bundle(SOURCE) + created = create(bundle, "job", "job-1", "create-1", {}) + assert created["state"] is not None + document = aggregate_envelope(bundle, created["state"]) + document["root_machine_version"] = "2" + + with pytest.raises(ArtifactError) as raised: + restore_aggregate(_redigest(document), _resolver_for(bundle)) + + assert raised.value.code == "invalid_aggregate_state" + + +@pytest.mark.parametrize( + ("field", "forged"), + [ + ("root_instance_id", "forged-root"), + ("owner_runtime_id", "sha256:" + "0" * 64), + ("component_id", "forged_component"), + ("component_runtime_id", "sha256:" + "1" * 64), + ("activation_sequence", "7"), + ], +) +def test_restore_rejects_forged_component_target_identity( + field: str, forged: str +) -> None: + bundle = load_bundle(COMPONENT_SOURCE) + created = create(bundle, "owner", "owner-1", "create-1", {}) + assert created["state"] is not None + document = aggregate_envelope(bundle, created["state"]) + runtime = next( + item for item in document["runtimes"] if item["relation"]["kind"] == "component" + ) + runtime["target_identity"]["component"][field] = forged + + with pytest.raises(ArtifactError) as raised: + restore_aggregate(_redigest(document), _resolver_for(bundle)) + + assert raised.value.code == "invalid_aggregate_state" + + +@pytest.mark.parametrize( + ("field", "forged"), + [ + ("root_instance_id", "forged-root"), + ("instance_id", "sha256:" + "2" * 64), + ("machine_id", "owner"), + ("machine_version", "7"), + ], +) +def test_restore_rejects_forged_spawned_target_identity( + field: str, forged: str +) -> None: + bundle = load_bundle(SPAWN_SOURCE) + created = create(bundle, "owner", "owner-1", "create-1", {}) + assert created["state"] is not None + document = aggregate_envelope(bundle, created["state"]) + runtime = next( + item + for item in document["runtimes"] + if item["relation"]["kind"] == "owned_spawned_instance" + ) + runtime["target_identity"]["spawned_instance"][field] = forged + + with pytest.raises(ArtifactError) as raised: + restore_aggregate(_redigest(document), _resolver_for(bundle)) + + assert raised.value.code == "invalid_aggregate_state" + + +def test_migration_definition_byte_limit_uses_typed_normalized_bundle() -> None: + version = 9_007_199_254_740_992 + source = load_bundle( + f""" +format: 1 +namespace: tests.persistence_unsafe_version +meta: {{release: source}} +machines: + - machine_id: job + version: {version} + root: {{}} +""" + ) + target = load_bundle( + f""" +format: 1 +namespace: tests.persistence_unsafe_version +meta: {{release: target}} +events: + notice: {{direction: internal}} +machines: + - machine_id: job + version: {version} + root: {{}} +""" + ) + shape = aggregate_shape_fingerprint(source) + assert shape == aggregate_shape_fingerprint(target) + descriptor = { + "migration_descriptor_format": "determa.aggregate_migration", + "migration_descriptor_schema_version": 1, + "source_machine_format": 1, + "target_machine_format": 1, + "source_validated_bundle_fingerprint": source.fingerprint, + "target_validated_bundle_fingerprint": target.fingerprint, + "source_aggregate_shape_fingerprint": shape, + "target_aggregate_shape_fingerprint": shape, + "mode": "compatible", + "mappings": { + "machines": [], + "active_states": [], + "variables": [], + "history": [], + "components": [], + "owned_runtimes": [], + "lifetime_holders": [], + "counters": [], + }, + "terminal_policy": {"completed": "preserve", "faulted": "preserve"}, + "resource_requirements": { + "maximum_transformed_output_bytes": "0", + "maximum_cel_expression_length": "0", + "maximum_cel_ast_nodes": "0", + "maximum_cel_evaluation_steps": "0", + }, + } + descriptor["migration_descriptor_digest"] = migration_descriptor_digest(descriptor) + resolver = MemoryArtifactResolver( + definitions={source.fingerprint: source, target.fingerprint: target}, + migration_descriptors={ + descriptor["migration_descriptor_digest"]: descriptor, + }, + ) + created = create(source, "job", "job-1", "create-1", {}) + assert created["state"] is not None + + result = migrate_aggregate( + serialize_aggregate(source, created["state"]), + target.fingerprint, + [descriptor["migration_descriptor_digest"]], + resolver, + maintenance_mode=False, + ) + + assert result.failure is None + assert result.aggregate_envelope is not None + assert result.aggregate_envelope["root_machine_version"] == str(version) + + +def test_untrusted_definition_fails_closed() -> None: + bundle = load_bundle(SOURCE) + created = create(bundle, "job", "job-1", "create-1", {}) + assert created["state"] is not None + encoded = serialize_aggregate(bundle, created["state"]) + resolver = MemoryArtifactResolver( + definitions={bundle.fingerprint: bundle}, trusted_definitions=[] + ) + + with pytest.raises(ArtifactError, match="definition_untrusted") as raised: + restore_aggregate(encoded, resolver) + + assert raised.value.code == "definition_untrusted" + + +def test_strict_json_rejects_duplicate_members_and_nonfinite_numbers() -> None: + with pytest.raises(ArtifactError) as duplicate: + strict_json(b'{"value":1,"value":2}') + assert duplicate.value.code == "duplicate_json_name" + + with pytest.raises(ArtifactError) as nonfinite: + strict_json(b'{"value":NaN}') + assert nonfinite.value.code == "invalid_json_value"