English fallback

This translation is missing; the canonical English reference follows.

fuzzy1337.adapters.page_capture

Experimental ADR 0052 adapter and finite records for one owned synthetic page. KnownSyntheticHtmlBody requires the explicit closed reviewed fixture ID; arbitrary caller body digests, file paths and default body selections are refused. The immutable catalog fields describe exact admitted bytes and supply no consent.

NativeLoopbackPageAdapter declares the distinct SAFE http.synthetic-page-capture capability of native.loopback-page, version 1. Health and preparation perform no network operation. Actual execution requires the current explicit URL/IP scope and capability permission through the existing policy/executor owners. Plain HTTP at exact literal 127.0.0.1 or ::1 is the only transport profile. NativeHTTP version 1 remains unchanged.

Normalization validates canonical flat worker stdout, exact invocation and terminal execution, bounded owning evidence and the closed response body. It produces only one observed source response envelope. It creates no hosts, ports, services, findings, connectivity or fetched-link claims. Internal parser helpers are consistency boundaries, not alternate authorization/launch APIs. ADR 0087 requires isolated imports for current prepared invocations and default provider validation. Only pure normalization admits an exact historical launch shape, retaining original argv/digests without enabling that old execution path.

ADR 0090 adds explicit PageRenderer.PLAYWRIGHT_OFFLINE selection and the separate SAFE http.synthetic-page-browser capability. The native default retains its exact descriptor and argv. BrowserPageCaptureLimits supplies the explicit larger cleanup/custody budget for the browser selection; mixing limit types is refused. The optional Linux browser worker renders only an already admitted closed native body, with no browser navigation or page JavaScript. Its separate internal grammar binds actual engine versions and finite DOM metrics to original input evidence.

fuzzy1337.adapters.page_capture

Closed synthetic loopback HTML inputs and exact verified worker record admission.

PageRenderer

Bases: str, Enum

Select one fixed page implementation without arbitrary programs, flags or content.

Source code in fuzzy1337/adapters/page_capture.py

class PageRenderer(str, Enum):
    """Select one fixed page implementation without arbitrary programs, flags or content."""

    NATIVE = "native"
    PLAYWRIGHT_OFFLINE = PAGE_BROWSER_PROFILE

PageCaptureRecordError

Bases: ValueError

Untrusted output cannot establish a bounded attributable synthetic page.

Source code in fuzzy1337/adapters/page_capture.py

class PageCaptureRecordError(ValueError):
    """Untrusted output cannot establish a bounded attributable synthetic page."""

KnownSyntheticHtmlBody dataclass

Select a finite reviewed non-secret body; selection supplies no execution authority.

Source code in fuzzy1337/adapters/page_capture.py

@dataclass(frozen=True, slots=True)
class KnownSyntheticHtmlBody:
    """Select a finite reviewed non-secret body; selection supplies no execution authority."""

    fixture_id: str
    sha256: str = field(init=False)
    size_bytes: int = field(init=False)

    def __post_init__(self) -> None:
        """Derive immutable catalog identity without files, arbitrary digests or defaults."""

        if type(self.fixture_id) is not str or len(self.fixture_id) > 64 or (
            self.fixture_id not in _REVIEWED_BODIES
        ):
            raise ValueError("page capture requires an explicit reviewed fixture identifier")

        digest, size = _REVIEWED_BODIES[self.fixture_id]
        object.__setattr__(self, "sha256", digest)
        object.__setattr__(self, "size_bytes", size)

__post_init__() -> None

Derive immutable catalog identity without files, arbitrary digests or defaults.

Source code in fuzzy1337/adapters/page_capture.py

def __post_init__(self) -> None:
    """Derive immutable catalog identity without files, arbitrary digests or defaults."""

    if type(self.fixture_id) is not str or len(self.fixture_id) > 64 or (
        self.fixture_id not in _REVIEWED_BODIES
    ):
        raise ValueError("page capture requires an explicit reviewed fixture identifier")

    digest, size = _REVIEWED_BODIES[self.fixture_id]
    object.__setattr__(self, "sha256", digest)
    object.__setattr__(self, "size_bytes", size)

PageCaptureLimits dataclass

Finite one-request body and whole-operation deadlines, with no expansion knobs.

Source code in fuzzy1337/adapters/page_capture.py

@dataclass(frozen=True, slots=True)
class PageCaptureLimits:
    """Finite one-request body and whole-operation deadlines, with no expansion knobs."""

    max_body_bytes: int = 16384
    request_timeout_seconds: float = 2.0
    timeout_seconds: float = 10.0

    def __post_init__(self) -> None:
        """Reject coercions, huge/nonfinite seconds and inconsistent startup budgets."""

        Integer(self.max_body_bytes, 1, MAX_BODY_BYTES)
        request = Seconds(self.request_timeout_seconds, 5)
        provider = Seconds(self.timeout_seconds, 15)

        if provider < request + 1:
            raise ValueError("page capture provider budget must include request plus startup")

__post_init__() -> None

Reject coercions, huge/nonfinite seconds and inconsistent startup budgets.

