flayer.core.lifecycle

Module source

flayer.core.lifecycle

Owned deployment orchestration with durable progress and fail-closed recovery.

MAX_PLAN_RESOURCES module-attribute

MAX_PLAN_RESOURCES = 128

LifecycleError

Bases: ContractError

An operation is blocked or requires explicit interrupted-operation recovery.

DeploymentPlan dataclass

DeploymentPlan(identity: StackIdentity, resources: tuple[ResourceSpec, ...])

Immutable desired resources with a deterministic dependency order.

identity instance-attribute

identity: StackIdentity

resources instance-attribute

resources: tuple[ResourceSpec, ...]

OrderedResources

OrderedResources() -> tuple[ResourceSpec, ...]

Return a stable topological ordering or reject cyclic dependencies.

源代码位于： installed/flayer/core/lifecycle.py

def OrderedResources(self) -> tuple[ResourceSpec, ...]:
    """Return a stable topological ordering or reject cyclic dependencies."""

    remaining = list(self.resources)
    result: list[ResourceSpec] = []
    complete: set[str] = set()

    while remaining:
        ready = [item for item in remaining if set(item.dependencies) <= complete]

        if not ready:
            raise LifecycleError("Plan contains a dependency cycle")

        for resource in ready:
            result.append(resource)
            complete.add(resource.logical_id)
            remaining.remove(resource)

    return tuple(result)

Fingerprint

Fingerprint() -> str

Bind recovery to exact intent without persisting desired parameters in state.

源代码位于： installed/flayer/core/lifecycle.py

def Fingerprint(self) -> str:
    """Bind recovery to exact intent without persisting desired parameters in state."""

    payload = json.dumps(asdict(self), sort_keys=True, separators=(",", ":"))

    return hashlib.sha256(payload.encode("utf-8")).hexdigest()

LifecycleObservation dataclass

LifecycleObservation(logical_id: str, status: str, resource_id: str | None = None, provider_status: str | None = None)

Sanitized local ownership and remote existence summary for one logical resource.

logical_id instance-attribute

logical_id: str

status instance-attribute

status: str

resource_id class-attribute instance-attribute

resource_id: str | None = None

provider_status class-attribute instance-attribute

provider_status: str | None = None

LifecycleReport dataclass

LifecycleReport(action: str, status: str, resources: tuple[LifecycleObservation, ...] = (), recovery_required: bool = False)

Truthful operation outcome without raw provider commands, errors, or parameters.

action instance-attribute

action: str

status instance-attribute

status: str

resources class-attribute instance-attribute

resources: tuple[LifecycleObservation, ...] = ()

recovery_required class-attribute instance-attribute

recovery_required: bool = False

ExitCode

ExitCode() -> int

Report nonzero whenever an operation failed or is awaiting recovery.

源代码位于： installed/flayer/core/lifecycle.py

def ExitCode(self) -> int:
    """Report nonzero whenever an operation failed or is awaiting recovery."""

    return 0 if self.status == "complete" else 1

LifecycleEngine

LifecycleEngine(plan: DeploymentPlan, provider: LifecycleProvider, state_path: str | Path)

Compose a scoped provider with durable owned-state lifecycle semantics.

Reject a provider or path that cannot represent the exact desired scope.

源代码位于： installed/flayer/core/lifecycle.py

def __init__(
    self, plan: DeploymentPlan, provider: LifecycleProvider, state_path: str | Path
) -> None:
    """Reject a provider or path that cannot represent the exact desired scope."""

    if not isinstance(plan, DeploymentPlan):
        raise LifecycleError("Lifecycle requires a validated DeploymentPlan")

    if (
        provider.Identity.provider_id != plan.identity.provider
        or provider.Identity.scope_id != plan.identity.scope_id
    ):
        raise LifecycleError("Provider identity must match the deployment plan")

    for spec in plan.resources:
        provider.ValidateSpec(spec)

    self.plan = plan
    self.provider = provider
    self.state_path = Path(state_path)
    _ValidatePath(self.state_path)
    self._specs = {resource.logical_id: resource for resource in plan.resources}

