English fallback

This translation is missing; the canonical English reference follows.

fuzzy1337.executors.contracts

fuzzy1337.executors.contracts

Typed contracts for governed executor providers and local process execution.

CapabilityDescriptor dataclass

One executable capability derived from a provider declaration.

Source code in fuzzy1337/executors/contracts.py

@dataclass(frozen=True, slots=True)
class CapabilityDescriptor:
    """One executable capability derived from a provider declaration."""

    identifier: str
    adapter_id: str
    maximum_impact: ImpactLevel
    required_privileges: tuple[str, ...] = ()

    def __post_init__(self) -> None:
        """Validate the capability declaration."""

        RequireIdentifier(self.identifier, "identifier")
        RequireIdentifier(self.adapter_id, "adapter_id")

        if not isinstance(self.maximum_impact, ImpactLevel):
            raise ValueError("maximum_impact must be an ImpactLevel")

        object.__setattr__(
            self,
            "required_privileges",
            UniqueIdentifiers(self.required_privileges, "required_privileges"),
        )

__post_init__() -> None

Validate the capability declaration.

Source code in fuzzy1337/executors/contracts.py

def __post_init__(self) -> None:
    """Validate the capability declaration."""

    RequireIdentifier(self.identifier, "identifier")
    RequireIdentifier(self.adapter_id, "adapter_id")

    if not isinstance(self.maximum_impact, ImpactLevel):
        raise ValueError("maximum_impact must be an ImpactLevel")

    object.__setattr__(
        self,
        "required_privileges",
        UniqueIdentifiers(self.required_privileges, "required_privileges"),
    )

ExecutionAuthorization dataclass

Upstream scope/policy decision bound to one exact prepared invocation.

Source code in fuzzy1337/executors/contracts.py

@dataclass(frozen=True, slots=True)
class ExecutionAuthorization:
    """Upstream scope/policy decision bound to one exact prepared invocation."""

    authorization_id: str
    scope_reference: str
    policy_reference: str
    invocation_sha256: str
    granted_privileges: tuple[str, ...] = ()

    def __post_init__(self) -> None:
        """Validate the authorization binding."""

        RequireReference(self.authorization_id, "authorization_id")
        RequireReference(self.scope_reference, "scope_reference")
        RequireReference(self.policy_reference, "policy_reference")

        if not _SHA256_PATTERN.fullmatch(self.invocation_sha256):
            raise ValueError("invocation_sha256 must be a lowercase SHA-256 digest")

        object.__setattr__(
            self,
            "granted_privileges",
            UniqueIdentifiers(self.granted_privileges, "granted_privileges"),
        )

__post_init__() -> None

Validate the authorization binding.

Source code in fuzzy1337/executors/contracts.py

def __post_init__(self) -> None:
    """Validate the authorization binding."""

    RequireReference(self.authorization_id, "authorization_id")
    RequireReference(self.scope_reference, "scope_reference")
    RequireReference(self.policy_reference, "policy_reference")

    if not _SHA256_PATTERN.fullmatch(self.invocation_sha256):
        raise ValueError("invocation_sha256 must be a lowercase SHA-256 digest")

    object.__setattr__(
        self,
        "granted_privileges",
        UniqueIdentifiers(self.granted_privileges, "granted_privileges"),
    )

ExecutionResources dataclass

Portable resource bounds enforced by the local executor.

Source code in fuzzy1337/executors/contracts.py

@dataclass(frozen=True, slots=True)
class ExecutionResources:
    """Portable resource bounds enforced by the local executor."""

    max_stdout_bytes: int = 4 * 1024 * 1024
    max_stderr_bytes: int = 4 * 1024 * 1024
    terminate_grace_seconds: float = 1.0

    def __post_init__(self) -> None:
        """Validate portable executor resource bounds."""

        for field_name in ("max_stdout_bytes", "max_stderr_bytes"):
            value = getattr(self, field_name)

            if isinstance(value, bool) or not isinstance(value, int) or value < 1:
                raise ValueError(f"{field_name} must be a positive integer")

        if (
            isinstance(self.terminate_grace_seconds, bool)
            or not isinstance(self.terminate_grace_seconds, (int, float))
            or not math.isfinite(self.terminate_grace_seconds)
            or self.terminate_grace_seconds <= 0
        ):
            raise ValueError("terminate_grace_seconds must be a positive finite number")

__post_init__() -> None

Validate portable executor resource bounds.

Source code in fuzzy1337/executors/contracts.py