Source code in fuzzy1337/adapters/page_capture.py

def __post_init__(self) -> None:
    """Reject coercions, huge/nonfinite seconds and inconsistent startup budgets."""

    Integer(self.max_body_bytes, 1, MAX_BODY_BYTES)
    request = Seconds(self.request_timeout_seconds, 5)
    provider = Seconds(self.timeout_seconds, 15)

    if provider < request + 1:
        raise ValueError("page capture provider budget must include request plus startup")

BrowserPageCaptureLimits dataclass

Browser-only budgets covering helper readiness, worker, cleanup and source custody.

Source code in fuzzy1337/adapters/page_capture.py

@dataclass(frozen=True, slots=True)
class BrowserPageCaptureLimits:
    """Browser-only budgets covering helper readiness, worker, cleanup and source custody."""

    max_body_bytes: int = 16384
    request_timeout_seconds: float = 2.0
    timeout_seconds: float = 18.0

    def __post_init__(self) -> None:
        """Require the finite whole-operation margin without expanding native page defaults."""

        Integer(self.max_body_bytes, 1, MAX_BODY_BYTES)
        request = Seconds(self.request_timeout_seconds, 5)
        provider = Seconds(self.timeout_seconds, 30)

        if provider < 3 + math.ceil(request + 7) + 3 + 1:
            raise ValueError(
                "browser provider budget must cover readiness, worker, cleanup and Save")

__post_init__() -> None

Require the finite whole-operation margin without expanding native page defaults.

Source code in fuzzy1337/adapters/page_capture.py

def __post_init__(self) -> None:
    """Require the finite whole-operation margin without expanding native page defaults."""

    Integer(self.max_body_bytes, 1, MAX_BODY_BYTES)
    request = Seconds(self.request_timeout_seconds, 5)
    provider = Seconds(self.timeout_seconds, 30)

    if provider < 3 + math.ceil(request + 7) + 3 + 1:
        raise ValueError(
            "browser provider budget must cover readiness, worker, cleanup and Save")

NativeLoopbackPageAdapter

Prepare and normalize only the distinct SAFE closed synthetic-page capability.

Source code in fuzzy1337/adapters/page_capture.py

