diff --git a/CHANGELOG.md b/CHANGELOG.md index fae87d7..71c6d51 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,11 @@ # Changelog +## 0.3.0 - 2026-08-02 +- Added deterministic compilation of admitted plans, typed provider contracts and + resolution, explicit execution state and evidence, dry run, retry, compensation, + and interruption-safe in-memory resume through a non-operational fake provider. +- Added an `execute` demonstration command and runtime architecture ADR. Durable + repeated reconciliation and operational providers remain deferred. + ## 0.2.0 - 2026-08-02 - Added canonical world and manifest digests, revision lineage, semantic change sets, offline observed-state evidence, mandate-aware admission, explicit risks and approvals, and deterministic provider-neutral reconciliation plans. - Added `diff`, `admit`, and `plan` commands. Planning is deliberately non-executable and keeps canonical intent, capabilities, provider bindings, and observed drift separate. diff --git a/docs/adr/0003-compile-admitted-plans-into-provider-bound-operations.md b/docs/adr/0003-compile-admitted-plans-into-provider-bound-operations.md new file mode 100644 index 0000000..67d0847 --- /dev/null +++ b/docs/adr/0003-compile-admitted-plans-into-provider-bound-operations.md @@ -0,0 +1,29 @@ +# ADR: Compile admitted plans into provider-bound executable operations + +**Status:** Accepted + +## Context + +v0.2 plans express sovereign meaning and provenance. Provider selection inside them +would make replaceable implementation details canonical. Execution also needs +recovery state and attributable evidence that ordinary logs cannot provide. + +## Decision + +Keep reconciliation plans provider-neutral and add a deterministic compilation stage +that resolves capabilities into provider-bound executable operations. Provider apply +success and observed convergence remain separate records. Operation state and valid +transitions are explicit, and evidence is an execution product with authority, +mandate, resource, capability, provider, and revision provenance. + +Providers do not offer a distributed transaction. The runtime may retry and perform +reverse-order semantic compensation, but does not promise atomicity, snapshots, or +equivalence between compensation, rollback, and cleanup. A complete in-memory fake +precedes operational providers so contracts and failures can be proven safely. + +## Consequences + +Compilation is deterministic and independently testable. Execution is explainable +and resumable within an injected in-memory repository. Callers must supply a registry +and bindings. Persistence, crash-safe recovery, and repeated reconciliation are +deferred to v0.4; operational naming is deferred to v0.5. diff --git a/docs/roadmap.md b/docs/roadmap.md index 106387e..a0d4710 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -1,5 +1,9 @@ # NetSovereign roadmap +Implementation status: v0.1, v0.2, and the bounded in-memory v0.3 runtime core are +implemented and tested. v0.4 durability and v0.5 naming materialisation remain +planned and are not implied by the v0.3 interfaces. + NetSovereign develops authority and intent before operational adapters. Versions v0.2 through v0.4 are a single dependency chain: real DNS, PKI, identity, or gateway providers must not begin until change planning, the runtime core, and durable local control have established their boundaries. diff --git a/docs/runtime-core-v0.3.md b/docs/runtime-core-v0.3.md new file mode 100644 index 0000000..3620ab2 --- /dev/null +++ b/docs/runtime-core-v0.3.md @@ -0,0 +1,58 @@ +# NetEngine runtime core (v0.3) + +v0.3 adds a provider-neutral execution boundary without replacing the legacy +phase-oriented application. The phase runner and its operational handlers remain a +compatibility path; the runtime core never calls them and no existing handler is +presented as a provider. + +## Pipeline and contracts + +An admitted v0.2 `ReconciliationPlan` remains an intent artefact. `compile_plan` +topologically orders its steps and deterministically resolves the generic +`resource.manage` capability through an injected `ProviderRegistry`. It emits a +provider-bound `ExecutablePlan` with revision, authority, mandate, target, binding, +preconditions, expected result, idempotency key, retry policy, failure posture, and a +deterministic integrity fingerprint. Compilation describes providers but never calls +validation, mutation, or observation. + +The provider protocol separates `validate`, `apply`, `observe`, `compensate`, and +`delete`. Apply acceptance is not convergence: only a later matching observation +verifies an operation. Resolution rejects missing, incompatible, ambiguous, +explicitly unsupported, and unhealthy providers with stable error codes. + +## State, evidence, and recovery + +Every operation owns explicit transitions, attempts, provider results, observations, +and failure classification. Invalid transitions fail closed. A run derives status +from operation state rather than logs. Evidence separately records validation, +application, observation, conformance, simulation, compensation, and failure with +world revision and authority provenance. Secret-looking mapping keys are redacted. + +Retries repeat the same transition only for explicit retryable classes. Compensation +is a semantic counter-action and runs in reverse completion order; rollback represents +restoring prior state; cleanup removes orphaned artefacts. They are distinct and no +cross-provider atomicity is claimed. If interruption leaves an operation running +after apply, resume observes first and accepts convergence without applying twice. +Changed plan fingerprints are rejected. This lasts only as long as the in-memory +repository: crash-safe persistence belongs to v0.4. + +## Dry run and demonstration + +Dry run resolves and validates normally, records predictions and evidence, and never +calls a mutating method. Providers must declare dry-run support. + +```bash +netsovereign plan examples/minimal/world.yaml proposed.yaml --output admitted-plan.json +netsovereign execute admitted-plan.json +netsovereign execute admitted-plan.json --dry-run +``` + +The fake supports deterministic create/update/delete-style application, observation, +idempotent replay, stable resource IDs, typed failure injection, and compensation +without network, process, container, database, DNS, or host access. + +## Boundaries + +v0.3 is sequential and in-process. It adds no daemon, polling, operational provider, +database journal, lock, infrastructure snapshot, distributed transaction, or API. +Durable storage and repeated reconciliation are v0.4; authoritative naming is v0.5. diff --git a/pyproject.toml b/pyproject.toml index 22a21c0..732b442 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,8 +4,8 @@ build-backend = "hatchling.build" [project] name = "netsovereign" -version = "0.2.0" -description = "Provider-neutral domain foundation for sovereign digital worlds" +version = "0.3.0" +description = "Provider-neutral compiler and reconciliation runtime for sovereign worlds" readme = "README.md" requires-python = ">=3.12" license = {file = "LICENSE"} diff --git a/src/netsovereign/cli.py b/src/netsovereign/cli.py index 54ac478..1ea4bb3 100644 --- a/src/netsovereign/cli.py +++ b/src/netsovereign/cli.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio import json from datetime import datetime from pathlib import Path @@ -17,10 +18,14 @@ ApprovalEvidence, ObservedStateSnapshot, ParentRevisionReference, + ReconciliationPlan, admit_change, build_plan, compare_worlds, ) +from .providers.fake import FakeProvider +from .providers.registry import ProviderRegistry +from .runtime import InMemoryExecutionRepository, RuntimeExecutor, compile_plan from .specification import WorldSpec from .validation import has_errors, validate_spec @@ -197,5 +202,29 @@ def plan( raise typer.Exit(3) +@app.command("execute") +def execute_command( + plan_file: Path, + dry_run: Annotated[bool, typer.Option("--dry-run")] = False, +) -> None: + """Compile and execute an admitted plan through the in-memory fake provider.""" + try: + intent_plan = ReconciliationPlan.model_validate( + yaml.safe_load(plan_file.read_text(encoding="utf-8")) + ) + registry = ProviderRegistry() + registry.register(FakeProvider()) + executable = compile_plan(intent_plan, registry) + report = asyncio.run( + RuntimeExecutor(registry, InMemoryExecutionRepository()).execute( + executable, dry_run=dry_run + ) + ) + except (OSError, ValidationError, ValueError) as exc: + typer.echo(f"EXECUTION_ERROR {plan_file}: {exc}", err=True) + raise typer.Exit(1) from exc + _emit(report) + + if __name__ == "__main__": app() diff --git a/src/netsovereign/providers/__init__.py b/src/netsovereign/providers/__init__.py new file mode 100644 index 0000000..4941b22 --- /dev/null +++ b/src/netsovereign/providers/__init__.py @@ -0,0 +1,7 @@ +"""Provider-neutral capability contracts and the non-operational fake provider.""" + +from .contracts import * # noqa: F403 +from .fake import FakeProvider +from .registry import ProviderRegistry + +__all__ = ["FakeProvider", "ProviderRegistry"] diff --git a/src/netsovereign/providers/contracts.py b/src/netsovereign/providers/contracts.py new file mode 100644 index 0000000..c2c548d --- /dev/null +++ b/src/netsovereign/providers/contracts.py @@ -0,0 +1,89 @@ +"""Small, capability-oriented provider contract used by the runtime compiler.""" + +from __future__ import annotations + +from enum import StrEnum +from typing import Any, Protocol, runtime_checkable + +from pydantic import Field + +from ..base import DomainModel + + +class FailureClass(StrEnum): + VALIDATION = "validation" + CAPABILITY_UNAVAILABLE = "capability_unavailable" + PRECONDITION = "precondition_failed" + CONFLICT = "conflict" + TRANSIENT = "transient_provider_failure" + PERMANENT = "permanent_provider_failure" + TIMEOUT = "timeout" + OBSERVATION_MISMATCH = "observation_mismatch" + COMPENSATION = "compensation_failure" + INTERNAL = "internal_engine_failure" + + +class CapabilityRequirement(DomainModel): + id: str + version: str = "1.0" + + +class CapabilityDeclaration(DomainModel): + id: str + version: str = "1.0" + dry_run: bool = True + idempotent: bool = True + compensation: bool = True + limitations: list[str] = Field(default_factory=list) + + +class ProviderDescriptor(DomainModel): + id: str + version: str + binding_id: str + available: bool = True + healthy: bool = True + capabilities: list[CapabilityDeclaration] + + +class ProviderContext(DomainModel): + run_id: str + operation_id: str + idempotency_key: str + dry_run: bool = False + + +class ValidationResult(DomainModel): + valid: bool + predicted_action: str | None = None + evidence: dict[str, Any] = Field(default_factory=dict) + failure: FailureClass | None = None + message: str | None = None + + +class ProviderResult(DomainModel): + success: bool + provider_resource_id: str | None = None + output: dict[str, Any] = Field(default_factory=dict) + evidence: dict[str, Any] = Field(default_factory=dict) + failure: FailureClass | None = None + retryable: bool = False + message: str | None = None + + +class ObservationResult(DomainModel): + observed: bool + matches_expected: bool + state: dict[str, Any] = Field(default_factory=dict) + evidence: dict[str, Any] = Field(default_factory=dict) + message: str | None = None + + +@runtime_checkable +class Provider(Protocol): + def describe(self) -> ProviderDescriptor: ... + async def validate(self, operation: Any, context: ProviderContext) -> ValidationResult: ... + async def apply(self, operation: Any, context: ProviderContext) -> ProviderResult: ... + async def observe(self, operation: Any, context: ProviderContext) -> ObservationResult: ... + async def compensate(self, operation: Any, context: ProviderContext) -> ProviderResult: ... + async def delete(self, operation: Any, context: ProviderContext) -> ProviderResult: ... diff --git a/src/netsovereign/providers/fake.py b/src/netsovereign/providers/fake.py new file mode 100644 index 0000000..e9a0b26 --- /dev/null +++ b/src/netsovereign/providers/fake.py @@ -0,0 +1,114 @@ +"""Inspectable, deterministic provider with no operational side effects.""" + +from __future__ import annotations + +from collections import defaultdict +from hashlib import sha256 +from typing import Any + +from .contracts import ( + CapabilityDeclaration, + FailureClass, + ObservationResult, + ProviderContext, + ProviderDescriptor, + ProviderResult, + ValidationResult, +) + + +class FakeProvider: + def __init__(self, provider_id: str = "fake", *, dry_run: bool = True) -> None: + self.provider_id = provider_id + self.dry_run = dry_run + self.available = True + self.healthy = True + self.state: dict[str, Any] = {} + self.call_history: list[tuple[str, str]] = [] + self.failures: dict[str, list[FailureClass]] = defaultdict(list) + self.observation_mismatches: set[str] = set() + + def describe(self) -> ProviderDescriptor: + return ProviderDescriptor( + id=self.provider_id, + version="1.0", + binding_id=f"{self.provider_id}:memory", + available=self.available, + healthy=self.healthy, + capabilities=[CapabilityDeclaration(id="resource.manage", dry_run=self.dry_run)], + ) + + def inject_failure(self, operation_id: str, failure: FailureClass) -> None: + self.failures[operation_id].append(failure) + + async def validate(self, operation: Any, context: ProviderContext) -> ValidationResult: + self.call_history.append(("validate", operation.id)) + if context.dry_run and not self.dry_run: + return ValidationResult( + valid=False, + failure=FailureClass.CAPABILITY_UNAVAILABLE, + message="dry run is unsupported", + ) + if ( + self.failures[operation.id] + and self.failures[operation.id][0] == FailureClass.VALIDATION + ): + self.failures[operation.id].pop(0) + return ValidationResult( + valid=False, failure=FailureClass.VALIDATION, message="injected" + ) + return ValidationResult( + valid=True, + predicted_action=f"{operation.operation_type}:{operation.target}", + evidence={"validated": True}, + ) + + async def apply(self, operation: Any, context: ProviderContext) -> ProviderResult: + self.call_history.append(("apply", operation.id)) + if self.failures[operation.id]: + failure = self.failures[operation.id].pop(0) + return ProviderResult( + success=False, + failure=failure, + retryable=failure == FailureClass.TRANSIENT, + message="injected", + ) + resource_id = "fake-" + sha256(operation.target.encode()).hexdigest()[:12] + if operation.operation_type in {"remove", "delete"}: + self.state.pop(operation.target, None) + else: + self.state[operation.target] = operation.expected + return ProviderResult( + success=True, + provider_resource_id=resource_id, + output={"action": operation.operation_type}, + evidence={"mutated": True}, + ) + + async def observe(self, operation: Any, context: ProviderContext) -> ObservationResult: + self.call_history.append(("observe", operation.id)) + actual = self.state.get(operation.target) + expected = operation.expected + matches = actual == expected and operation.id not in self.observation_mismatches + return ObservationResult( + observed=actual is not None, + matches_expected=matches, + state={"value": actual}, + evidence={"matches": matches}, + ) + + async def compensate(self, operation: Any, context: ProviderContext) -> ProviderResult: + self.call_history.append(("compensate", operation.id)) + if ( + self.failures[operation.id] + and self.failures[operation.id][0] == FailureClass.COMPENSATION + ): + self.failures[operation.id].pop(0) + return ProviderResult( + success=False, failure=FailureClass.COMPENSATION, message="injected" + ) + self.state.pop(operation.target, None) + return ProviderResult(success=True, evidence={"compensated": True}) + + async def delete(self, operation: Any, context: ProviderContext) -> ProviderResult: + return await self.compensate(operation, context) diff --git a/src/netsovereign/providers/registry.py b/src/netsovereign/providers/registry.py new file mode 100644 index 0000000..3336508 --- /dev/null +++ b/src/netsovereign/providers/registry.py @@ -0,0 +1,74 @@ +"""Deterministic, injected provider registry.""" + +from __future__ import annotations + +from .contracts import CapabilityRequirement, Provider, ProviderDescriptor + + +class ProviderResolutionError(ValueError): + """A stable resolution failure with a machine-readable code.""" + + def __init__(self, code: str, message: str): + self.code = code + super().__init__(message) + + +def _compatible(required: str, offered: str) -> bool: + return required.split(".", 1)[0] == offered.split(".", 1)[0] + + +class ProviderRegistry: + def __init__(self) -> None: + self._providers: dict[str, Provider] = {} + + def register(self, provider: Provider) -> None: + identity = provider.describe().id + if identity in self._providers: + raise ProviderResolutionError( + "duplicate_provider", f"provider {identity!r} is registered" + ) + self._providers[identity] = provider + + def descriptors(self) -> list[ProviderDescriptor]: + return [self._providers[key].describe() for key in sorted(self._providers)] + + def get(self, provider_id: str) -> Provider: + try: + return self._providers[provider_id] + except KeyError as exc: + raise ProviderResolutionError("requested_provider_absent", provider_id) from exc + + def resolve( + self, requirement: CapabilityRequirement, provider_id: str | None = None + ) -> Provider: + candidates = [self.get(provider_id)] if provider_id else list(self._providers.values()) + supported: list[Provider] = [] + incompatible = False + for provider in candidates: + descriptor = provider.describe() + declarations = [item for item in descriptor.capabilities if item.id == requirement.id] + if declarations and any( + _compatible(requirement.version, item.version) for item in declarations + ): + supported.append(provider) + elif declarations: + incompatible = True + if not supported: + code = ( + "capability_version_incompatible" + if incompatible + else ( + "requested_provider_unsupported" + if provider_id + else "no_provider_supports_capability" + ) + ) + raise ProviderResolutionError( + code, f"cannot resolve {requirement.id}@{requirement.version}" + ) + if len(supported) > 1: + raise ProviderResolutionError("ambiguous_provider", requirement.id) + descriptor = supported[0].describe() + if not descriptor.available or not descriptor.healthy: + raise ProviderResolutionError("provider_unavailable", descriptor.id) + return supported[0] diff --git a/src/netsovereign/runtime/__init__.py b/src/netsovereign/runtime/__init__.py new file mode 100644 index 0000000..e92283b --- /dev/null +++ b/src/netsovereign/runtime/__init__.py @@ -0,0 +1,48 @@ +"""Provider-neutral runtime public API. + +The implementation is split by responsibility while this module preserves the +original v0.3 import surface. +""" + +from .compiler import compile_plan +from .executor import RuntimeExecutor +from .models import ( + TERMINAL_STATES, + TRANSITIONS, + Attempt, + Evidence, + EvidenceKind, + ExecutableOperation, + ExecutablePlan, + ExecutionReport, + ExecutionRun, + FailurePosture, + OperationExecution, + OperationState, + PlanStatus, + RetryPolicy, + Transition, +) +from .repository import ExecutionRepository, InMemoryExecutionRepository + +__all__ = [ + "Attempt", + "Evidence", + "EvidenceKind", + "ExecutableOperation", + "ExecutablePlan", + "ExecutionReport", + "ExecutionRepository", + "ExecutionRun", + "FailurePosture", + "InMemoryExecutionRepository", + "OperationExecution", + "OperationState", + "PlanStatus", + "RetryPolicy", + "RuntimeExecutor", + "TERMINAL_STATES", + "TRANSITIONS", + "Transition", + "compile_plan", +] diff --git a/src/netsovereign/runtime/compiler.py b/src/netsovereign/runtime/compiler.py new file mode 100644 index 0000000..5928c87 --- /dev/null +++ b/src/netsovereign/runtime/compiler.py @@ -0,0 +1,87 @@ +"""Pure compilation of admitted intent plans into provider-bound plans.""" + +from __future__ import annotations + +from typing import Any + +from ..canonical import digest +from ..planning import ReconciliationPlan +from ..providers.contracts import CapabilityRequirement +from ..providers.registry import ProviderRegistry +from .models import ExecutableOperation, ExecutablePlan, FailurePosture, RetryPolicy + + +def _ordered_steps(plan: ReconciliationPlan) -> list[Any]: + by_id = {step.id: step for step in plan.steps} + remaining = set(by_id) + ordered: list[Any] = [] + while remaining: + ready = sorted( + item for item in remaining if set(by_id[item].depends_on) <= {x.id for x in ordered} + ) + if not ready: + raise ValueError("dependency_cycle") + for item in ready: + ordered.append(by_id[item]) + remaining.remove(item) + return ordered + + +def compile_plan( + plan: ReconciliationPlan, + registry: ProviderRegistry, + bindings: dict[str, str] | None = None, + *, + retry_policy: RetryPolicy | None = None, + failure_posture: FailurePosture = FailurePosture.STOP, +) -> ExecutablePlan: + """Purely bind a copied v0.2 plan; providers are described but never invoked.""" + if not plan.admitted: + raise ValueError("only admitted plans can be compiled") + operations: list[ExecutableOperation] = [] + for order, step in enumerate(_ordered_steps(plan)): + requirement = CapabilityRequirement(id="resource.manage") + requested = (bindings or {}).get(step.id) + provider = registry.resolve(requirement, requested) + descriptor = provider.describe() + expected = step.expected_outcomes[0].value if step.expected_outcomes else None + stable = {"plan": plan.plan_digest, "step": step.id, "provider": descriptor.id} + operations.append( + ExecutableOperation( + id="operation-" + digest(stable)[:16], + source_step_id=step.id, + authority_id=step.authority_id, + mandate_id=step.mandate_id, + target=step.target, + operation_type=step.action, + capability=requirement, + provider_id=descriptor.id, + provider_binding_id=descriptor.binding_id, + depends_on=list(step.depends_on), + preconditions=[item.model_dump(mode="json") for item in step.preconditions], + expected=expected, + idempotency_key=digest({**stable, "expected": expected}), + retry_policy=retry_policy or RetryPolicy(), + failure_posture=failure_posture, + dry_run_compatible=next( + x for x in descriptor.capabilities if x.id == requirement.id + ).dry_run, + order=order, + ) + ) + core = { + "source": plan.plan_digest, + "from": plan.from_revision, + "to": plan.to_revision, + "operations": [item.model_dump(mode="json") for item in operations], + } + fingerprint = digest(core) + return ExecutablePlan( + id="execution-plan-" + fingerprint[:16], + source_plan_id=plan.plan_digest, + world_id=plan.world_id, + from_revision=plan.from_revision, + desired_revision=plan.to_revision, + fingerprint=fingerprint, + operations=operations, + ) diff --git a/src/netsovereign/runtime/executor.py b/src/netsovereign/runtime/executor.py new file mode 100644 index 0000000..5de576b --- /dev/null +++ b/src/netsovereign/runtime/executor.py @@ -0,0 +1,296 @@ +"""Sequential runtime executor for provider-bound plans.""" + +from __future__ import annotations + +import asyncio +from collections.abc import Awaitable, Callable +from datetime import UTC, datetime +from typing import Any + +from ..canonical import digest +from ..providers.contracts import FailureClass, ObservationResult, ProviderContext, ProviderResult +from ..providers.registry import ProviderRegistry +from .models import ( + Attempt, + Evidence, + EvidenceKind, + ExecutableOperation, + ExecutablePlan, + ExecutionReport, + ExecutionRun, + FailurePosture, + OperationState, + _redact, +) +from .repository import InMemoryExecutionRepository + +Clock = Callable[[], datetime] +Delay = Callable[[float], Awaitable[None]] + + +class RuntimeExecutor: + def __init__( + self, + registry: ProviderRegistry, + repository: InMemoryExecutionRepository, + *, + clock: Clock | None = None, + delay: Delay | None = None, + ) -> None: + self.registry = registry + self.repository = repository + self.clock = clock or (lambda: datetime.now(UTC)) + self.delay = delay or asyncio.sleep + + def _evidence( + self, + run: ExecutionRun, + operation: ExecutableOperation, + kind: EvidenceKind, + payload: dict[str, Any], + ) -> None: + stable = {"run": run.id, "operation": operation.id, "kind": kind, "n": len(run.evidence)} + run.evidence.append( + Evidence( + id="evidence-" + digest(stable)[:16], + run_id=run.id, + operation_id=operation.id, + provider_id=operation.provider_id, + capability_id=operation.capability.id, + authority_id=operation.authority_id, + mandate_id=operation.mandate_id, + target_resource=operation.target, + desired_revision=run.desired_revision, + kind=kind, + timestamp=self.clock(), + payload=_redact(payload), + ) + ) + + async def execute( + self, plan: ExecutablePlan, *, dry_run: bool = False, run_id: str | None = None + ) -> ExecutionReport: + if ( + digest( + { + "source": plan.source_plan_id, + "from": plan.from_revision, + "to": plan.desired_revision, + "operations": [item.model_dump(mode="json") for item in plan.operations], + } + ) + != plan.fingerprint + ): + raise ValueError("execution plan integrity check failed") + run = ( + self.repository.get(run_id) + if run_id + else self.repository.create(plan, dry_run, self.clock()) + ) + if run.plan_fingerprint != plan.fingerprint or run.dry_run != dry_run: + raise ValueError("incompatible plan or execution mode for resume") + completed: list[ExecutableOperation] = [] + failed = False + operations_by_step = {item.source_step_id: item for item in plan.operations} + for operation in plan.operations: + record = run.operations[operation.id] + if record.state == OperationState.VERIFIED: + completed.append(operation) + continue + unmet_dependencies = [ + dependency + for dependency in operation.depends_on + if run.operations[operations_by_step[dependency].id].state + != OperationState.VERIFIED + ] + if unmet_dependencies: + if record.state == OperationState.PENDING: + record.transition( + OperationState.SKIPPED, + f"dependencies not verified: {', '.join(sorted(unmet_dependencies))}", + self.clock(), + ) + continue + if failed and operation.failure_posture != FailurePosture.CONTINUE_INDEPENDENT: + if record.state == OperationState.PENDING: + record.transition( + OperationState.SKIPPED, "prior operation failed", self.clock() + ) + continue + provider = self.registry.resolve(operation.capability, operation.provider_id) + context = ProviderContext( + run_id=run.id, + operation_id=operation.id, + idempotency_key=operation.idempotency_key, + dry_run=dry_run, + ) + # An interrupted apply or observation is resolved by observing before replay. + if record.state in {OperationState.RUNNING, OperationState.OBSERVING}: + observed = await provider.observe(operation, context) + record.observation = observed + self._evidence( + run, operation, EvidenceKind.OBSERVATION, observed.model_dump(mode="json") + ) + if observed.matches_expected: + if record.state == OperationState.RUNNING: + record.transition( + OperationState.SUCCEEDED, + "resume observation found applied state", + self.clock(), + ) + record.transition( + OperationState.OBSERVING, "verify resumed operation", self.clock() + ) + record.transition( + OperationState.VERIFIED, "expected state observed", self.clock() + ) + completed.append(operation) + continue + if record.state == OperationState.OBSERVING: + record.failure = FailureClass.OBSERVATION_MISMATCH + record.transition( + OperationState.FAILED, + "resumed observation did not converge", + self.clock(), + ) + failed = True + continue + if record.state != OperationState.PENDING: + raise ValueError(f"operation {operation.id} cannot resume from {record.state}") + validation = await provider.validate(operation, context) + self._evidence( + run, operation, EvidenceKind.VALIDATION, validation.model_dump(mode="json") + ) + if not validation.valid: + record.failure = validation.failure + record.transition( + OperationState.FAILED, validation.message or "validation failed", self.clock() + ) + self._evidence( + run, + operation, + EvidenceKind.FAILURE, + {"failure": validation.failure, "message": validation.message}, + ) + failed = True + continue + record.transition(OperationState.VALIDATED, "provider validation passed", self.clock()) + record.transition(OperationState.READY, "dependencies satisfied", self.clock()) + if dry_run: + if not operation.dry_run_compatible: + record.transition( + OperationState.FAILED, "provider does not support dry run", self.clock() + ) + failed = True + continue + record.simulated = True + record.transition( + OperationState.SUCCEEDED, "predicted without mutation", self.clock() + ) + record.transition(OperationState.OBSERVING, "simulation conformance", self.clock()) + record.observation = ObservationResult( + observed=False, + matches_expected=True, + state={"predicted_action": validation.predicted_action}, + ) + record.transition(OperationState.VERIFIED, "simulation completed", self.clock()) + self._evidence( + run, + operation, + EvidenceKind.SIMULATION, + {"predicted_action": validation.predicted_action}, + ) + completed.append(operation) + continue + result: ProviderResult | None = None + for number in range(1, operation.retry_policy.maximum_attempts + 1): + record.transition(OperationState.RUNNING, f"attempt {number}", self.clock()) + attempt = Attempt(number=number, started_at=self.clock()) + record.attempts.append(attempt) + result = await provider.apply(operation, context) + attempt.completed_at = self.clock() + attempt.result = result + record.provider_result = result + self._evidence( + run, operation, EvidenceKind.APPLICATION, result.model_dump(mode="json") + ) + if result.success: + record.transition( + OperationState.SUCCEEDED, "provider apply succeeded", self.clock() + ) + break + record.failure = result.failure + if ( + number < operation.retry_policy.maximum_attempts + and result.retryable + and result.failure in operation.retry_policy.retryable + ): + record.transition( + OperationState.RETRY_WAIT, "retry policy permits retry", self.clock() + ) + await self.delay(operation.retry_policy.delay_seconds) + else: + record.transition( + OperationState.FAILED, + result.message or "provider apply failed", + self.clock(), + ) + failed = True + break + if not result or not result.success: + continue + record.transition(OperationState.OBSERVING, "observe provider state", self.clock()) + observation = await provider.observe(operation, context) + record.observation = observation + self._evidence( + run, operation, EvidenceKind.OBSERVATION, observation.model_dump(mode="json") + ) + if observation.matches_expected: + record.transition(OperationState.VERIFIED, "expected state observed", self.clock()) + self._evidence(run, operation, EvidenceKind.CONFORMANCE, {"converged": True}) + completed.append(operation) + else: + record.failure = FailureClass.OBSERVATION_MISMATCH + record.transition( + OperationState.FAILED, "applied state did not converge", self.clock() + ) + failed = True + if ( + not dry_run + and failed + and any( + item.failure_posture == FailurePosture.COMPENSATE_ALL for item in plan.operations + ) + ): + for operation in reversed(completed): + record = run.operations[operation.id] + if record.simulated: + continue + record.transition( + OperationState.COMPENSATING, "reverse-order compensation", self.clock() + ) + provider = self.registry.get(operation.provider_id) + result = await provider.compensate( + operation, + ProviderContext( + run_id=run.id, + operation_id=operation.id, + idempotency_key=operation.idempotency_key, + ), + ) + self._evidence( + run, operation, EvidenceKind.COMPENSATION, result.model_dump(mode="json") + ) + record.transition( + OperationState.COMPENSATED if result.success else OperationState.FAILED, + "compensation completed" if result.success else "compensation failed", + self.clock(), + ) + if not result.success: + record.failure = FailureClass.COMPENSATION + break + return ExecutionReport( + run=run, + status=run.status, + explanation=f"realised revision {run.desired_revision}: {run.status}", + ) diff --git a/src/netsovereign/runtime/models.py b/src/netsovereign/runtime/models.py new file mode 100644 index 0000000..361d808 --- /dev/null +++ b/src/netsovereign/runtime/models.py @@ -0,0 +1,237 @@ +"""Serializable execution plans, state, evidence, and reports.""" + +from __future__ import annotations + +from datetime import datetime +from enum import StrEnum +from typing import Any + +from pydantic import Field + +from ..base import DomainModel +from ..providers.contracts import ( + CapabilityRequirement, + FailureClass, + ObservationResult, + ProviderResult, +) + + +class FailurePosture(StrEnum): + STOP = "stop" + COMPENSATE_ALL = "compensate_all" + CONTINUE_INDEPENDENT = "continue_independent" + CLEANUP = "best_effort_cleanup" + + +class OperationState(StrEnum): + PENDING = "pending" + VALIDATED = "validated" + READY = "ready" + RUNNING = "running" + SUCCEEDED = "succeeded" + OBSERVING = "observing" + VERIFIED = "verified" + RETRY_WAIT = "retry_wait" + COMPENSATING = "compensating" + COMPENSATED = "compensated" + ROLLING_BACK = "rolling_back" + ROLLED_BACK = "rolled_back" + FAILED = "failed" + CANCELLED = "cancelled" + SKIPPED = "skipped" + + +TERMINAL_STATES = { + OperationState.VERIFIED, + OperationState.COMPENSATED, + OperationState.ROLLED_BACK, + OperationState.FAILED, + OperationState.CANCELLED, + OperationState.SKIPPED, +} +TRANSITIONS: dict[OperationState, set[OperationState]] = { + OperationState.PENDING: { + OperationState.VALIDATED, + OperationState.FAILED, + OperationState.SKIPPED, + }, + OperationState.VALIDATED: {OperationState.READY, OperationState.FAILED}, + OperationState.READY: { + OperationState.RUNNING, + OperationState.SUCCEEDED, + OperationState.FAILED, + }, + OperationState.RUNNING: { + OperationState.SUCCEEDED, + OperationState.RETRY_WAIT, + OperationState.FAILED, + }, + OperationState.SUCCEEDED: {OperationState.OBSERVING, OperationState.COMPENSATING}, + OperationState.OBSERVING: {OperationState.VERIFIED, OperationState.FAILED}, + OperationState.RETRY_WAIT: {OperationState.RUNNING, OperationState.FAILED}, + OperationState.VERIFIED: {OperationState.COMPENSATING, OperationState.ROLLING_BACK}, + OperationState.COMPENSATING: {OperationState.COMPENSATED, OperationState.FAILED}, + OperationState.ROLLING_BACK: {OperationState.ROLLED_BACK, OperationState.FAILED}, +} + + +class PlanStatus(StrEnum): + PENDING = "pending" + RUNNING = "running" + SUCCEEDED = "succeeded" + PARTIAL = "partially_succeeded" + FAILED = "failed" + ROLLED_BACK = "rolled_back" + CANCELLED = "cancelled" + + +class RetryPolicy(DomainModel): + maximum_attempts: int = Field(default=1, ge=1) + delay_seconds: float = Field(default=0, ge=0) + retryable: set[FailureClass] = Field( + default_factory=lambda: {FailureClass.TRANSIENT, FailureClass.TIMEOUT} + ) + + +class ExecutableOperation(DomainModel): + id: str + source_step_id: str + authority_id: str | None + mandate_id: str | None + target: str + operation_type: str + capability: CapabilityRequirement + provider_id: str + provider_binding_id: str + depends_on: list[str] + preconditions: list[dict[str, Any]] + expected: Any + idempotency_key: str + retry_policy: RetryPolicy + failure_posture: FailurePosture + dry_run_compatible: bool + order: int + + +class ExecutablePlan(DomainModel): + id: str + source_plan_id: str + world_id: str + from_revision: str + desired_revision: str + fingerprint: str + operations: list[ExecutableOperation] + + +class Transition(DomainModel): + from_state: OperationState + to_state: OperationState + at: datetime + reason: str + + +class Attempt(DomainModel): + number: int + started_at: datetime + completed_at: datetime | None = None + result: ProviderResult | None = None + + +class EvidenceKind(StrEnum): + VALIDATION = "provider_validation" + APPLICATION = "provider_application" + OBSERVATION = "provider_observation" + CONFORMANCE = "conformance" + COMPENSATION = "compensation" + FAILURE = "execution_failure" + SIMULATION = "simulation" + + +_SECRET_KEYS = {"password", "secret", "token", "private_key", "credential"} + + +def _redact(value: Any) -> Any: + if isinstance(value, dict): + return { + key: "[REDACTED]" + if any(word in key.lower() for word in _SECRET_KEYS) + else _redact(item) + for key, item in value.items() + } + if isinstance(value, list): + return [_redact(item) for item in value] + return value + + +class Evidence(DomainModel): + id: str + run_id: str + operation_id: str + provider_id: str + capability_id: str + authority_id: str | None + mandate_id: str | None + target_resource: str + desired_revision: str + kind: EvidenceKind + timestamp: datetime + payload: dict[str, Any] + sensitivity: str = "redacted" + + +class OperationExecution(DomainModel): + operation_id: str + state: OperationState = OperationState.PENDING + attempts: list[Attempt] = Field(default_factory=list) + transitions: list[Transition] = Field(default_factory=list) + provider_result: ProviderResult | None = None + observation: ObservationResult | None = None + failure: FailureClass | None = None + simulated: bool = False + + def transition(self, target: OperationState, reason: str, now: datetime) -> None: + if target not in TRANSITIONS.get(self.state, set()): + raise ValueError(f"invalid transition {self.state} -> {target}") + self.transitions.append( + Transition(from_state=self.state, to_state=target, at=now, reason=reason) + ) + self.state = target + + +class ExecutionRun(DomainModel): + id: str + plan_id: str + plan_fingerprint: str + desired_revision: str + dry_run: bool + created_at: datetime + operations: dict[str, OperationExecution] + evidence: list[Evidence] = Field(default_factory=list) + + @property + def status(self) -> PlanStatus: + states = {item.state for item in self.operations.values()} + if not states: + return PlanStatus.SUCCEEDED + if states and states <= {OperationState.VERIFIED}: + return PlanStatus.SUCCEEDED + if states and states <= {OperationState.COMPENSATED}: + return PlanStatus.ROLLED_BACK + if OperationState.CANCELLED in states: + return PlanStatus.CANCELLED + if OperationState.FAILED in states: + return ( + PlanStatus.PARTIAL + if states & {OperationState.VERIFIED, OperationState.COMPENSATED} + else PlanStatus.FAILED + ) + if states == {OperationState.PENDING}: + return PlanStatus.PENDING + return PlanStatus.RUNNING + + +class ExecutionReport(DomainModel): + run: ExecutionRun + status: PlanStatus + explanation: str diff --git a/src/netsovereign/runtime/repository.py b/src/netsovereign/runtime/repository.py new file mode 100644 index 0000000..f6dede1 --- /dev/null +++ b/src/netsovereign/runtime/repository.py @@ -0,0 +1,40 @@ +"""Execution-state repository contracts and the supported in-memory implementation.""" + +from __future__ import annotations + +from datetime import datetime +from typing import Protocol +from uuid import uuid4 + +from .models import ExecutablePlan, ExecutionRun, OperationExecution, OperationState + + +class ExecutionRepository(Protocol): + def create(self, plan: ExecutablePlan, dry_run: bool, now: datetime) -> ExecutionRun: ... + def get(self, run_id: str) -> ExecutionRun: ... + + +class InMemoryExecutionRepository: + def __init__(self) -> None: + self.runs: dict[str, ExecutionRun] = {} + + def create(self, plan: ExecutablePlan, dry_run: bool, now: datetime) -> ExecutionRun: + run = ExecutionRun( + id=str(uuid4()), + plan_id=plan.id, + plan_fingerprint=plan.fingerprint, + desired_revision=plan.desired_revision, + dry_run=dry_run, + created_at=now, + operations={ + item.id: OperationExecution(operation_id=item.id) for item in plan.operations + }, + ) + self.runs[run.id] = run + return run + + def get(self, run_id: str) -> ExecutionRun: + return self.runs[run_id] + + def list_by_status(self, run_id: str, status: OperationState) -> list[OperationExecution]: + return [item for item in self.get(run_id).operations.values() if item.state == status] diff --git a/tests/test_runtime.py b/tests/test_runtime.py new file mode 100644 index 0000000..7b5eb74 --- /dev/null +++ b/tests/test_runtime.py @@ -0,0 +1,294 @@ +import asyncio +from copy import deepcopy +from datetime import UTC, datetime +from pathlib import Path + +import pytest +import yaml + +from netsovereign.planning import admit_change, build_plan +from netsovereign.providers.contracts import CapabilityRequirement, FailureClass +from netsovereign.providers.fake import FakeProvider +from netsovereign.providers.registry import ProviderRegistry, ProviderResolutionError +from netsovereign.runtime import ( + FailurePosture, + InMemoryExecutionRepository, + OperationExecution, + OperationState, + RetryPolicy, + RuntimeExecutor, + compile_plan, +) +from netsovereign.specification import WorldSpec + +ROOT = Path(__file__).parents[1] + + +def admitted_plan(two=False): + current_data = yaml.safe_load((ROOT / "examples/minimal/world.yaml").read_text()) + proposed_data = deepcopy(current_data) + proposed_data["world"]["revision"] = "2" + proposed_data["providerBindings"][0]["provider"] = "fake" + current = WorldSpec.model_validate(current_data) + proposed = WorldSpec.model_validate(proposed_data) + plan = build_plan(admit_change(current, proposed)) + if two: + second = plan.steps[0].model_copy( + update={"id": plan.steps[0].id + "-second", "target": "resources/second"} + ) + plan.steps.append(second) + return current, proposed, plan + + +def setup(provider=None): + registry = ProviderRegistry() + registry.register(provider or FakeProvider()) + return registry + + +def run(awaitable): + return asyncio.run(awaitable) + + +def test_registry_contract_and_resolution_errors(): + provider = FakeProvider() + registry = setup(provider) + assert registry.resolve(CapabilityRequirement(id="resource.manage")).describe().id == "fake" + assert registry.resolve(CapabilityRequirement(id="resource.manage"), "fake") is provider + with pytest.raises(ProviderResolutionError, match="registered"): + registry.register(provider) + with pytest.raises(ProviderResolutionError) as absent: + registry.resolve(CapabilityRequirement(id="resource.manage"), "absent") + assert absent.value.code == "requested_provider_absent" + with pytest.raises(ProviderResolutionError) as missing: + registry.resolve(CapabilityRequirement(id="missing")) + assert missing.value.code == "no_provider_supports_capability" + with pytest.raises(ProviderResolutionError) as incompatible: + registry.resolve(CapabilityRequirement(id="resource.manage", version="2.0")) + assert incompatible.value.code == "capability_version_incompatible" + registry.register(FakeProvider("other")) + with pytest.raises(ProviderResolutionError) as ambiguous: + registry.resolve(CapabilityRequirement(id="resource.manage")) + assert ambiguous.value.code == "ambiguous_provider" + provider.available = False + with pytest.raises(ProviderResolutionError) as unavailable: + setup(provider).resolve(CapabilityRequirement(id="resource.manage")) + assert unavailable.value.code == "provider_unavailable" + + +def test_compilation_is_stable_preserves_provenance_and_does_not_mutate(): + _, _, plan = admitted_plan() + original = plan.model_dump(mode="json") + first = compile_plan(plan, setup()) + second = compile_plan(plan, setup()) + assert first == second and first.fingerprint == second.fingerprint + assert plan.model_dump(mode="json") == original + operation = first.operations[0] + step = plan.steps[0] + assert (operation.authority_id, operation.mandate_id, operation.target) == ( + step.authority_id, + step.mandate_id, + step.target, + ) + broken = plan.model_copy(deep=True) + broken.steps[0].depends_on = [broken.steps[0].id] + with pytest.raises(ValueError, match="dependency_cycle"): + compile_plan(broken, setup()) + + +def test_state_machine_fails_closed_and_records_time(): + now = datetime.now(UTC) + record = OperationExecution(operation_id="one") + record.transition(OperationState.VALIDATED, "ok", now) + assert record.transitions[0].at == now + with pytest.raises(ValueError, match="invalid transition"): + record.transition(OperationState.VERIFIED, "skip", now) + + +def test_apply_observe_evidence_and_idempotent_reexecution(): + _, _, plan = admitted_plan() + provider = FakeProvider() + registry = setup(provider) + executable = compile_plan(plan, registry) + repository = InMemoryExecutionRepository() + executor = RuntimeExecutor(registry, repository) + report = run(executor.execute(executable)) + assert report.status == "succeeded" + operation = report.run.operations[executable.operations[0].id] + assert operation.provider_result and operation.observation and operation.state == "verified" + assert {item.kind for item in report.run.evidence} >= { + "provider_validation", + "provider_application", + "provider_observation", + "conformance", + } + count = len([call for call in provider.call_history if call[0] == "apply"]) + resumed = run(executor.execute(executable, run_id=report.run.id)) + assert resumed.status == "succeeded" + assert len([call for call in provider.call_history if call[0] == "apply"]) == count + evidence = report.run.evidence[0] + assert {"authority_id", "mandate_id", "target_resource", "desired_revision"} <= set( + evidence.model_dump() + ) + + +def test_dry_run_validates_without_mutation_and_reports_prediction(): + _, _, plan = admitted_plan() + provider = FakeProvider() + registry = setup(provider) + report = run( + RuntimeExecutor(registry, InMemoryExecutionRepository()).execute( + compile_plan(plan, registry), dry_run=True + ) + ) + assert report.status == "succeeded" and provider.state == {} + assert not any(call[0] == "apply" for call in provider.call_history) + assert any(item.kind == "simulation" for item in report.run.evidence) + + +def test_validation_observation_and_retry_failure_semantics(): + _, _, plan = admitted_plan() + provider = FakeProvider() + registry = setup(provider) + executable = compile_plan(plan, registry, retry_policy=RetryPolicy(maximum_attempts=2)) + op = executable.operations[0] + provider.inject_failure(op.id, FailureClass.TRANSIENT) + report = run( + RuntimeExecutor( + registry, InMemoryExecutionRepository(), delay=lambda _: asyncio.sleep(0) + ).execute(executable) + ) + assert report.status == "succeeded" and len(report.run.operations[op.id].attempts) == 2 + provider.observation_mismatches.add(op.id) + second = run(RuntimeExecutor(registry, InMemoryExecutionRepository()).execute(executable)) + assert second.status == "failed" + assert second.run.operations[op.id].failure == "observation_mismatch" + + +def test_partial_failure_compensates_completed_operations_in_reverse_order(): + _, _, plan = admitted_plan(two=True) + provider = FakeProvider() + registry = setup(provider) + executable = compile_plan(plan, registry, failure_posture=FailurePosture.COMPENSATE_ALL) + assert len(executable.operations) >= 2 + provider.inject_failure(executable.operations[-1].id, FailureClass.PERMANENT) + report = run(RuntimeExecutor(registry, InMemoryExecutionRepository()).execute(executable)) + compensated = [item for item in report.run.operations.values() if item.state == "compensated"] + assert compensated and not provider.state + calls = [op_id for action, op_id in provider.call_history if action == "compensate"] + assert calls == [item.id for item in reversed(executable.operations[:-1])] + + +def test_interruption_after_apply_resumes_by_observation_without_duplicate(): + _, _, plan = admitted_plan() + provider = FakeProvider() + registry = setup(provider) + executable = compile_plan(plan, registry) + repository = InMemoryExecutionRepository() + run_record = repository.create(executable, False, datetime.now(UTC)) + operation = executable.operations[0] + record = run_record.operations[operation.id] + record.transition(OperationState.VALIDATED, "validated", datetime.now(UTC)) + record.transition(OperationState.READY, "ready", datetime.now(UTC)) + record.transition(OperationState.RUNNING, "interrupted apply", datetime.now(UTC)) + context = __import__( + "netsovereign.providers.contracts", fromlist=["ProviderContext"] + ).ProviderContext( + run_id=run_record.id, operation_id=operation.id, idempotency_key=operation.idempotency_key + ) + run(provider.apply(operation, context)) + report = run(RuntimeExecutor(registry, repository).execute(executable, run_id=run_record.id)) + assert report.status == "succeeded" + assert len([call for call in provider.call_history if call[0] == "apply"]) == 1 + changed = executable.model_copy(update={"fingerprint": "changed"}) + with pytest.raises(ValueError): + run(RuntimeExecutor(registry, repository).execute(changed, run_id=run_record.id)) + + +def test_dry_run_failure_never_compensates_simulated_operations(): + _, _, plan = admitted_plan(two=True) + provider = FakeProvider() + registry = setup(provider) + executable = compile_plan(plan, registry, failure_posture=FailurePosture.COMPENSATE_ALL) + executable.operations[-1].dry_run_compatible = False + # Restore integrity after changing an explicit compilation input for this scenario. + from netsovereign.canonical import digest + + executable.fingerprint = digest( + { + "source": executable.source_plan_id, + "from": executable.from_revision, + "to": executable.desired_revision, + "operations": [item.model_dump(mode="json") for item in executable.operations], + } + ) + report = run( + RuntimeExecutor(registry, InMemoryExecutionRepository()).execute(executable, dry_run=True) + ) + assert report.status == "partially_succeeded" + assert not any( + action in {"apply", "compensate", "delete"} for action, _ in provider.call_history + ) + + +def test_continue_independent_skips_failed_dependants_but_runs_independent_work(): + _, _, plan = admitted_plan(two=True) + third = plan.steps[0].model_copy( + update={ + "id": plan.steps[0].id + "-dependent", + "target": "resources/dependent", + "depends_on": [plan.steps[0].id], + } + ) + plan.steps.append(third) + provider = FakeProvider() + registry = setup(provider) + executable = compile_plan(plan, registry, failure_posture=FailurePosture.CONTINUE_INDEPENDENT) + first = next(item for item in executable.operations if item.source_step_id == plan.steps[0].id) + independent = next( + item for item in executable.operations if item.source_step_id == plan.steps[1].id + ) + dependent = next(item for item in executable.operations if item.source_step_id == third.id) + provider.inject_failure(first.id, FailureClass.PERMANENT) + report = run(RuntimeExecutor(registry, InMemoryExecutionRepository()).execute(executable)) + assert report.run.operations[first.id].state == "failed" + assert report.run.operations[independent.id].state == "verified" + assert report.run.operations[dependent.id].state == "skipped" + assert ("apply", dependent.id) not in provider.call_history + + +def test_resume_while_observing_repeats_observation_only(): + _, _, plan = admitted_plan() + provider = FakeProvider() + registry = setup(provider) + executable = compile_plan(plan, registry) + repository = InMemoryExecutionRepository() + run_record = repository.create(executable, False, datetime.now(UTC)) + operation = executable.operations[0] + record = run_record.operations[operation.id] + record.transition(OperationState.VALIDATED, "validated", datetime.now(UTC)) + record.transition(OperationState.READY, "ready", datetime.now(UTC)) + record.transition(OperationState.RUNNING, "apply", datetime.now(UTC)) + context = __import__( + "netsovereign.providers.contracts", fromlist=["ProviderContext"] + ).ProviderContext( + run_id=run_record.id, + operation_id=operation.id, + idempotency_key=operation.idempotency_key, + ) + result = run(provider.apply(operation, context)) + record.provider_result = result + record.transition(OperationState.SUCCEEDED, "applied", datetime.now(UTC)) + record.transition(OperationState.OBSERVING, "interrupted", datetime.now(UTC)) + report = run(RuntimeExecutor(registry, repository).execute(executable, run_id=run_record.id)) + assert report.status == "succeeded" + assert len([call for call in provider.call_history if call[0] == "apply"]) == 1 + + +def test_empty_plan_is_terminally_successful(): + _, _, plan = admitted_plan() + plan.steps = [] + registry = setup() + executable = compile_plan(plan, registry) + report = run(RuntimeExecutor(registry, InMemoryExecutionRepository()).execute(executable)) + assert report.status == "succeeded" diff --git a/uv.lock b/uv.lock index ae46510..038e907 100644 --- a/uv.lock +++ b/uv.lock @@ -236,7 +236,7 @@ wheels = [ [[package]] name = "netsovereign" -version = "0.2.0" +version = "0.3.0" source = { editable = "." } dependencies = [ { name = "pydantic" },