def __post_init__(self) -> None:
    """Validate portable executor resource bounds."""

    for field_name in ("max_stdout_bytes", "max_stderr_bytes"):
        value = getattr(self, field_name)

        if isinstance(value, bool) or not isinstance(value, int) or value < 1:
            raise ValueError(f"{field_name} must be a positive integer")

    if (
        isinstance(self.terminate_grace_seconds, bool)
        or not isinstance(self.terminate_grace_seconds, (int, float))
        or not math.isfinite(self.terminate_grace_seconds)
        or self.terminate_grace_seconds <= 0
    ):
        raise ValueError("terminate_grace_seconds must be a positive finite number")

LocalExecutionRequest dataclass

Fully bound request accepted by the local subprocess executor.

Source code in fuzzy1337/executors/contracts.py

@dataclass(frozen=True, slots=True)
class LocalExecutionRequest:
    """Fully bound request accepted by the local subprocess executor."""

    invocation: AdapterInvocation
    capability: CapabilityDescriptor
    authorization: ExecutionAuthorization
    workspace: str = "."
    environment: Mapping[str, str] = field(default_factory=dict)
    resources: ExecutionResources = field(default_factory=ExecutionResources)

    def __post_init__(self) -> None:
        """Validate that the request is internally consistent and authorized."""

        if self.invocation.adapter_id != self.capability.adapter_id:
            raise ValueError("capability adapter_id must match the invocation")

        if self.invocation.request.capability != self.capability.identifier:
            raise ValueError("capability identifier must match the invocation")

        if (
            _IMPACT_ORDER[self.invocation.request.impact]
            > _IMPACT_ORDER[self.capability.maximum_impact]
        ):
            raise ValueError("invocation impact exceeds the capability maximum")

        if self.authorization.invocation_sha256 != InvocationDigest(self.invocation):
            raise ValueError("authorization does not match the invocation")

        if not set(self.capability.required_privileges).issubset(
            self.authorization.granted_privileges
        ):
            raise ValueError("authorization does not grant every required privilege")

        RequireReference(self.workspace, "workspace")
        path = PurePosixPath(self.workspace)

        if "\\" in self.workspace or path.is_absolute() or ".." in path.parts:
            raise ValueError("workspace must be a relative path inside the executor root")

        object.__setattr__(self, "workspace", path.as_posix())
        object.__setattr__(self, "environment", FreezeEnvironment(self.environment))

__post_init__() -> None

Validate that the request is internally consistent and authorized.

Source code in fuzzy1337/executors/contracts.py

def __post_init__(self) -> None:
    """Validate that the request is internally consistent and authorized."""

    if self.invocation.adapter_id != self.capability.adapter_id:
        raise ValueError("capability adapter_id must match the invocation")

    if self.invocation.request.capability != self.capability.identifier:
        raise ValueError("capability identifier must match the invocation")

    if (
        _IMPACT_ORDER[self.invocation.request.impact]
        > _IMPACT_ORDER[self.capability.maximum_impact]
    ):
        raise ValueError("invocation impact exceeds the capability maximum")

    if self.authorization.invocation_sha256 != InvocationDigest(self.invocation):
        raise ValueError("authorization does not match the invocation")

    if not set(self.capability.required_privileges).issubset(
        self.authorization.granted_privileges
    ):
        raise ValueError("authorization does not grant every required privilege")

    RequireReference(self.workspace, "workspace")
    path = PurePosixPath(self.workspace)

    if "\\" in self.workspace or path.is_absolute() or ".." in path.parts:
        raise ValueError("workspace must be a relative path inside the executor root")

    object.__setattr__(self, "workspace", path.as_posix())
    object.__setattr__(self, "environment", FreezeEnvironment(self.environment))

ExecutionEventKind

Bases: StrEnum

Structured progress events emitted by an executor.

Source code in fuzzy1337/executors/contracts.py

class ExecutionEventKind(StrEnum):
    """Structured progress events emitted by an executor."""

    STARTED = "started"
    STDOUT = "stdout"
    STDERR = "stderr"
    COMPLETED = "completed"

ExecutionTermination

Bases: StrEnum

Why the local executor stopped owning the process.

Source code in fuzzy1337/executors/contracts.py

class ExecutionTermination(StrEnum):
    """Why the local executor stopped owning the process."""

    PROCESS_EXIT = "process_exit"
    TIMEOUT = "timeout"
    CANCELLATION = "cancellation"
    OUTPUT_LIMIT = "output_limit"

ExecutionEvent dataclass