class NativeLoopbackPageAdapter:
    """Prepare and normalize only the distinct SAFE closed synthetic-page capability."""

    def __init__(self, evidence_reader: Callable[[EvidenceReference], bytes], *,
                 renderer: PageRenderer = PageRenderer.NATIVE) -> None:
        """Bind an existing owning evidence reader without target probes or filesystem reads.

        Args:
            evidence_reader: Actual owner integrity reader, never a caller-selected file loader.
            renderer: Exact finite implementation selection; the native default stays unchanged.
        """

        if not callable(evidence_reader) or type(renderer) is not PageRenderer:
            raise ValueError("page capture evidence reader must be callable")

        self._read_evidence = evidence_reader
        self._renderer = renderer

    @property
    def Descriptor(self) -> AdapterDescriptor:
        """Return packaged provider/version/capability metadata without granting authority.

        Returns:
            Immutable separate SAFE synthetic-page declaration with no privilege requirement.
        """

        return _DESCRIPTOR if self._renderer is PageRenderer.NATIVE else _BROWSER_DESCRIPTOR

    def CheckHealth(self) -> AdapterHealth:
        """Report first-party package availability without network or tool discovery.

        Returns:
            Packaged provider availability/version, without permission to launch.
        """

        return AdapterHealth(PAGE_CAPTURE_PROVIDER_ID, AdapterHealthState.AVAILABLE,
                             PAGE_CAPTURE_VERSION)

    def PrepareInvocation(self, request: AdapterRequest) -> AdapterInvocation:
        """Prepare the exact bounded argv; actual launch requires existing current policy.

        Args:
            request: Exact SAFE request with URL, timeout, quota and explicit closed fixture ID.

        Returns:
            Exact current interpreter/module/argv with startup-inclusive executor timeout.

        Raises:
            ValueError: Types, capability, parameters, target or closed catalog are invalid.
        """

        capability = PAGE_CAPTURE_CAPABILITY if self._renderer is PageRenderer.NATIVE else (
            PAGE_BROWSER_CAPABILITY)

        if type(request) is not AdapterRequest or type(request.capability) is not str or (
            request.capability != capability
            or type(request.target_reference) is not str
            or request.impact is not ImpactLevel.SAFE
            or type(request.credential_references) is not tuple
            or request.credential_references or type(request.parameters) is not MappingProxyType
            or len(request.parameters) != 4
            or set(request.parameters) != {"url", "timeout_seconds", "fixture_id", "max_body_bytes"}
        ):
            raise ValueError("page capture requires its exact anonymous SAFE input parameters")

        target, _, _, _, _ = Destination(request.parameters["url"])
        timeout = Seconds(request.parameters["timeout_seconds"], 5)
        fixture_id = request.parameters["fixture_id"]

        if type(fixture_id) is not str:
            raise ValueError("page capture requires an exact text fixture identifier")

        body = KnownSyntheticHtmlBody(fixture_id)
        maximum = Integer(request.parameters["max_body_bytes"], 1, MAX_BODY_BYTES)

        if body.size_bytes > maximum or request.target_reference != target.Reference():
            raise ValueError("page capture selected body or exact target binding differs")

        module = "fuzzy1337.page_capture_worker" if self._renderer is PageRenderer.NATIVE else (
            "fuzzy1337.browser_page_worker")
        startup = 1 if self._renderer is PageRenderer.NATIVE else 7

        return AdapterInvocation(PAGE_CAPTURE_PROVIDER_ID, request, (
            sys.executable, "-I", "-m", module, target.value,
            str(timeout), body.fixture_id, str(maximum),
        ), math.ceil(timeout + startup), PAGE_CAPTURE_VERSION)

    def ValidateInvocation(
        self, invocation: object, *, allow_legacy: bool = False,
    ) -> AdapterInvocation:
        """Reconstruct exact nested request/argv fields and their owning invocation digest.

        Args:
            invocation: Typed prepared invocation or copied wrapper requiring nested revalidation.
            allow_legacy: Admit exact historical argv only for read-only report interpretation.
                Current launch/provider validation leaves this disabled.

        Returns:
            Original invocation only after exact reconstruction and owning SHA agreement.
        """

        if type(invocation) is not AdapterInvocation or type(invocation.argv) is not tuple or (
            len(invocation.argv) not in ((7, 8) if allow_legacy is True else (8,))
            or any(type(argument) is not str for argument in invocation.argv)
            or type(invocation.timeout_seconds) is not int
            or type(invocation.adapter_id) is not str
            or type(invocation.provider_version) is not str
        ):
            raise PageCaptureRecordError("page capture requires an exact prepared invocation")

        prepared = self.PrepareInvocation(invocation.request)

        if allow_legacy is True and self._renderer is PageRenderer.NATIVE and (
            invocation.argv == (prepared.argv[0], *prepared.argv[2:])
        ):
            prepared = replace(prepared, argv=invocation.argv)

        if invocation != prepared or InvocationDigest(invocation) != InvocationDigest(prepared):
            raise PageCaptureRecordError("page capture invocation differs from preparation")

        return invocation

    def NormalizeReport(self, report: AdapterReport) -> AdapterResult:
        """Return one source response envelope after exact invocation/evidence/body validation.

        No Host/Port/Service, relationships, findings or lexical connectivity are synthesized.

        Args:
            report: Successful real execution, exact invocation and sole bounded stdout evidence.

        Returns:
            One normalized source response observation, without mutation or new connectivity.

        Raises:
            PageCaptureRecordError: Any successful report binding is unsupported or inconsistent.
        """

        try:
            if type(report) is not AdapterReport or type(report.evidence) is not tuple or (
                len(report.evidence) != 1
            ):
                raise ValueError("invalid report shape")

            invocation = self.ValidateInvocation(report.invocation, allow_legacy=True)
            execution = Execution(report.execution)

            if execution.state is not ExecutionState.SUCCEEDED or execution.exit_code != 0:
                raise ValueError("unsuccessful report")

            reference = Reference(report.evidence[0])
            raw = self._read_evidence(reference)

            if type(raw) is not bytes or len(raw) != reference.size_bytes or (
                hashlib.sha256(raw).hexdigest() != reference.sha256
            ):
                raise ValueError("stdout evidence differs from its exact reference")

            parameters = invocation.request.parameters
            parser = ParseRecord

            if self._renderer is PageRenderer.PLAYWRIGHT_OFFLINE:
                from fuzzy1337.adapters.browser_page import ParseBrowserRecord

                parser = ParseBrowserRecord

            record = parser(raw, cast(str, parameters["url"]),
                KnownSyntheticHtmlBody(cast(str, parameters["fixture_id"])),
                cast(int, parameters["max_body_bytes"]))

            if "error" in record:
                raise ValueError("failure record cannot become a successful page")

            metadata = {key: value for key, value in record.items() if key != "body_base64"}

            kind = "http.synthetic-page" if self._renderer is PageRenderer.NATIVE else (
                "http.synthetic-page-browser")

            return AdapterResult(report, (NormalizedObservation(kind,
                invocation.request.target_reference, metadata),))

        except (ValueError, TypeError, AttributeError, OverflowError, KeyError, OSError):
            raise PageCaptureRecordError("page capture report could not be safely normalized") \
                from None

Descriptor: AdapterDescriptor property

Return packaged provider/version/capability metadata without granting authority.

Returns:

Type	Description
AdapterDescriptor	Immutable separate SAFE synthetic-page declaration with no privilege requirement.

__init__(evidence_reader: Callable[[EvidenceReference], bytes], *, renderer: PageRenderer = PageRenderer.NATIVE) -> None

Bind an existing owning evidence reader without target probes or filesystem reads.

Parameters:

Name	Type	Description	Default
evidence_reader	Callable[[EvidenceReference], bytes]	Actual owner integrity reader, never a caller-selected file loader.	required
renderer	PageRenderer	Exact finite implementation selection; the native default stays unchanged.	NATIVE