plan instance-attribute

plan = plan

provider instance-attribute

provider = provider

state_path instance-attribute

state_path = Path(state_path)

Status

Status() -> LifecycleReport

Read exact owned remote existence without changing state or creating resources.

源代码位于： installed/flayer/core/lifecycle.py

def Status(self) -> LifecycleReport:
    """Read exact owned remote existence without changing state or creating resources."""

    with _OperationLock(self.state_path):
        state = self._State()
        journal = _LoadJournal(self.state_path, self.plan)
        resources: list[LifecycleObservation] = []

        for locator in state.resources:
            observed = self._RequireLocator(locator)
            resources.append(LifecycleObservation(
                locator.logical_id, "present" if observed is not None else "missing",
                locator.resource_id, observed.status if observed is not None else None,
            ))

        missing = any(resource.status == "missing" for resource in resources)
        resources.extend(LifecycleObservation(spec.logical_id, "not-recorded")
                         for spec in self.plan.resources
                         if spec.logical_id not in {item.logical_id for item in state.resources})
        complete = not missing and len(state.resources) == len(self.plan.resources)

        return LifecycleReport(
            "status", "complete" if complete and journal is None else "incomplete",
            tuple(resources), journal is not None,
        )

Create

Create() -> LifecycleReport

Create missing owned resources and roll back only this operation's creations.

源代码位于： installed/flayer/core/lifecycle.py

def Create(self) -> LifecycleReport:
    """Create missing owned resources and roll back only this operation's creations."""

    with _OperationLock(self.state_path):
        state = self._State()

        if _LoadJournal(self.state_path, self.plan) is not None:
            raise LifecycleError("An interrupted operation requires explicit recovery")

        preexisting: list[ResourceState] = []
        observations: dict[str, ProviderResource] = {}
        recorded = {resource.logical_id: resource for resource in state.resources}

        for spec in self.plan.OrderedResources():
            observed = self._Observe(spec)

            if spec.logical_id in recorded and (
                observed is None or observed.reference.resource_id != recorded[spec.logical_id].resource_id
            ):
                raise LifecycleError("Recorded resource is missing or replaced")

            if observed is not None:
                observations[spec.logical_id] = observed
                preexisting.append(self._Locator(spec, observed))

        for spec in self.plan.resources:
            if spec.logical_id in observations and not set(spec.dependencies) <= observations.keys():
                raise LifecycleError("Preexisting resource depends on a missing resource")

        journal = _Journal(
            self.plan.identity, self.plan.Fingerprint(), "create", tuple(preexisting),
            operation_id=uuid4().hex,
        )
        _WriteJournal(self.state_path, journal)
        self._Save(tuple(preexisting))

        return self._Create(journal)

Destroy

Destroy() -> LifecycleReport

Remove only recorded exact-owned resources in reverse dependency order.

源代码位于： installed/flayer/core/lifecycle.py

def Destroy(self) -> LifecycleReport:
    """Remove only recorded exact-owned resources in reverse dependency order."""

    with _OperationLock(self.state_path):
        state = self._State()

        if _LoadJournal(self.state_path, self.plan) is not None:
            raise LifecycleError("An interrupted operation requires explicit recovery")

        journal = _Journal(
            self.plan.identity, self.plan.Fingerprint(), "destroy", state.resources,
            operation_id=uuid4().hex,
        )
        _WriteJournal(self.state_path, journal)

        return self._Destroy(journal)

Recover

Recover(rollback: bool = False) -> LifecycleReport

Explicitly resume durable intent, never recreate an unresolved ambiguous create.

源代码位于： installed/flayer/core/lifecycle.py