One monotonically sequenced executor observation.

Source code in fuzzy1337/executors/contracts.py

@dataclass(frozen=True, slots=True)
class ExecutionEvent:
    """One monotonically sequenced executor observation."""

    sequence: int
    kind: ExecutionEventKind
    data: bytes = b""
    termination: ExecutionTermination | None = None

    def __post_init__(self) -> None:
        """Validate the event payload and lifecycle semantics."""

        if (
            isinstance(self.sequence, bool)
            or not isinstance(self.sequence, int)
            or self.sequence < 0
        ):
            raise ValueError("sequence must be a non-negative integer")

        if not isinstance(self.kind, ExecutionEventKind):
            raise ValueError("kind must be an ExecutionEventKind")

        if not isinstance(self.data, bytes):
            raise ValueError("data must be bytes")

        if self.kind in {ExecutionEventKind.STDOUT, ExecutionEventKind.STDERR}:
            if not self.data or self.termination is not None:
                raise ValueError("stream events require data and no termination")

        elif self.data:
            raise ValueError("lifecycle events must not contain stream data")

        if self.kind is ExecutionEventKind.COMPLETED:
            if not isinstance(self.termination, ExecutionTermination):
                raise ValueError("completed events require a termination reason")

        elif self.termination is not None:
            raise ValueError("only completed events may declare termination")

__post_init__() -> None

Validate the event payload and lifecycle semantics.

Source code in fuzzy1337/executors/contracts.py

def __post_init__(self) -> None:
    """Validate the event payload and lifecycle semantics."""

    if (
        isinstance(self.sequence, bool)
        or not isinstance(self.sequence, int)
        or self.sequence < 0
    ):
        raise ValueError("sequence must be a non-negative integer")

    if not isinstance(self.kind, ExecutionEventKind):
        raise ValueError("kind must be an ExecutionEventKind")

    if not isinstance(self.data, bytes):
        raise ValueError("data must be bytes")

    if self.kind in {ExecutionEventKind.STDOUT, ExecutionEventKind.STDERR}:
        if not self.data or self.termination is not None:
            raise ValueError("stream events require data and no termination")

    elif self.data:
        raise ValueError("lifecycle events must not contain stream data")

    if self.kind is ExecutionEventKind.COMPLETED:
        if not isinstance(self.termination, ExecutionTermination):
            raise ValueError("completed events require a termination reason")

    elif self.termination is not None:
        raise ValueError("only completed events may declare termination")

LocalExecutionResult dataclass

Complete local process outcome plus bounded raw output and event history.

Source code in fuzzy1337/executors/contracts.py

@dataclass(frozen=True, slots=True)
class LocalExecutionResult:
    """Complete local process outcome plus bounded raw output and event history."""

    execution: AdapterExecution
    termination: ExecutionTermination
    stdout: bytes
    stderr: bytes
    events: tuple[ExecutionEvent, ...]

    def __post_init__(self) -> None:
        """Validate the completed result and deterministic event sequence."""

        if not isinstance(self.termination, ExecutionTermination):
            raise ValueError("termination must be an ExecutionTermination")

        if not isinstance(self.stdout, bytes) or not isinstance(self.stderr, bytes):
            raise ValueError("stdout and stderr must be bytes")

        events = tuple(self.events)

        if not events or events[-1].kind is not ExecutionEventKind.COMPLETED:
            raise ValueError("events must end with a completed event")

        if tuple(event.sequence for event in events) != tuple(range(len(events))):
            raise ValueError("events must use contiguous sequence numbers")

        if events[-1].termination is not self.termination:
            raise ValueError("terminal event must match the result termination")

        object.__setattr__(self, "events", events)

__post_init__() -> None

Validate the completed result and deterministic event sequence.

Source code in fuzzy1337/executors/contracts.py

def __post_init__(self) -> None:
    """Validate the completed result and deterministic event sequence."""

    if not isinstance(self.termination, ExecutionTermination):
        raise ValueError("termination must be an ExecutionTermination")

    if not isinstance(self.stdout, bytes) or not isinstance(self.stderr, bytes):
        raise ValueError("stdout and stderr must be bytes")

    events = tuple(self.events)

    if not events or events[-1].kind is not ExecutionEventKind.COMPLETED:
        raise ValueError("events must end with a completed event")

    if tuple(event.sequence for event in events) != tuple(range(len(events))):
        raise ValueError("events must use contiguous sequence numbers")

    if events[-1].termination is not self.termination:
        raise ValueError("terminal event must match the result termination")

    object.__setattr__(self, "events", events)