Source code in fuzzy1337/adapters/page_capture.py

def __init__(self, evidence_reader: Callable[[EvidenceReference], bytes], *,
             renderer: PageRenderer = PageRenderer.NATIVE) -> None:
    """Bind an existing owning evidence reader without target probes or filesystem reads.

    Args:
        evidence_reader: Actual owner integrity reader, never a caller-selected file loader.
        renderer: Exact finite implementation selection; the native default stays unchanged.
    """

    if not callable(evidence_reader) or type(renderer) is not PageRenderer:
        raise ValueError("page capture evidence reader must be callable")

    self._read_evidence = evidence_reader
    self._renderer = renderer

CheckHealth() -> AdapterHealth

Report first-party package availability without network or tool discovery.

Returns:

Type	Description
AdapterHealth	Packaged provider availability/version, without permission to launch.

Source code in fuzzy1337/adapters/page_capture.py

def CheckHealth(self) -> AdapterHealth:
    """Report first-party package availability without network or tool discovery.

    Returns:
        Packaged provider availability/version, without permission to launch.
    """

    return AdapterHealth(PAGE_CAPTURE_PROVIDER_ID, AdapterHealthState.AVAILABLE,
                         PAGE_CAPTURE_VERSION)

PrepareInvocation(request: AdapterRequest) -> AdapterInvocation

Prepare the exact bounded argv; actual launch requires existing current policy.

Parameters:

Name	Type	Description	Default
request	AdapterRequest	Exact SAFE request with URL, timeout, quota and explicit closed fixture ID.	required

Returns:

Type	Description
AdapterInvocation	Exact current interpreter/module/argv with startup-inclusive executor timeout.

Raises:

Type	Description
ValueError	Types, capability, parameters, target or closed catalog are invalid.

Source code in fuzzy1337/adapters/page_capture.py

def PrepareInvocation(self, request: AdapterRequest) -> AdapterInvocation:
    """Prepare the exact bounded argv; actual launch requires existing current policy.

    Args:
        request: Exact SAFE request with URL, timeout, quota and explicit closed fixture ID.

    Returns:
        Exact current interpreter/module/argv with startup-inclusive executor timeout.

    Raises:
        ValueError: Types, capability, parameters, target or closed catalog are invalid.
    """

    capability = PAGE_CAPTURE_CAPABILITY if self._renderer is PageRenderer.NATIVE else (
        PAGE_BROWSER_CAPABILITY)

    if type(request) is not AdapterRequest or type(request.capability) is not str or (
        request.capability != capability
        or type(request.target_reference) is not str
        or request.impact is not ImpactLevel.SAFE
        or type(request.credential_references) is not tuple
        or request.credential_references or type(request.parameters) is not MappingProxyType
        or len(request.parameters) != 4
        or set(request.parameters) != {"url", "timeout_seconds", "fixture_id", "max_body_bytes"}
    ):
        raise ValueError("page capture requires its exact anonymous SAFE input parameters")

    target, _, _, _, _ = Destination(request.parameters["url"])
    timeout = Seconds(request.parameters["timeout_seconds"], 5)
    fixture_id = request.parameters["fixture_id"]

    if type(fixture_id) is not str:
        raise ValueError("page capture requires an exact text fixture identifier")

    body = KnownSyntheticHtmlBody(fixture_id)
    maximum = Integer(request.parameters["max_body_bytes"], 1, MAX_BODY_BYTES)

    if body.size_bytes > maximum or request.target_reference != target.Reference():
        raise ValueError("page capture selected body or exact target binding differs")

    module = "fuzzy1337.page_capture_worker" if self._renderer is PageRenderer.NATIVE else (
        "fuzzy1337.browser_page_worker")
    startup = 1 if self._renderer is PageRenderer.NATIVE else 7

    return AdapterInvocation(PAGE_CAPTURE_PROVIDER_ID, request, (
        sys.executable, "-I", "-m", module, target.value,
        str(timeout), body.fixture_id, str(maximum),
    ), math.ceil(timeout + startup), PAGE_CAPTURE_VERSION)

ValidateInvocation(invocation: object, *, allow_legacy: bool = False) -> AdapterInvocation

Reconstruct exact nested request/argv fields and their owning invocation digest.

Parameters:

Name	Type	Description	Default
invocation	object	Typed prepared invocation or copied wrapper requiring nested revalidation.	required
allow_legacy	bool	Admit exact historical argv only for read-only report interpretation. Current launch/provider validation leaves this disabled.	False

Returns:

Type	Description
AdapterInvocation	Original invocation only after exact reconstruction and owning SHA agreement.

Source code in fuzzy1337/adapters/page_capture.py

