558 lines
20 KiB
Python
558 lines
20 KiB
Python
"""Strict M7-07 capability, receipt, and host-attempt conformance evidence."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import re
|
|
from collections.abc import Mapping
|
|
from dataclasses import dataclass, replace
|
|
from typing import Final, cast
|
|
|
|
from comfyui_sigmax.core import (
|
|
ArtifactBuildMetadata,
|
|
ArtifactField,
|
|
CapabilityDimension,
|
|
CompatibilityDecision,
|
|
ExecutionBehavior,
|
|
ExecutionComponent,
|
|
ExecutionFeatureRequest,
|
|
ExecutionHost,
|
|
ExecutionReceiptMetadata,
|
|
ExecutionRngOwnership,
|
|
ExecutionStatus,
|
|
ModelCapabilities,
|
|
NoiseOwnership,
|
|
PortableExecutionBundle,
|
|
ProfileCapabilities,
|
|
SamplerCapabilities,
|
|
SamplerState,
|
|
ScheduleOwnership,
|
|
TerminalRequirement,
|
|
TypedArtifactValue,
|
|
build_execution_receipt,
|
|
build_schedule_artifact,
|
|
canonical_projection_bytes,
|
|
deserialize_execution_receipt,
|
|
deserialize_portable_execution_bundle,
|
|
deserialize_schedule_artifact,
|
|
evaluate_compatibility,
|
|
serialize_execution_receipt,
|
|
serialize_portable_execution_bundle,
|
|
serialize_schedule_artifact,
|
|
)
|
|
from comfyui_sigmax.profiles import (
|
|
KREA2_TURBO_PROFILE,
|
|
KREA2_TURBO_SCHEMA,
|
|
build_krea2_turbo_schedule,
|
|
profile_schema_fingerprint,
|
|
)
|
|
from comfyui_sigmax.workflows.validation import (
|
|
CANONICAL_HOST_REVISION,
|
|
CANONICAL_HOST_VERSION,
|
|
)
|
|
|
|
REPORT_SCHEMA: Final = "sigmax.capability-receipt-conformance/1"
|
|
HOST_ATTEMPT_SCHEMA: Final = "sigmax.host-attempt-transition/1"
|
|
_SHA256_PATTERN: Final = re.compile(r"^sha256:[0-9a-f]{64}$")
|
|
_REASON_PATTERN: Final = re.compile(r"^[a-z][a-z0-9_.-]{0,127}$")
|
|
_STATUS_PATTERN: Final = re.compile(r"^[a-z][a-z0-9_.-]{0,63}$")
|
|
_LANE_PATTERN: Final = re.compile(r"^[A-Z][A-Z0-9_]{0,63}$")
|
|
_ROOT_FIELDS: Final = frozenset(
|
|
{
|
|
"cases",
|
|
"coverage",
|
|
"host_attempt_regressions",
|
|
"portable_execution",
|
|
"profile",
|
|
"schema",
|
|
"status",
|
|
}
|
|
)
|
|
_TRANSITION_FIELDS: Final = frozenset(
|
|
{"accepted", "first", "lane", "repeat", "schema", "transition"}
|
|
)
|
|
_ATTEMPT_FIELDS: Final = frozenset(
|
|
{"observed_status", "ordinal", "reason_code", "result_fingerprint", "verdict"}
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True, kw_only=True)
|
|
class HostAttempt:
|
|
"""One bounded non-pure host conformance attempt."""
|
|
|
|
ordinal: int
|
|
verdict: str
|
|
observed_status: str
|
|
reason_code: str | None
|
|
result_fingerprint: str
|
|
|
|
def __post_init__(self) -> None:
|
|
if type(self.ordinal) is not int or self.ordinal not in (1, 2):
|
|
raise ValueError("host attempt ordinal must be one or two")
|
|
if not isinstance(self.verdict, str) or self.verdict not in {"pass", "fail"}:
|
|
raise ValueError("host attempt verdict is unsupported")
|
|
if not isinstance(self.observed_status, str) or not _STATUS_PATTERN.fullmatch(
|
|
self.observed_status
|
|
):
|
|
raise ValueError("host attempt observed status is invalid")
|
|
if self.reason_code is not None and (
|
|
not isinstance(self.reason_code, str) or not _REASON_PATTERN.fullmatch(self.reason_code)
|
|
):
|
|
raise ValueError("host attempt reason code is invalid")
|
|
if self.verdict == "fail" and self.reason_code is None:
|
|
raise ValueError("failed host attempt requires a reason code")
|
|
if not isinstance(self.result_fingerprint, str) or not _SHA256_PATTERN.fullmatch(
|
|
self.result_fingerprint
|
|
):
|
|
raise ValueError("host attempt result fingerprint is invalid")
|
|
|
|
def projection(self) -> dict[str, object]:
|
|
return {
|
|
"observed_status": self.observed_status,
|
|
"ordinal": self.ordinal,
|
|
"reason_code": self.reason_code,
|
|
"result_fingerprint": self.result_fingerprint,
|
|
"verdict": self.verdict,
|
|
}
|
|
|
|
|
|
def _mapping(value: object, *, name: str) -> dict[str, object]:
|
|
if not isinstance(value, Mapping) or any(not isinstance(key, str) for key in value):
|
|
raise ValueError(f"{name} must be an object with string keys")
|
|
return dict(cast(Mapping[str, object], value))
|
|
|
|
|
|
def _exact(value: Mapping[str, object], fields: frozenset[str], *, name: str) -> None:
|
|
if set(value) != fields:
|
|
raise ValueError(f"{name} fields are incomplete or unknown")
|
|
|
|
|
|
def _attempt_from_projection(value: object, *, ordinal: int) -> HostAttempt:
|
|
projection = _mapping(value, name="host attempt")
|
|
_exact(projection, _ATTEMPT_FIELDS, name="host attempt")
|
|
attempt = HostAttempt(
|
|
ordinal=cast(int, projection["ordinal"]),
|
|
verdict=cast(str, projection["verdict"]),
|
|
observed_status=cast(str, projection["observed_status"]),
|
|
reason_code=cast(str | None, projection["reason_code"]),
|
|
result_fingerprint=cast(str, projection["result_fingerprint"]),
|
|
)
|
|
if attempt.ordinal != ordinal:
|
|
raise ValueError("host attempt ordinal does not match its position")
|
|
return attempt
|
|
|
|
|
|
def build_host_attempt_transition(
|
|
*,
|
|
lane: str,
|
|
first: HostAttempt,
|
|
repeat: HostAttempt,
|
|
) -> dict[str, object]:
|
|
"""Build first-attempt/repeat evidence without retry masking."""
|
|
|
|
if not isinstance(lane, str) or not _LANE_PATTERN.fullmatch(lane):
|
|
raise ValueError("host attempt lane is invalid")
|
|
if not isinstance(first, HostAttempt) or not isinstance(repeat, HostAttempt):
|
|
raise TypeError("host attempts must use HostAttempt")
|
|
if first.ordinal != 1 or repeat.ordinal != 2:
|
|
raise ValueError("host attempts must retain first/repeat order")
|
|
unchanged = (
|
|
first.result_fingerprint == repeat.result_fingerprint
|
|
and first.observed_status == repeat.observed_status
|
|
and first.reason_code == repeat.reason_code
|
|
)
|
|
transition = f"{first.verdict}_to_{repeat.verdict}"
|
|
if first.verdict == repeat.verdict == "pass" and not unchanged:
|
|
transition = "pass_to_pass_changed"
|
|
accepted = first.verdict == repeat.verdict == "pass" and unchanged
|
|
return {
|
|
"accepted": accepted,
|
|
"first": first.projection(),
|
|
"lane": lane,
|
|
"repeat": repeat.projection(),
|
|
"schema": HOST_ATTEMPT_SCHEMA,
|
|
"transition": transition,
|
|
}
|
|
|
|
|
|
def validate_host_attempt_transition(value: object) -> dict[str, object]:
|
|
"""Validate one host attempt transition."""
|
|
|
|
projection = _mapping(value, name="host attempt transition")
|
|
_exact(projection, _TRANSITION_FIELDS, name="host attempt transition")
|
|
if projection.get("schema") != HOST_ATTEMPT_SCHEMA:
|
|
raise ValueError("host attempt transition schema is unsupported")
|
|
rebuilt = build_host_attempt_transition(
|
|
lane=cast(str, projection["lane"]),
|
|
first=_attempt_from_projection(projection["first"], ordinal=1),
|
|
repeat=_attempt_from_projection(projection["repeat"], ordinal=2),
|
|
)
|
|
if projection != rebuilt:
|
|
raise ValueError("host attempt transition evidence is inconsistent")
|
|
return cast(dict[str, object], value)
|
|
|
|
|
|
def _decision_projection(decision: CompatibilityDecision) -> dict[str, object]:
|
|
if decision.considered != tuple(CapabilityDimension):
|
|
raise ValueError("compatibility dimensions are incomplete or reordered")
|
|
return {
|
|
"considered": [item.value for item in decision.considered],
|
|
"level": decision.level.value,
|
|
"reasons": [item.value for item in decision.reasons],
|
|
}
|
|
|
|
|
|
def _capability_case(
|
|
identifier: str,
|
|
*,
|
|
model: ModelCapabilities | None = None,
|
|
profile: ProfileCapabilities | None = None,
|
|
sampler: SamplerCapabilities | None = None,
|
|
request: ExecutionFeatureRequest | None = None,
|
|
) -> dict[str, object]:
|
|
effective_model = model or KREA2_TURBO_SCHEMA.model_capabilities
|
|
effective_profile = profile or KREA2_TURBO_SCHEMA.profile_capabilities
|
|
effective_sampler = sampler or KREA2_TURBO_SCHEMA.reference_sampler_capabilities
|
|
effective_request = request or ExecutionFeatureRequest()
|
|
decision = evaluate_compatibility(
|
|
model=effective_model,
|
|
profile=effective_profile,
|
|
sampler=effective_sampler,
|
|
request=effective_request,
|
|
)
|
|
return {
|
|
"decision": _decision_projection(decision),
|
|
"id": identifier,
|
|
"semantics": {
|
|
"execution_behavior": effective_sampler.execution_behavior.value,
|
|
"noise_ownership": effective_sampler.noise_ownership.value,
|
|
"partial_denoise_requested": effective_request.use_partial_denoise,
|
|
"per_token_timesteps_requested": effective_request.use_per_token_timesteps,
|
|
"required_state": [item.value for item in effective_sampler.required_state],
|
|
"sampler_id": effective_sampler.sampler_id,
|
|
"schedule_ownership": effective_profile.ownership.name.casefold(),
|
|
"terminal_requirement": effective_sampler.terminal_requirement.value,
|
|
},
|
|
}
|
|
|
|
|
|
def _capability_cases() -> list[dict[str, object]]:
|
|
model = KREA2_TURBO_SCHEMA.model_capabilities
|
|
sampler = KREA2_TURBO_SCHEMA.reference_sampler_capabilities
|
|
return sorted(
|
|
[
|
|
_capability_case(
|
|
"model_owned_sigmas_reject",
|
|
model=replace(
|
|
model,
|
|
accepted_ownerships=(ScheduleOwnership.MODEL_NATIVE,),
|
|
),
|
|
),
|
|
_capability_case("native_euler_allow"),
|
|
_capability_case(
|
|
"nonreference_euler_warn",
|
|
sampler=replace(sampler, sampler_id="fixture.nonreference-euler"),
|
|
),
|
|
_capability_case(
|
|
"partial_denoise_reject",
|
|
sampler=replace(sampler, supports_partial_denoise=False),
|
|
request=ExecutionFeatureRequest(use_partial_denoise=True),
|
|
),
|
|
_capability_case(
|
|
"per_token_timesteps_reject",
|
|
request=ExecutionFeatureRequest(use_per_token_timesteps=True),
|
|
),
|
|
_capability_case(
|
|
"resume_state_reject",
|
|
sampler=replace(sampler, required_state=(SamplerState.RESUME,)),
|
|
),
|
|
_capability_case(
|
|
"stochastic_noise_reject",
|
|
sampler=replace(
|
|
sampler,
|
|
execution_behavior=ExecutionBehavior.STOCHASTIC,
|
|
noise_ownership=NoiseOwnership.SAMPLER,
|
|
),
|
|
),
|
|
_capability_case(
|
|
"terminal_zero_reject",
|
|
sampler=replace(
|
|
sampler,
|
|
terminal_requirement=TerminalRequirement.FORBIDS_ZERO,
|
|
),
|
|
),
|
|
],
|
|
key=lambda item: cast(str, item["id"]),
|
|
)
|
|
|
|
|
|
def _sha256(payload: bytes) -> str:
|
|
return "sha256:" + hashlib.sha256(payload).hexdigest()
|
|
|
|
|
|
def _component(
|
|
*,
|
|
identifier: str,
|
|
version: str,
|
|
projection: Mapping[str, object],
|
|
) -> ExecutionComponent:
|
|
return ExecutionComponent(
|
|
identifier=identifier,
|
|
version=version,
|
|
fingerprint=_sha256(canonical_projection_bytes(dict(projection))),
|
|
)
|
|
|
|
|
|
def _portable_execution() -> dict[str, object]:
|
|
decision = evaluate_compatibility(
|
|
model=KREA2_TURBO_SCHEMA.model_capabilities,
|
|
profile=KREA2_TURBO_SCHEMA.profile_capabilities,
|
|
sampler=KREA2_TURBO_SCHEMA.reference_sampler_capabilities,
|
|
request=ExecutionFeatureRequest(),
|
|
)
|
|
schedule = build_krea2_turbo_schedule(steps=8, width=1024, height=1024)
|
|
artifact = build_schedule_artifact(
|
|
schedule,
|
|
metadata=ArtifactBuildMetadata(
|
|
source_id=KREA2_TURBO_SCHEMA.primary_source_id,
|
|
source_label=KREA2_TURBO_SCHEMA.display_name,
|
|
base_grid_parameters=(ArtifactField(name="steps", value=8),),
|
|
transform_parameters=(
|
|
(
|
|
ArtifactField(
|
|
name="mu",
|
|
value=TypedArtifactValue(
|
|
value=KREA2_TURBO_PROFILE.fixed_mu,
|
|
precision="float64",
|
|
),
|
|
),
|
|
),
|
|
(),
|
|
),
|
|
compatibility=(
|
|
ArtifactField(name="decision", value=decision.level.value),
|
|
ArtifactField(name="double_shift", value=False),
|
|
ArtifactField(
|
|
name="reference_sampler",
|
|
value=KREA2_TURBO_SCHEMA.reference_sampler_capabilities.sampler_id,
|
|
),
|
|
),
|
|
),
|
|
precision="float64",
|
|
)
|
|
receipt = build_execution_receipt(
|
|
artifact,
|
|
metadata=ExecutionReceiptMetadata(
|
|
compatibility=decision,
|
|
host=ExecutionHost(
|
|
identifier="comfyui",
|
|
version=CANONICAL_HOST_VERSION,
|
|
revision=CANONICAL_HOST_REVISION,
|
|
api_version="legacy_v1",
|
|
),
|
|
model=_component(
|
|
identifier="sigmax.controlled-flow-model",
|
|
version="1",
|
|
projection={"fixture": "m7-07", "profile": KREA2_TURBO_SCHEMA.profile_id},
|
|
),
|
|
sampler=_component(
|
|
identifier=KREA2_TURBO_SCHEMA.reference_sampler_capabilities.sampler_id,
|
|
version=CANONICAL_HOST_VERSION,
|
|
projection={
|
|
"fixture": "m7-07",
|
|
"sampler": KREA2_TURBO_SCHEMA.reference_sampler_capabilities.sampler_id,
|
|
},
|
|
),
|
|
rng_ownership=ExecutionRngOwnership(
|
|
schedule=NoiseOwnership.NONE,
|
|
model=NoiseOwnership.NONE,
|
|
sampler=NoiseOwnership.NONE,
|
|
),
|
|
requested_transitions=8,
|
|
effective_transitions=8,
|
|
requested_model_evaluations=8,
|
|
effective_model_evaluations=8,
|
|
status=ExecutionStatus.SUCCEEDED,
|
|
),
|
|
)
|
|
bundle = PortableExecutionBundle(artifact=artifact, receipt=receipt)
|
|
artifact_payload = serialize_schedule_artifact(artifact)
|
|
receipt_payload = serialize_execution_receipt(receipt)
|
|
bundle_payload = serialize_portable_execution_bundle(bundle)
|
|
artifact_round_trip = deserialize_schedule_artifact(artifact_payload)
|
|
receipt_round_trip = deserialize_execution_receipt(receipt_payload)
|
|
bundle_round_trip = deserialize_portable_execution_bundle(bundle_payload)
|
|
construction = artifact.construction_projection()
|
|
transforms = cast(list[dict[str, object]], construction["transforms"])
|
|
shift_count = sum(item["id"] == "krea.exponential_mu" for item in transforms)
|
|
receipt_projection = receipt.projection()
|
|
effective = _mapping(receipt_projection["effective_inputs"], name="effective inputs")
|
|
return {
|
|
"artifact": {
|
|
"construction_fingerprint": artifact.construction_fingerprint,
|
|
"numerical_fingerprint": artifact.numerical_fingerprint,
|
|
},
|
|
"artifact_round_trip": artifact_round_trip == artifact,
|
|
"bundle_fingerprint": _sha256(bundle_payload),
|
|
"bundle_round_trip": (
|
|
bundle_round_trip == bundle
|
|
and serialize_portable_execution_bundle(bundle_round_trip) == bundle_payload
|
|
),
|
|
"counts": receipt_projection["counts"],
|
|
"double_shift": False,
|
|
"effective_inputs": {
|
|
field: effective[field]
|
|
for field in (
|
|
"height",
|
|
"precision",
|
|
"profile",
|
|
"profile_version",
|
|
"steps",
|
|
"width",
|
|
)
|
|
},
|
|
"receipt_fingerprint": receipt.receipt_fingerprint,
|
|
"receipt_round_trip": (
|
|
receipt_round_trip == receipt
|
|
and serialize_execution_receipt(receipt_round_trip) == receipt_payload
|
|
),
|
|
"rng_ownership": receipt_projection["rng_ownership"],
|
|
"shift_count": shift_count,
|
|
"status": _mapping(receipt_projection["execution"], name="execution")["status"],
|
|
}
|
|
|
|
|
|
def _fingerprint_projection(value: Mapping[str, object]) -> str:
|
|
return _sha256(canonical_projection_bytes(value))
|
|
|
|
|
|
def _host_attempt_regressions() -> list[dict[str, object]]:
|
|
stable_rejection = _fingerprint_projection(
|
|
{"reason_code": "execution.partial_denoise_unsupported", "status": "error"}
|
|
)
|
|
return [
|
|
{
|
|
"id": "first_failure_then_pass",
|
|
**build_host_attempt_transition(
|
|
lane="H3_EULER_M5_01",
|
|
first=HostAttempt(
|
|
ordinal=1,
|
|
verdict="fail",
|
|
observed_status="error",
|
|
reason_code="host.first_attempt_failure",
|
|
result_fingerprint=_fingerprint_projection({"attempt": 1, "status": "error"}),
|
|
),
|
|
repeat=HostAttempt(
|
|
ordinal=2,
|
|
verdict="pass",
|
|
observed_status="succeeded",
|
|
reason_code=None,
|
|
result_fingerprint=_fingerprint_projection(
|
|
{"attempt": 2, "status": "succeeded"}
|
|
),
|
|
),
|
|
),
|
|
},
|
|
{
|
|
"id": "repeat_result_drift",
|
|
**build_host_attempt_transition(
|
|
lane="H2_TURBO_M2_05",
|
|
first=HostAttempt(
|
|
ordinal=1,
|
|
verdict="pass",
|
|
observed_status="not_executed",
|
|
reason_code=None,
|
|
result_fingerprint=_fingerprint_projection({"schedule": "first"}),
|
|
),
|
|
repeat=HostAttempt(
|
|
ordinal=2,
|
|
verdict="pass",
|
|
observed_status="not_executed",
|
|
reason_code=None,
|
|
result_fingerprint=_fingerprint_projection({"schedule": "changed"}),
|
|
),
|
|
),
|
|
},
|
|
{
|
|
"id": "stable_expected_rejection",
|
|
**build_host_attempt_transition(
|
|
lane="H3_EULER_M5_01",
|
|
first=HostAttempt(
|
|
ordinal=1,
|
|
verdict="pass",
|
|
observed_status="error",
|
|
reason_code="execution.partial_denoise_unsupported",
|
|
result_fingerprint=stable_rejection,
|
|
),
|
|
repeat=HostAttempt(
|
|
ordinal=2,
|
|
verdict="pass",
|
|
observed_status="error",
|
|
reason_code="execution.partial_denoise_unsupported",
|
|
result_fingerprint=stable_rejection,
|
|
),
|
|
),
|
|
},
|
|
]
|
|
|
|
|
|
def _build_unchecked() -> dict[str, object]:
|
|
return {
|
|
"cases": _capability_cases(),
|
|
"coverage": {
|
|
"capabilities": [
|
|
"deterministic_native_euler",
|
|
"effective_model_evaluations",
|
|
"noise_ownership",
|
|
"partial_denoise",
|
|
"per_token_timesteps",
|
|
"resume",
|
|
"rng_separation",
|
|
"sampler_state",
|
|
"schedule_ownership",
|
|
"stochastic_execution",
|
|
"terminal_sigma",
|
|
],
|
|
"levels": ["allow", "warn", "reject"],
|
|
"skipped": [],
|
|
"unsupported": [
|
|
"advanced_workflows",
|
|
"partial_denoise_execution",
|
|
"resume_execution",
|
|
"stochastic_euler_execution",
|
|
],
|
|
},
|
|
"host_attempt_regressions": _host_attempt_regressions(),
|
|
"portable_execution": _portable_execution(),
|
|
"profile": {
|
|
"fingerprint": profile_schema_fingerprint(KREA2_TURBO_SCHEMA),
|
|
"id": KREA2_TURBO_SCHEMA.profile_id,
|
|
"version": KREA2_TURBO_SCHEMA.profile_version,
|
|
},
|
|
"schema": REPORT_SCHEMA,
|
|
"status": "PASS",
|
|
}
|
|
|
|
|
|
def build_conformance_report() -> dict[str, object]:
|
|
"""Build canonical M7-07 conformance evidence from production contracts."""
|
|
|
|
return _build_unchecked()
|
|
|
|
|
|
def validate_conformance_report(value: object) -> dict[str, object]:
|
|
"""Rebuild and validate the complete conformance report."""
|
|
|
|
report = _mapping(value, name="conformance report")
|
|
_exact(report, _ROOT_FIELDS, name="conformance report")
|
|
if report != _build_unchecked():
|
|
raise ValueError("conformance report differs from recomputed production contracts")
|
|
return cast(dict[str, object], value)
|
|
|
|
|
|
def canonical_json(value: object) -> str:
|
|
"""Serialize validated evidence deterministically."""
|
|
|
|
report = validate_conformance_report(value)
|
|
return canonical_projection_bytes(report).decode("utf-8") + "\n"
|