AdapterRegistration dataclass

Store one lazy adapter registration.

Source code in fuzzy1337/executors/contracts.py

@dataclass(slots=True)
class AdapterRegistration:
    """Store one lazy adapter registration."""

    descriptor: AdapterDescriptor
    loader: AdapterLoader
    instance: ToolAdapter | None = None

LazyAdapterRegistry

Resolve declared providers only when one of their capabilities is requested.

Source code in fuzzy1337/executors/contracts.py

class LazyAdapterRegistry:
    """Resolve declared providers only when one of their capabilities is requested."""

    def __init__(self) -> None:
        """Create an empty registry without loading providers."""

        self._registrations: dict[str, AdapterRegistration] = {}
        self._providers_by_capability: dict[str, list[str]] = {}

    @property
    def Capabilities(self) -> tuple[str, ...]:
        """Return declared capabilities without loading provider implementations.

        Returns:
            Sorted declared capability identifiers without loading providers.
        """

        return tuple(sorted(self._providers_by_capability))

    def Register(self, descriptor: AdapterDescriptor, loader: AdapterLoader) -> None:
        """Register immutable metadata and a lazy provider factory.

        Mutates registry indexes without loading or executing the provider.

        Args:
            descriptor: Immutable declaration to index
            loader: Deferred factory called on first resolution.

        Raises:
            ValueError: The adapter identifier is already registered.
            TypeError: The loader is not callable.
        """

        if descriptor.adapter_id in self._registrations:
            raise ValueError(f"Adapter is already registered: {descriptor.adapter_id}")

        if not callable(loader):
            raise TypeError("loader must be callable")

        self._registrations[descriptor.adapter_id] = AdapterRegistration(descriptor, loader)

        for capability in descriptor.capabilities:
            self._providers_by_capability.setdefault(capability, []).append(descriptor.adapter_id)

    def Resolve(self, capability: str, adapter_id: str | None = None) -> ToolAdapter:
        """Load one provider deterministically and validate its declared identity.

        First resolution invokes the loader and caches its result. Loader exceptions propagate.
        Resolution grants no execution authority.

        Args:
            capability: Declared capability to resolve
            adapter_id: Explicit provider when selection would otherwise be ambiguous.

        Returns:
            Cached or newly loaded provider with a matching descriptor.

        Raises:
            LookupError: The capability is unknown, ambiguous or absent from the adapter.
            TypeError: The loader result does not implement ToolAdapter.
            ValueError: The loaded descriptor differs from the registered metadata.
        """

        providers = self._providers_by_capability.get(capability)

        if not providers:
            raise LookupError(f"Unknown capability: {capability}")

        if adapter_id is None:
            if len(providers) != 1:
                raise LookupError(f"Capability requires an explicit adapter_id: {capability}")

            adapter_id = providers[0]

        elif adapter_id not in providers:
            raise LookupError(f"Adapter {adapter_id} does not provide capability {capability}")

        registration = self._registrations[adapter_id]

        if registration.instance is None:
            candidate = registration.loader()

            if not isinstance(candidate, ToolAdapter):
                raise TypeError("loader must return a ToolAdapter")

            if candidate.Descriptor != registration.descriptor:
                raise ValueError("loaded ToolAdapter descriptor does not match registration")

            registration.instance = candidate

        return registration.instance

Capabilities: tuple[str, ...] property

Return declared capabilities without loading provider implementations.

Returns:

Type	Description
tuple[str, ...]	Sorted declared capability identifiers without loading providers.

__init__() -> None

Create an empty registry without loading providers.

Source code in fuzzy1337/executors/contracts.py

def __init__(self) -> None:
    """Create an empty registry without loading providers."""

    self._registrations: dict[str, AdapterRegistration] = {}
    self._providers_by_capability: dict[str, list[str]] = {}

Register(descriptor: AdapterDescriptor, loader: AdapterLoader) -> None

Register immutable metadata and a lazy provider factory.

Mutates registry indexes without loading or executing the provider.

Parameters:

Name	Type	Description	Default
descriptor	AdapterDescriptor	Immutable declaration to index	required
loader	AdapterLoader	Deferred factory called on first resolution.	required

Raises:

Type	Description
ValueError	The adapter identifier is already registered.
TypeError	The loader is not callable.

Source code in fuzzy1337/executors/contracts.py