def ValidateInvocation(
    self, invocation: object, *, allow_legacy: bool = False,
) -> AdapterInvocation:
    """Reconstruct exact nested request/argv fields and their owning invocation digest.

    Args:
        invocation: Typed prepared invocation or copied wrapper requiring nested revalidation.
        allow_legacy: Admit exact historical argv only for read-only report interpretation.
            Current launch/provider validation leaves this disabled.

    Returns:
        Original invocation only after exact reconstruction and owning SHA agreement.
    """

    if type(invocation) is not AdapterInvocation or type(invocation.argv) is not tuple or (
        len(invocation.argv) not in ((7, 8) if allow_legacy is True else (8,))
        or any(type(argument) is not str for argument in invocation.argv)
        or type(invocation.timeout_seconds) is not int
        or type(invocation.adapter_id) is not str
        or type(invocation.provider_version) is not str
    ):
        raise PageCaptureRecordError("page capture requires an exact prepared invocation")

    prepared = self.PrepareInvocation(invocation.request)

    if allow_legacy is True and self._renderer is PageRenderer.NATIVE and (
        invocation.argv == (prepared.argv[0], *prepared.argv[2:])
    ):
        prepared = replace(prepared, argv=invocation.argv)

    if invocation != prepared or InvocationDigest(invocation) != InvocationDigest(prepared):
        raise PageCaptureRecordError("page capture invocation differs from preparation")

    return invocation

NormalizeReport(report: AdapterReport) -> AdapterResult

Return one source response envelope after exact invocation/evidence/body validation.

No Host/Port/Service, relationships, findings or lexical connectivity are synthesized.

Parameters:

Name	Type	Description	Default
report	AdapterReport	Successful real execution, exact invocation and sole bounded stdout evidence.	required

Returns:

Type	Description
AdapterResult	One normalized source response observation, without mutation or new connectivity.

Raises:

Type	Description
PageCaptureRecordError	Any successful report binding is unsupported or inconsistent.

Source code in fuzzy1337/adapters/page_capture.py

def NormalizeReport(self, report: AdapterReport) -> AdapterResult:
    """Return one source response envelope after exact invocation/evidence/body validation.

    No Host/Port/Service, relationships, findings or lexical connectivity are synthesized.

    Args:
        report: Successful real execution, exact invocation and sole bounded stdout evidence.

    Returns:
        One normalized source response observation, without mutation or new connectivity.

    Raises:
        PageCaptureRecordError: Any successful report binding is unsupported or inconsistent.
    """

    try:
        if type(report) is not AdapterReport or type(report.evidence) is not tuple or (
            len(report.evidence) != 1
        ):
            raise ValueError("invalid report shape")

        invocation = self.ValidateInvocation(report.invocation, allow_legacy=True)
        execution = Execution(report.execution)

        if execution.state is not ExecutionState.SUCCEEDED or execution.exit_code != 0:
            raise ValueError("unsuccessful report")

        reference = Reference(report.evidence[0])
        raw = self._read_evidence(reference)

        if type(raw) is not bytes or len(raw) != reference.size_bytes or (
            hashlib.sha256(raw).hexdigest() != reference.sha256
        ):
            raise ValueError("stdout evidence differs from its exact reference")

        parameters = invocation.request.parameters
        parser = ParseRecord

        if self._renderer is PageRenderer.PLAYWRIGHT_OFFLINE:
            from fuzzy1337.adapters.browser_page import ParseBrowserRecord

            parser = ParseBrowserRecord

        record = parser(raw, cast(str, parameters["url"]),
            KnownSyntheticHtmlBody(cast(str, parameters["fixture_id"])),
            cast(int, parameters["max_body_bytes"]))

        if "error" in record:
            raise ValueError("failure record cannot become a successful page")

        metadata = {key: value for key, value in record.items() if key != "body_base64"}

        kind = "http.synthetic-page" if self._renderer is PageRenderer.NATIVE else (
            "http.synthetic-page-browser")

        return AdapterResult(report, (NormalizedObservation(kind,
            invocation.request.target_reference, metadata),))

    except (ValueError, TypeError, AttributeError, OverflowError, KeyError, OSError):
        raise PageCaptureRecordError("page capture report could not be safely normalized") \
            from None

Integer(value: object, minimum: int, maximum: int) -> int

Validate exact bounded integers before conversions or input-derived allocation.

Parameters:

Name	Type	Description	Default
value	object	Explicit integer, excluding booleans and coercion.	required
minimum	int	Inclusive admitted lower bound.	required
maximum	int	Inclusive admitted upper bound.	required

Returns:

Type	Description
int	Original exact integer.

Source code in fuzzy1337/adapters/page_capture.py

def Integer(value: object, minimum: int, maximum: int) -> int:
    """Validate exact bounded integers before conversions or input-derived allocation.

    Args:
        value: Explicit integer, excluding booleans and coercion.
        minimum: Inclusive admitted lower bound.
        maximum: Inclusive admitted upper bound.

    Returns:
        Original exact integer.
    """

    if type(value) is not int or not minimum <= value <= maximum:
        raise ValueError("page capture integer is outside its admitted bounds")

    return value

Seconds(value: object, maximum: float) -> float

Validate numeric request/provider seconds before finite floating-point conversion.

Parameters:

Name	Type	Description	Default
value	object	Exact finite int/float seconds, excluding booleans.	required
maximum	float	Inclusive profile ceiling; the minimum is 0.05 seconds.	required