def Recover(self, rollback: bool = False) -> LifecycleReport:
    """Explicitly resume durable intent, never recreate an unresolved ambiguous create."""

    with _OperationLock(self.state_path):
        self._State()
        journal = _LoadJournal(self.state_path, self.plan)

        if journal is None:
            raise LifecycleError("No interrupted operation is recorded")

        if journal.action == "destroy":
            if rollback:
                raise LifecycleError("A partial destroy cannot be rolled back")

            return self._Destroy(journal)

        if journal.action == "rollback":
            return self._Rollback(journal)

        if journal.pending is not None:
            if journal.pending in {item.logical_id for item in (*journal.original, *journal.created)}:
                raise LifecycleError("Pending creation cannot become rollback eligible twice")

            spec = self._specs[journal.pending]
            resource = self._Observe(spec)

            if resource is None or dict(resource.labels).get("flayer-operation") != journal.operation_id:
                return self._Report("recover", "uncertain", True)

            locator = self._Locator(spec, resource)
            journal = replace(journal, action="create", created=(*journal.created, locator), pending=None)
            _WriteJournal(self.state_path, journal)

        locators = {item.logical_id: item for item in self._State().resources}
        locators.update((item.logical_id, item) for item in (*journal.original, *journal.created))
        self._Save(tuple(locators.values()))

        if rollback:
            journal = replace(journal, action="rollback", pending=None)
            _WriteJournal(self.state_path, journal)

            return self._Rollback(journal)

        return self._Create(journal)

CoreKind

CoreKind(kind: ResourceKind) -> str

Bridge provider enum spelling to the portable hyphenated state contract.

源代码位于： installed/flayer/core/lifecycle.py

def CoreKind(kind: ResourceKind) -> str:
    """Bridge provider enum spelling to the portable hyphenated state contract."""

    return kind.value.replace("_", "-")

ProviderKind

ProviderKind(kind: str) -> ResourceKind

Accept only an explicitly recognized portable lifecycle resource kind.

源代码位于： installed/flayer/core/lifecycle.py

def ProviderKind(kind: str) -> ResourceKind:
    """Accept only an explicitly recognized portable lifecycle resource kind."""

    try:
        return ResourceKind(kind.replace("-", "_"))

    except ValueError:
        raise LifecycleError("Lifecycle resource kind is unsupported") from None

LoadDeploymentPlan

LoadDeploymentPlan(path: str | Path) -> DeploymentPlan

Load explicit version-one lifecycle TOML without resolving credential material.

源代码位于： installed/flayer/core/lifecycle.py

def LoadDeploymentPlan(path: str | Path) -> DeploymentPlan:
    """Load explicit version-one lifecycle TOML without resolving credential material."""

    try:
        with Path(path).open("rb") as stream:
            data = tomllib.load(stream)

        ValidateFields(
            data,
            frozenset({"schema_version", "identity", "resources"}),
            frozenset({"profile"}),
            "deployment plan",
        )
        ValidateSchemaVersion(data["schema_version"])

        if "profile" in data:
            ValidateName(data["profile"], "profile")

        raw_resources = data["resources"]

        if not isinstance(raw_resources, list):
            raise LifecycleError("Plan resources must be an array of tables")

        resources: list[ResourceSpec] = []

        for item in raw_resources:
            table = RequireTable(item, "resources")
            ValidateFields(
                table,
                frozenset({"logical_id", "kind", "name"}),
                frozenset({"parameters", "dependencies"}),
                "resources",
            )
            parameters = RequireTable(table.get("parameters", {}), "parameters")
            dependencies = table.get("dependencies", [])

            if not isinstance(dependencies, list) or any(
                not isinstance(value, str) for value in dependencies
            ):
                raise LifecycleError("Dependencies must be an array of logical identifiers")

            kind = ValidateName(table["kind"], "resources.kind")
            resources.append(ResourceSpec(
                logical_id=ValidateName(table["logical_id"], "resources.logical_id"),
                kind=ProviderKind(kind),
                name=ValidateName(table["name"], "resources.name"),
                parameters=tuple((key, _FreezeValue(value)) for key, value in parameters.items()),
                dependencies=tuple(dependencies),
            ))

        return DeploymentPlan(ParseIdentity(data["identity"]), tuple(resources))

    except (OSError, ValueError, RecursionError):
        raise LifecycleError("Unable to load a valid lifecycle plan") from None