def Register(self, descriptor: AdapterDescriptor, loader: AdapterLoader) -> None:
    """Register immutable metadata and a lazy provider factory.

    Mutates registry indexes without loading or executing the provider.

    Args:
        descriptor: Immutable declaration to index
        loader: Deferred factory called on first resolution.

    Raises:
        ValueError: The adapter identifier is already registered.
        TypeError: The loader is not callable.
    """

    if descriptor.adapter_id in self._registrations:
        raise ValueError(f"Adapter is already registered: {descriptor.adapter_id}")

    if not callable(loader):
        raise TypeError("loader must be callable")

    self._registrations[descriptor.adapter_id] = AdapterRegistration(descriptor, loader)

    for capability in descriptor.capabilities:
        self._providers_by_capability.setdefault(capability, []).append(descriptor.adapter_id)

Resolve(capability: str, adapter_id: str | None = None) -> ToolAdapter

Load one provider deterministically and validate its declared identity.

First resolution invokes the loader and caches its result. Loader exceptions propagate. Resolution grants no execution authority.

Parameters:

Name	Type	Description	Default
capability	str	Declared capability to resolve	required
adapter_id	str | None	Explicit provider when selection would otherwise be ambiguous.	None

Returns:

Type	Description
ToolAdapter	Cached or newly loaded provider with a matching descriptor.

Raises:

Type	Description
LookupError	The capability is unknown, ambiguous or absent from the adapter.
TypeError	The loader result does not implement ToolAdapter.
ValueError	The loaded descriptor differs from the registered metadata.

Source code in fuzzy1337/executors/contracts.py

def Resolve(self, capability: str, adapter_id: str | None = None) -> ToolAdapter:
    """Load one provider deterministically and validate its declared identity.

    First resolution invokes the loader and caches its result. Loader exceptions propagate.
    Resolution grants no execution authority.

    Args:
        capability: Declared capability to resolve
        adapter_id: Explicit provider when selection would otherwise be ambiguous.

    Returns:
        Cached or newly loaded provider with a matching descriptor.

    Raises:
        LookupError: The capability is unknown, ambiguous or absent from the adapter.
        TypeError: The loader result does not implement ToolAdapter.
        ValueError: The loaded descriptor differs from the registered metadata.
    """

    providers = self._providers_by_capability.get(capability)

    if not providers:
        raise LookupError(f"Unknown capability: {capability}")

    if adapter_id is None:
        if len(providers) != 1:
            raise LookupError(f"Capability requires an explicit adapter_id: {capability}")

        adapter_id = providers[0]

    elif adapter_id not in providers:
        raise LookupError(f"Adapter {adapter_id} does not provide capability {capability}")

    registration = self._registrations[adapter_id]

    if registration.instance is None:
        candidate = registration.loader()

        if not isinstance(candidate, ToolAdapter):
            raise TypeError("loader must return a ToolAdapter")

        if candidate.Descriptor != registration.descriptor:
            raise ValueError("loaded ToolAdapter descriptor does not match registration")

        registration.instance = candidate

    return registration.instance

RequireIdentifier(value: str, field_name: str) -> None

Require a portable lowercase machine identifier.

Parameters:

Name	Type	Description	Default
value	str	Candidate portable lowercase identifier	required
field_name	str	Field label in validation errors.	required

Raises:

Type	Description
ValueError	The value fails the identifier grammar.

Source code in fuzzy1337/executors/contracts.py

def RequireIdentifier(value: str, field_name: str) -> None:
    """Require a portable lowercase machine identifier.

    Args:
        value: Candidate portable lowercase identifier
        field_name: Field label in validation errors.

    Raises:
        ValueError: The value fails the identifier grammar.
    """

    if not isinstance(value, str) or not _IDENTIFIER_PATTERN.fullmatch(value):
        raise ValueError(f"{field_name} must be a lowercase machine-readable identifier")

RequireReference(value: str, field_name: str) -> None

Require a non-empty single-line reference value.

Parameters:

Name	Type	Description	Default
value	str	Nonempty single-line text without NUL	required
field_name	str	Field label in validation errors.	required

Raises:

Type	Description
ValueError	The reference is not valid nonempty single-line text.

Source code in fuzzy1337/executors/contracts.py

def RequireReference(value: str, field_name: str) -> None:
    """Require a non-empty single-line reference value.

    Args:
        value: Nonempty single-line text without NUL
        field_name: Field label in validation errors.

    Raises:
        ValueError: The reference is not valid nonempty single-line text.
    """

    if not isinstance(value, str) or not value or "\x00" in value or "\n" in value or "\r" in value:
        raise ValueError(f"{field_name} must be a non-empty single-line reference")