Returns:

Type	Description
float	Validated deadline as a float.

Source code in fuzzy1337/adapters/page_capture.py

def Seconds(value: object, maximum: float) -> float:
    """Validate numeric request/provider seconds before finite floating-point conversion.

    Args:
        value: Exact finite int/float seconds, excluding booleans.
        maximum: Inclusive profile ceiling; the minimum is 0.05 seconds.

    Returns:
        Validated deadline as a float.
    """

    if not isinstance(value, (int, float)) or type(value) not in (int, float) or (
        not 0.05 <= value <= maximum
    ):
        raise ValueError("page capture deadline is outside its admitted bounds")

    if not math.isfinite(value):
        raise ValueError("page capture deadline must be finite")

    return float(value)

KnownBody(value: object) -> KnownSyntheticHtmlBody

Revalidate copied catalog records including the exact types of derived fields.

Parameters:

Name	Type	Description	Default
value	object	Explicit closed fixture selection requiring complete typed validation.	required

Returns:

Type	Description
KnownSyntheticHtmlBody	Original unchanged catalog record; never arbitrary body approval.

Source code in fuzzy1337/adapters/page_capture.py

def KnownBody(value: object) -> KnownSyntheticHtmlBody:
    """Revalidate copied catalog records including the exact types of derived fields.

    Args:
        value: Explicit closed fixture selection requiring complete typed validation.

    Returns:
        Original unchanged catalog record; never arbitrary body approval.
    """

    if type(value) is not KnownSyntheticHtmlBody or type(value.sha256) is not str or (
        type(value.size_bytes) is not int
        or KnownSyntheticHtmlBody(value.fixture_id) != value
    ):
        raise ValueError("page capture requires an unchanged typed closed body selection")

    return value

Destination(value: object) -> tuple[Target, str, int, str, str]

Require the exact canonical anonymous HTTP URL at 127.0.0.1 or ::1, without DNS.

Parameters:

Name	Type	Description	Default
value	object	Exact canonical ASCII URL, at most 2048 bytes, without query or credentials.	required

Returns:

Type	Description
tuple[Target, str, int, str, str]	Canonical target, literal address, port, concrete path and Host authority.

Raises:

Type	Description
ValueError	URL text or any structural destination component is unsupported.

Source code in fuzzy1337/adapters/page_capture.py

def Destination(value: object) -> tuple[Target, str, int, str, str]:
    """Require the exact canonical anonymous HTTP URL at 127.0.0.1 or ::1, without DNS.

    Args:
        value: Exact canonical ASCII URL, at most 2048 bytes, without query or credentials.

    Returns:
        Canonical target, literal address, port, concrete path and Host authority.

    Raises:
        ValueError: URL text or any structural destination component is unsupported.
    """

    if type(value) is not str or not value or len(value) > 2048:
        raise ValueError("page capture requires bounded canonical loopback URL text")

    try:
        value.encode("ascii")
        target = Target(TargetKind.URL, value)
        parsed = urlsplit(value)
        address = parsed.hostname
        path = parsed.path
        port = 80 if parsed.port is None else parsed.port

        if target.value != value or parsed.scheme != "http" or address not in (
            "127.0.0.1", "::1"
        ) or parsed.username is not None or parsed.password is not None or (
            parsed.query or parsed.fragment or not path or len(path) > 1024
            or any(character in "%{}?\\#" or character.isspace() or ord(character) < 32
                   or ord(character) == 127 for character in path)
        ):
            raise ValueError("unsupported destination")

    except (ValueError, UnicodeError):
        raise ValueError("page capture requires an exact anonymous literal-loopback HTTP URL") \
            from None

    return target, address, port, path, parsed.netloc

CanonicalRecord(record: Mapping[str, object]) -> bytes

Encode the one flat strict worker record in sorted compact UTF-8 JSON plus LF.

Parameters:

Name	Type	Description	Default
record	Mapping[str, object]	Already bounded validated worker fields or fixed failure metadata.	required

Returns:

Type	Description
bytes	Canonical sorted compact JSON bytes with exactly one final LF.

Source code in fuzzy1337/adapters/page_capture.py

def CanonicalRecord(record: Mapping[str, object]) -> bytes:
    """Encode the one flat strict worker record in sorted compact UTF-8 JSON plus LF.

    Args:
        record: Already bounded validated worker fields or fixed failure metadata.

    Returns:
        Canonical sorted compact JSON bytes with exactly one final LF.
    """

    return json.dumps(record, sort_keys=True, ensure_ascii=False, allow_nan=False,
                      separators=(",", ":")).encode("utf-8") + b"\n"

UniqueObject(pairs: list[tuple[str, object]]) -> dict[str, object]

Refuse duplicate JSON fields instead of allowing last-key-wins interpretation.

Parameters:

Name	Type	Description	Default
pairs	list[tuple[str, object]]	JSON decoder's bounded object members in observed order.	required

Returns:

Type	Description
dict[str, object]	Original fields as a dictionary, only when every name is unique.

Source code in fuzzy1337/adapters/page_capture.py

def UniqueObject(pairs: list[tuple[str, object]]) -> dict[str, object]:
    """Refuse duplicate JSON fields instead of allowing last-key-wins interpretation.

    Args:
        pairs: JSON decoder's bounded object members in observed order.

    Returns:
        Original fields as a dictionary, only when every name is unique.
    """

    result: dict[str, object] = {}

    for key, value in pairs:
        if key in result:
            raise ValueError("duplicate worker field")

        result[key] = value

    return result

RejectConstant(value: str) -> None

Refuse nonstandard NaN/Infinity JSON without returning input-derived text.

Parameters:

Name	Type	Description	Default
value	str	Transient unsupported numeric token; it is never echoed or retained.	required

Source code in fuzzy1337/adapters/page_capture.py

def RejectConstant(value: str) -> None:
    """Refuse nonstandard NaN/Infinity JSON without returning input-derived text.

    Args:
        value: Transient unsupported numeric token; it is never echoed or retained.
    """

    raise ValueError("nonfinite worker field")

FlatJson(raw: bytes) -> dict[str, object]

Bound nesting before JSON decode and require byte-for-byte canonical flat output.

Parameters:

Name	Type	Description	Default
raw	bytes	Exact immutable stdout bytes, at most 98304 bytes.	required

Returns:

Type	Description
dict[str, object]	The flat decoded record after duplicate/nonfinite/encoding rejection.

Source code in fuzzy1337/adapters/page_capture.py

def FlatJson(raw: bytes) -> dict[str, object]:
    """Bound nesting before JSON decode and require byte-for-byte canonical flat output.

    Args:
        raw: Exact immutable stdout bytes, at most 98304 bytes.

    Returns:
        The flat decoded record after duplicate/nonfinite/encoding rejection.
    """

    if type(raw) is not bytes or not 1 <= len(raw) <= MAX_RECORD_BYTES:
        raise ValueError("worker record exceeds its byte bound")

    depth = 0
    quoted = False
    escaped = False

    for byte in raw:
        if quoted:
            if escaped:
                escaped = False

            elif byte == 92:
                escaped = True

            elif byte == 34:
                quoted = False

        elif byte == 34:
            quoted = True

        elif byte == 123:
            depth += 1

            if depth > 1:
                raise ValueError("worker record must be flat")

        elif byte == 125:
            depth -= 1

        elif byte in (91, 93):
            raise ValueError("worker record must have scalar fields")

    record = json.loads(raw.decode("utf-8"), object_pairs_hook=UniqueObject,
                        parse_constant=RejectConstant)

    if type(record) is not dict or raw != CanonicalRecord(record):
        raise ValueError("worker record must use the exact canonical encoding")

    return record

DecodeBody(record: Mapping[str, object], expected: KnownSyntheticHtmlBody) -> bytes

Admit only canonical standard base64 for the exact reviewed UTF-8 body sequence.

Parameters:

Name	Type	Description	Default
record	Mapping[str, object]	Flat metadata with a bounded canonical body_base64 field.	required
expected	KnownSyntheticHtmlBody	Previously validated closed known-body catalog selection.	required

Returns:

Type	Description
bytes	Exact reviewed non-secret body bytes, with strict UTF-8 and no BOM.

Source code in fuzzy1337/adapters/page_capture.py

def DecodeBody(record: Mapping[str, object], expected: KnownSyntheticHtmlBody) -> bytes:
    """Admit only canonical standard base64 for the exact reviewed UTF-8 body sequence.

    Args:
        record: Flat metadata with a bounded canonical body_base64 field.
        expected: Previously validated closed known-body catalog selection.

    Returns:
        Exact reviewed non-secret body bytes, with strict UTF-8 and no BOM.
    """

    encoded = record["body_base64"]

    if type(encoded) is not str or len(encoded) != 4 * math.ceil(expected.size_bytes / 3):
        raise ValueError("body encoding differs from the reviewed declaration")

    body = base64.b64decode(encoded, validate=True)

    if base64.b64encode(body).decode("ascii") != encoded or len(body) != expected.size_bytes or (
        hashlib.sha256(body).hexdigest() != expected.sha256 or body.startswith(b"\xef\xbb\xbf")
    ):
        raise ValueError("body bytes differ from the reviewed declaration")

    body.decode("utf-8", errors="strict")

    return body

ParseRecord(raw: bytes, url: str, expected_body: KnownSyntheticHtmlBody, max_body_bytes: int) -> dict[str, object]

Independently validate all fixed failure/success fields before evidence publication.

Parameters:

Name	Type	Description	Default
raw	bytes	Exact bounded stdout, never stderr or an HTTP transcript.	required
url	str	Exact prepared approved canonical URL.	required
expected_body	KnownSyntheticHtmlBody	Explicit unchanged closed catalog selection.	required
max_body_bytes	int	Selected admitted integer body bound.	required

Returns:

Type	Description
dict[str, object]	Exact flat record, containing known non-secret bytes only on success.

Raises:

Type	Description
PageCaptureRecordError	Shape, encoding, metadata, endpoint or body is inconsistent.