UniqueIdentifiers(values: tuple[str, ...], field_name: str) -> tuple[str, ...]

Validate and freeze a sequence of unique identifiers.

Parameters:

Name	Type	Description	Default
values	tuple[str, ...]	Identifiers to validate without reordering	required
field_name	str	Field label in validation errors.	required

Returns:

Type	Description
tuple[str, ...]	Validated immutable tuple in declaration order.

Raises:

Type	Description
ValueError	An identifier is invalid or duplicated.

Source code in fuzzy1337/executors/contracts.py

def UniqueIdentifiers(values: tuple[str, ...], field_name: str) -> tuple[str, ...]:
    """Validate and freeze a sequence of unique identifiers.

    Args:
        values: Identifiers to validate without reordering
        field_name: Field label in validation errors.

    Returns:
        Validated immutable tuple in declaration order.

    Raises:
        ValueError: An identifier is invalid or duplicated.
    """

    normalized = tuple(values)

    for value in normalized:
        RequireIdentifier(value, field_name)

    if len(set(normalized)) != len(normalized):
        raise ValueError(f"{field_name} must contain unique values")

    return normalized

InvocationDigest(invocation: AdapterInvocation) -> str

Bind an authorization record to the exact immutable adapter invocation.

A digest binds metadata; it does not establish authorization by itself.

Parameters:

Name	Type	Description	Default
invocation	AdapterInvocation	Exact prepared invocation to bind to authorization.	required

Returns:

Type	Description
str	Lowercase SHA-256 of the deterministic serialized invocation.

Source code in fuzzy1337/executors/contracts.py

def InvocationDigest(invocation: AdapterInvocation) -> str:
    """Bind an authorization record to the exact immutable adapter invocation.

    A digest binds metadata; it does not establish authorization by itself.

    Args:
        invocation: Exact prepared invocation to bind to authorization.

    Returns:
        Lowercase SHA-256 of the deterministic serialized invocation.
    """

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

CapabilityDescriptors(descriptor: AdapterDescriptor) -> tuple[CapabilityDescriptor, ...]

Expand one adapter declaration into deterministic capability records.

Parameters:

Name	Type	Description	Default
descriptor	AdapterDescriptor	Validated adapter metadata and privilege declarations.	required

Returns:

Type	Description
tuple[CapabilityDescriptor, ...]	Capability records in the adapter declaration order.

Source code in fuzzy1337/executors/contracts.py

def CapabilityDescriptors(descriptor: AdapterDescriptor) -> tuple[CapabilityDescriptor, ...]:
    """Expand one adapter declaration into deterministic capability records.

    Args:
        descriptor: Validated adapter metadata and privilege declarations.

    Returns:
        Capability records in the adapter declaration order.
    """

    return tuple(
        CapabilityDescriptor(
            identifier=capability,
            adapter_id=descriptor.adapter_id,
            maximum_impact=descriptor.maximum_impact,
            required_privileges=descriptor.required_privileges,
        )
        for capability in descriptor.capabilities
    )

FreezeEnvironment(environment: Mapping[str, str]) -> Mapping[str, str]

Return validated immutable environment overrides.

Syntax validation does not detect secrets; callers must avoid recording secret values.

Parameters:

Name	Type	Description	Default
environment	Mapping[str, str]	Explicit name/value overrides for the child environment.	required

Returns:

Type	Description
Mapping[str, str]	Immutable mapping in deterministic name order.

Raises:

Type	Description
ValueError	A name is not portable or a value is not text or contains NUL.

Source code in fuzzy1337/executors/contracts.py

def FreezeEnvironment(environment: Mapping[str, str]) -> Mapping[str, str]:
    """Return validated immutable environment overrides.

    Syntax validation does not detect secrets; callers must avoid recording secret values.

    Args:
        environment: Explicit name/value overrides for the child environment.

    Returns:
        Immutable mapping in deterministic name order.

    Raises:
        ValueError: A name is not portable or a value is not text or contains NUL.
    """

    frozen: dict[str, str] = {}

    for name, value in environment.items():
        if not isinstance(name, str) or not _ENVIRONMENT_NAME_PATTERN.fullmatch(name):
            raise ValueError("environment names must use portable variable syntax")

        if not isinstance(value, str) or "\x00" in value:
            raise ValueError("environment values must be strings without NUL bytes")

        frozen[name] = value

    return MappingProxyType(dict(sorted(frozen.items())))