Source code in fuzzy1337/adapters/page_capture.py

def ParseRecord(raw: bytes, url: str, expected_body: KnownSyntheticHtmlBody,
                max_body_bytes: int) -> dict[str, object]:
    """Independently validate all fixed failure/success fields before evidence publication.

    Args:
        raw: Exact bounded stdout, never stderr or an HTTP transcript.
        url: Exact prepared approved canonical URL.
        expected_body: Explicit unchanged closed catalog selection.
        max_body_bytes: Selected admitted integer body bound.

    Returns:
        Exact flat record, containing known non-secret bytes only on success.

    Raises:
        PageCaptureRecordError: Shape, encoding, metadata, endpoint or body is inconsistent.
    """

    try:
        expected = KnownBody(expected_body)
        Integer(max_body_bytes, 1, MAX_BODY_BYTES)

        if expected.size_bytes > max_body_bytes:
            raise ValueError("selected body does not fit")

        target, address, port, _, _ = Destination(url)
        record = FlatJson(raw)
        Integer(record.get("schema_version"), 1, 1)

        if "error" in record:
            if set(record) != {"schema_version", "error"} or (
                type(record["error"]) is not str or record["error"] not in ERROR_EXITS
            ):
                raise ValueError("unknown failure record")

            return record

        if set(record) != _SUCCESS_FIELDS or record["fixture_id"] != expected.fixture_id or (
            record["url"] != target.value or record["address"] != address
            or record["scheme"] != "http" or record["http_version"] not in (
                "HTTP/1.0", "HTTP/1.1"
            ) or record["content_type"] != "text/html"
            or record["declared_charset"] not in (None, "utf-8")
            or record["body_sha256"] != expected.sha256
        ):
            raise ValueError("worker metadata differs from the prepared destination/body")

        Integer(record["port"], port, port)
        Integer(record["status"], 200, 200)
        Integer(record["content_length"], expected.size_bytes, expected.size_bytes)
        count = Integer(record["header_count"], 2, MAX_HEADER_FIELDS)
        minimum = 58 + len(str(expected.size_bytes)) + (
            14 if record["declared_charset"] is not None else 0
        ) + (
            4 * (count - 2) + max(0, count - 53)
        )
        Integer(record["header_size_bytes"], minimum,
                min(MAX_HEADER_BYTES, MAX_LINE_BYTES * (count + 1) + 2))
        DecodeBody(record, expected)

        return record

    except (ValueError, TypeError, AttributeError, OverflowError, KeyError, RecursionError,
            binascii.Error):
        raise PageCaptureRecordError("page capture record violated the closed safe grammar") \
            from None

Execution(execution: object) -> AdapterExecution

Revalidate exact terminal execution records before comparing equality or persisting.

Parameters:

Name	Type	Description	Default
execution	object	Exact owning record, including finite nonnegative duration and terminal state.	required

Returns:

Type	Description
AdapterExecution	Original validated execution; later job failure remains a separate outcome.

Source code in fuzzy1337/adapters/page_capture.py

def Execution(execution: object) -> AdapterExecution:
    """Revalidate exact terminal execution records before comparing equality or persisting.

    Args:
        execution: Exact owning record, including finite nonnegative duration and terminal state.

    Returns:
        Original validated execution; later job failure remains a separate outcome.
    """

    if type(execution) is not AdapterExecution or type(execution.state) is not ExecutionState or (
        type(execution.duration_seconds) not in (int, float)
        or not 0 <= execution.duration_seconds <= sys.float_info.max
        or not math.isfinite(execution.duration_seconds)
        or execution.exit_code is not None and type(execution.exit_code) is not int
    ):
        raise PageCaptureRecordError("page capture requires exact finite execution metadata")

    if AdapterExecution(execution.state, execution.exit_code,
                        execution.duration_seconds) != execution:
        raise PageCaptureRecordError("page capture execution metadata is inconsistent")

    return execution

Reference(reference: object) -> EvidenceReference

Validate the exact bounded immutable stdout reference before invoking its reader.

Parameters:

Name	Type	Description	Default
reference	object	Exact immutable application/json stdout reference, at most 98304 bytes.	required

Returns:

Type	Description
EvidenceReference	Original role/locator/digest/size-validated reference.

Source code in fuzzy1337/adapters/page_capture.py

def Reference(reference: object) -> EvidenceReference:
    """Validate the exact bounded immutable stdout reference before invoking its reader.

    Args:
        reference: Exact immutable application/json stdout reference, at most 98304 bytes.

    Returns:
        Original role/locator/digest/size-validated reference.
    """

    if type(reference) is not EvidenceReference or type(reference.sha256) is not str or (
        not re.fullmatch(r"[0-9a-f]{64}", reference.sha256)
        or reference.role != "stdout" or reference.media_type != "application/json"
        or reference.locator != "blobs/sha256/" + reference.sha256
    ):
        raise PageCaptureRecordError("page capture requires its exact stdout evidence role")

    Integer(reference.size_bytes, 1, MAX_RECORD_BYTES)

    return reference