From 2d5bf50880d09914a2cae9edd044b9d2a72ea256 Mon Sep 17 00:00:00 2001
From: halfaipg <justin@techtoaster.com>
Date: Tue, 20 Oct 2026 18:20:07 -0400
Subject: [PATCH 1/5] feat(video): prepare offline FastH3 execution contracts

Keep FastH3 outside live recipe loading, pricing and admission while preparing exact frame units and distinct image/audio source roles. Model-use rights, native qualification and the worker/billing/UI rollout remain separate gates.
---
 grid_api/services/AGENTS.md                   |   9 +
 grid_api/services/fasth3_contract.py          | 185 +++++++++++++
 .../services/tests/test_fasth3_contract.py    | 243 ++++++++++++++++++
 recipes/AGENTS.md                             |   2 +-
 recipes/qualification/AGENTS.md               |   6 +-
 recipes/qualification/fasth3/AGENTS.md        |  31 +++
 recipes/qualification/fasth3/README.md        |  68 +++++
 .../fasth3/omni-nvfp4-qad-v1.json             |  92 +++++++
 8 files changed, 634 insertions(+), 2 deletions(-)
 create mode 100644 grid_api/services/fasth3_contract.py
 create mode 100644 grid_api/services/tests/test_fasth3_contract.py
 create mode 100644 recipes/qualification/fasth3/AGENTS.md
 create mode 100644 recipes/qualification/fasth3/README.md
 create mode 100644 recipes/qualification/fasth3/omni-nvfp4-qad-v1.json

diff --git a/grid_api/services/AGENTS.md b/grid_api/services/AGENTS.md
index cf6834a7..2f3ce1a3 100644
--- a/grid_api/services/AGENTS.md
+++ b/grid_api/services/AGENTS.md
@@ -7,6 +7,15 @@ content sanitization, and reward settlement.
 
 ## Ownership
 
+- `fasth3_contract.py` owns an offline-only FastH3 qualification compiler,
+  exact17k+5frame units, integer quote math and separate appearance/opening/
+  conditioning-audio/preserved-soundtrack slots. Its fixed template lives in
+  `recipes/qualification/fasth3/`; the live registry rejects its schema and no
+  HTTP, price-book or worker-advertisement path calls it. Candidate bounds are
+  not native capacity/quality. Model-use rights, source protocol, purchased
+  billing and public rollout remain unresolved release gates. Tests prove
+  startup/public-async/unpriced-money denial, not license or native execution.
+
 - Live book `2026-10-20-a` adds the owner-approved `MOSS Image SFT BF16`
   image-only launch peg of USD0.010/output on Core0d2ee0b7.
   Its exact model-specific defaults are50steps/CFG3/Euler; other model defaults
diff --git a/grid_api/services/fasth3_contract.py b/grid_api/services/fasth3_contract.py
new file mode 100644
index 00000000..f53ad7d0
--- /dev/null
+++ b/grid_api/services/fasth3_contract.py
@@ -0,0 +1,185 @@
+# SPDX-FileCopyrightText: 2026 AI Power Grid
+# SPDX-License-Identifier: AGPL-3.0-or-later
+
+"""Offline FastH3 qualification compiler, deliberately outside live recipe admission.
+
+The live registry does not parse this schema. No model registration, pricing-book
+entry, credit authorization, source decoding, storage or network request occurs.
+"""
+
+import copy
+import json
+import math
+import re
+from fractions import Fraction
+from pathlib import Path
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
+
+MODEL = "FastH3 Omni NVFP4 QAD"
+CANDIDATE = Path(__file__).resolve().parents[2] / "recipes/qualification/fasth3/omni-nvfp4-qad-v1.json"
+
+
+class FastH3Contract(BaseModel):
+    model_config = ConfigDict(extra="forbid", strict=True, frozen=True)
+
+    contract_schema: Literal["aipg.fasth3-video.v1"] = Field(alias="schema")
+    variant: Literal["omni-nvfp4-qad"]
+    min_frames: int = Field(ge=124, le=362)
+    max_frames: int = Field(ge=124, le=362)
+    canvases: list[Literal["1344x768", "768x1344", "768x768"]] = Field(min_length=1, max_length=3)
+
+    @model_validator(mode="after")
+    def validate_matrix(self):
+        if (
+            self.min_frames > self.max_frames
+            or self.min_frames % 17 != 5
+            or self.max_frames % 17 != 5
+            or len(set(self.canvases)) != len(self.canvases)
+        ):
+            raise ValueError("invalid FastH3 candidate matrix")
+        return self
+
+
+class FastH3Inputs(BaseModel):
+    model_config = ConfigDict(extra="forbid", strict=True, frozen=True, allow_inf_nan=False)
+
+    prompt: str = Field(default="", max_length=8000)
+    seed: int = Field(default=29, ge=0, le=2**53 - 1)
+    frames: int = Field(default=124, ge=124, le=362)
+    seconds: float | None = Field(default=None, gt=0, le=362 / 24)
+    width: int = Field(default=1344)
+    height: int = Field(default=768)
+    fps: Literal[24] = 24
+    steps: Literal[8] = 8
+    cfg: Literal[1.0] = 1.0
+    sampler: Literal["res_multistep"] = "res_multistep"
+    scheduler: Literal["simple"] = "simple"
+    audio: Literal[True] = True
+    appearance_image: bool = False
+    opening_image: bool = False
+    audio_mode: Literal["generated", "reference", "preserve"] = "generated"
+
+    @field_validator("fps", "steps", "audio", mode="before")
+    @classmethod
+    def primitive_literals(cls, value, info):
+        if type(value) is not (bool if info.field_name == "audio" else int):
+            raise ValueError("invalid primitive type")
+        return value
+
+    @field_validator("cfg", mode="before")
+    @classmethod
+    def numeric_cfg(cls, value):
+        if type(value) not in (int, float):
+            raise ValueError("CFG must be numeric, not boolean")
+        return value
+
+    @model_validator(mode="after")
+    def validate_roles(self):
+        if not self.prompt.strip() or len(self.prompt.encode()) > 8000:
+            raise ValueError("invalid prompt")
+        if self.frames % 17 != 5:
+            raise ValueError("frames must be exact 17k+5; no silent rounding")
+        if self.seconds is not None and not math.isclose(
+            self.seconds,
+            self.frames / 24,
+            rel_tol=0,
+            abs_tol=1e-9,
+        ):
+            raise ValueError("seconds disagree with exact frames")
+        if self.audio_mode != "generated" and not (self.appearance_image or self.opening_image):
+            raise ValueError("audio candidates require an explicit image")
+        available = set()
+        if self.appearance_image:
+            available.add(("Picture", "1"))
+        if self.audio_mode == "reference":
+            available.add(("Audio", "1"))
+        if any(label not in available for label in re.findall(r"<(Picture|Audio|Video)\s*(\d+)>", self.prompt)):
+            raise ValueError("prompt refers to an absent reference slot")
+        return self
+
+
+def normalize(contract: FastH3Contract, values: dict) -> dict:
+    """Return exact candidate units, never a price, live capability or permission."""
+    try:
+        contract = FastH3Contract.model_validate(contract.model_dump(by_alias=True))
+        inputs = FastH3Inputs.model_validate(values)
+        if not contract.min_frames <= inputs.frames <= contract.max_frames or f"{inputs.width}x{inputs.height}" not in contract.canvases:
+            raise ValueError("outside candidate matrix")
+    except (ValueError, TypeError, AttributeError):
+        raise ValueError("Invalid FastH3 qualification inputs") from None
+    duration = Fraction(inputs.frames, 24)
+    return {
+        **inputs.model_dump(exclude_none=True),
+        "seconds": float(duration),
+        "duration_numerator": duration.numerator,
+        "duration_denominator": duration.denominator,
+    }
+
+
+def quote_units(frames: int, micro_usd_per_second: int) -> int:
+    """Offline integer ceiling for a supplied server rate; no tariff is approved here."""
+    if (
+        type(frames) is not int
+        or not 124 <= frames <= 362
+        or frames % 17 != 5
+        or type(micro_usd_per_second) is not int
+        or micro_usd_per_second <= 0
+    ):
+        raise ValueError("invalid FastH3 quote units")
+    return (frames * micro_usd_per_second + 23) // 24
+
+
+def compile_candidate(values: dict) -> dict:
+    """Materialize one Core-owned candidate and explicit unbound source slots.
+
+    Source bytes must be independently validated and committed by a future H3
+    source adapter before execution. Placeholder filenames are not input assets.
+    This is intentionally not called by recipes.resolve or public HTTP routes.
+    """
+    workflow = json.loads(CANDIDATE.read_text())
+    metadata = workflow["_grid"]
+    if (
+        metadata["modelName"] != MODEL
+        or metadata.get("qualificationOnly") is not True
+        or metadata.get("engine") != "comfyui"
+        or metadata.get("jobType") != "video"
+    ):
+        raise ValueError("invalid FastH3 qualification metadata")
+    contract = FastH3Contract.model_validate(metadata["videoContract"])
+    execution = normalize(contract, values)
+    graph = copy.deepcopy({key: node for key, node in workflow.items() if key != "_grid"})
+    graph["conditioning"]["inputs"].update(
+        {
+            "prompt": execution["prompt"],
+            "width": execution["width"],
+            "height": execution["height"],
+            "length": execution["frames"],
+        },
+    )
+    graph["noise"]["inputs"]["noise_seed"] = execution["seed"]
+    bindings = []
+    if execution["appearance_image"]:
+        graph["appearance"] = {"class_type": "LoadImage", "inputs": {"image": "grid_appearance.png"}}
+        graph["conditioning"]["inputs"]["ref_images.ref_image_0"] = ["appearance", 0]
+        bindings.append({"role": "appearance_image", "path": "appearance.inputs.image"})
+    if execution["audio_mode"] == "reference":
+        graph["reference_sound"] = {"class_type": "LoadAudio", "inputs": {"audio": "grid_reference.wav"}}
+        graph["conditioning"]["inputs"]["ref_audios.ref_audio_0"] = ["reference_sound", 0]
+        bindings.append({"role": "conditioning_audio", "path": "reference_sound.inputs.audio"})
+    if execution["opening_image"] or execution["audio_mode"] == "preserve":
+        guide = {"positive": ["conditioning", 0], "latent": ["conditioning", 1], "frame_idx": 0}
+        if execution["opening_image"]:
+            graph["opening"] = {"class_type": "LoadImage", "inputs": {"image": "grid_opening.png"}}
+            guide.update(image=["opening", 0], vae=["video_vae", 0])
+            bindings.append({"role": "opening_image", "path": "opening.inputs.image"})
+        if execution["audio_mode"] == "preserve":
+            graph["soundtrack"] = {"class_type": "LoadAudio", "inputs": {"audio": "grid_soundtrack.wav"}}
+            guide.update(audio=["soundtrack", 0], audio_vae=["audio_vae", 0])
+            graph["video"]["inputs"]["audio"] = ["soundtrack", 0]
+            del graph["decode_audio"]
+            bindings.append({"role": "soundtrack", "path": "soundtrack.inputs.audio"})
+        graph["guide"] = {"class_type": "MiniMaxH3AddGuide", "inputs": guide}
+        graph["guider"]["inputs"]["conditioning"] = ["guide", 0]
+    return {"model": MODEL, "execution": execution, "spec": graph, "source_bindings": bindings, "qualification_only": True}
diff --git a/grid_api/services/tests/test_fasth3_contract.py b/grid_api/services/tests/test_fasth3_contract.py
new file mode 100644
index 00000000..ca659a05
--- /dev/null
+++ b/grid_api/services/tests/test_fasth3_contract.py
@@ -0,0 +1,243 @@
+# SPDX-FileCopyrightText: 2026 AI Power Grid
+# SPDX-License-Identifier: AGPL-3.0-or-later
+
+"""Offline H3 contract checks, not GPU, license, paid-billing or deployment proof."""
+
+import copy
+import itertools
+import json
+from fractions import Fraction
+from pathlib import Path
+from uuid import uuid4
+
+import pytest
+from fastapi import HTTPException
+from pydantic import ValidationError
+
+from grid_api.config import GridSettings
+from grid_api.services import credits, generation_admission, pricing, recipes
+from grid_api.services import fasth3_contract as h3
+
+
+@pytest.fixture
+def registry(monkeypatch):
+    for name in ("_BY_ROOT", "_BY_NAME", "_BY_ID", "_BY_MODEL", "_LOCAL_RECIPES", "_ONCHAIN_RECIPES"):
+        monkeypatch.setattr(recipes, name, {})
+
+
+def candidate(**values):
+    return h3.compile_candidate({"prompt": "A sunny park.", **values})
+
+
+def test_startup_catalog_never_loads_or_advertises_candidate(registry):
+    root = Path(__file__).resolve().parents[3]
+    assert recipes.load_local_recipes(str(root / "recipes")) > 0
+    assert recipes.get_recipe(h3.MODEL) is None
+    assert recipes.video_contract_for_model(h3.MODEL) is None
+    assert recipes.generation_modes(h3.MODEL) == []
+    assert recipes.param_schema(h3.MODEL) is None
+    assert h3.MODEL not in {recipe.model_name for recipe in recipes.list_recipes()}
+    assert pricing.get_price(h3.MODEL) is None
+    # This low-level helper returns zero for a missing rate; enforcement must
+    # distinguish unpriced from free before reserving or dispatching work.
+    assert pricing.quote_video(h3.MODEL, 124 / 24) == 0
+    assert not pricing.is_priced_for(h3.MODEL, "video")
+
+
+def test_unreviewed_schema_cannot_register_as_live_ltx(registry):
+    workflow = json.loads(h3.CANDIDATE.read_text())
+    with pytest.raises(ValidationError):
+        recipes.register_recipe("0xunqualified-h3", h3.MODEL, workflow)
+    assert recipes.list_recipes() == []
+
+
+@pytest.mark.parametrize("source", [False, True])
+def test_open_public_async_switch_does_not_enable_h3(source):
+    settings = GridSettings(
+        _env_file=None,
+        media_async_enabled=True,
+        media_async_admission_mode="open",
+        media_async_nvfp4_enabled=True,
+        media_async_source_admission_mode="open",
+    )
+    with pytest.raises(HTTPException) as error:
+        generation_admission.require_async_video(settings, uuid4(), model=h3.MODEL, has_source=source)
+    assert error.value.status_code == 503
+
+
+def test_fixed_sampling_and_component_provenance():
+    result = candidate(appearance_image=True, seed=0)
+    graph = result["spec"]
+    meta = json.loads(h3.CANDIDATE.read_text())["_grid"]
+    assert result["qualification_only"] is True
+    assert result["model"] == h3.MODEL
+    assert graph["model"]["inputs"]["unet_name"] == "fasth3_omniref_qad_nvfp4_comfy.safetensors"
+    assert graph["encoder"]["inputs"]["type"] == "minimax"
+    assert graph["conditioning"]["inputs"]["ref_image_size"] == "match"
+    assert graph["noise"]["inputs"]["noise_seed"] == 0
+    assert graph["sampler"]["inputs"]["sampler_name"] == "res_multistep"
+    assert graph["sigmas"]["inputs"]["steps"] == 8
+    assert graph["sigmas"]["inputs"]["scheduler"] == "simple"
+    assert graph["guider"]["class_type"] == "BasicGuider"
+    assert graph["shift"]["inputs"] == {"model": ["model", 0], "shift_video": 12.0, "shift_audio": 3.0}
+    assert meta["provenance"]["model_revision"] == "b1b4d67d79930007a3f73ca8e98739796ded95a0"
+    assert meta["provenance"]["comfy_revision"] == "b1a247c9cf9dbfda4b2c0a0c1d50950667eeaa0c"
+    assert meta["provenance"]["license_sha256"] == "59b99642b95ea21630e311198ddbfffbfe05aadba0c2f5d884cbdf4efcc90f44"
+    for node_name, field in (("model", "unet_name"), ("encoder", "clip_name"), ("video_vae", "vae_name"), ("audio_vae", "vae_name")):
+        filename = graph[node_name]["inputs"][field]
+        assert len(meta["provenance"]["components"][filename]) == 64
+
+
+@pytest.mark.parametrize(
+    "appearance,opening,audio_mode",
+    list(itertools.product([False, True], [False, True], ["generated", "reference", "preserve"])),
+)
+def test_all_explicit_source_combinations_have_exact_slots(appearance, opening, audio_mode):
+    values = {"appearance_image": appearance, "opening_image": opening, "audio_mode": audio_mode}
+    if not appearance and not opening and audio_mode != "generated":
+        with pytest.raises(ValueError, match="Invalid FastH3"):
+            candidate(**values)
+        return
+    result = candidate(**values)
+    graph = result["spec"]
+    roles = [binding["role"] for binding in result["source_bindings"]]
+    assert ("appearance_image" in roles) == appearance
+    assert ("opening_image" in roles) == opening
+    assert ("conditioning_audio" in roles) == (audio_mode == "reference")
+    assert ("soundtrack" in roles) == (audio_mode == "preserve")
+    for binding in result["source_bindings"]:
+        node, inputs, field = binding["path"].split(".")
+        assert isinstance(graph[node][inputs][field], str)
+    for node in graph.values():
+        for value in node["inputs"].values():
+            if isinstance(value, list):
+                assert value[0] in graph
+    conditioning = graph["conditioning"]["inputs"]
+    assert ("ref_images.ref_image_0" in conditioning) == appearance
+    assert ("ref_audios.ref_audio_0" in conditioning) == (audio_mode == "reference")
+    assert ("guide" in graph) == (opening or audio_mode == "preserve")
+    if opening:
+        assert graph["guide"]["inputs"]["image"] == ["opening", 0]
+    if audio_mode == "preserve":
+        assert graph["guide"]["inputs"]["audio"] == ["soundtrack", 0]
+        assert graph["video"]["inputs"]["audio"] == ["soundtrack", 0]
+        assert "decode_audio" not in graph
+    else:
+        assert graph["video"]["inputs"]["audio"] == ["decode_audio", 0]
+
+
+@pytest.mark.parametrize("frames", range(123, 364))
+def test_exact_h3_grid_never_rounds_ltx_counts_or_underquotes(frames):
+    if 124 <= frames <= 362 and frames % 17 == 5:
+        result = candidate(frames=frames)
+        execution = result["execution"]
+        assert result["spec"]["conditioning"]["inputs"]["length"] == frames
+        assert Fraction(execution["duration_numerator"], execution["duration_denominator"]) == Fraction(frames, 24)
+        for rate in (1, 1999, 10000, 20000):  # Synthetic rates, not approved prices.
+            amount = h3.quote_units(frames, rate)
+            exact = Fraction(frames * rate, 24)
+            assert amount - 1 < exact <= amount
+    else:
+        with pytest.raises(ValueError):
+            candidate(frames=frames)
+        with pytest.raises(ValueError):
+            h3.quote_units(frames, 10000)
+
+
+@pytest.mark.parametrize("size", [(1344, 768), (768, 1344), (768, 768)])
+def test_qualification_canvases_are_exact_not_auto_fit(size):
+    result = candidate(width=size[0], height=size[1])
+    assert result["execution"]["width"] == size[0]
+    assert result["spec"]["conditioning"]["inputs"]["height"] == size[1]
+
+
+@pytest.mark.parametrize(
+    "changes",
+    [
+        {"frames": True},
+        {"frames": "124"},
+        {"frames": 125},
+        {"frames": 625},
+        {"seconds": 5.0},
+        {"seconds": float("nan")},
+        {"seconds": float("inf")},
+        {"width": True},
+        {"width": 1344.0},
+        {"width": 512},
+        {"height": 512},
+        {"seed": True},
+        {"seed": -1},
+        {"seed": 2**53},
+        {"seed": "29"},
+        {"steps": 7},
+        {"steps": 8.0},
+        {"cfg": True},
+        {"cfg": 2.0},
+        {"fps": 24.0},
+        {"audio": 1},
+        {"audio": False},
+        {"sampler": "euler"},
+        {"scheduler": "normal"},
+        {"precision": "bf16"},
+        {"reference_video": "x.mp4"},
+        {"appearance_image": "x.png"},
+        {"opening_image": 1},
+        {"audio_mode": "audio"},
+        {"prompt": " "},
+        {"prompt": "x" * 8001},
+        {"prompt": "\u00e9" * 4001},
+        {"prompt": "<Picture 1>"},
+        {"prompt": "<Audio 1>"},
+        {"prompt": "<Video 1>"},
+        {"prompt": "<Picture 2>", "appearance_image": True},
+        {"prompt": "<Audio 1>", "appearance_image": True, "audio_mode": "preserve"},
+    ],
+)
+def test_invalid_inputs_reject_without_reflecting_private_values(changes):
+    with pytest.raises(ValueError) as error:
+        candidate(**changes)
+    assert str(error.value) == "Invalid FastH3 qualification inputs"
+
+
+@pytest.mark.parametrize("rate", [True, 0, -1, 1.0, "10000", None])
+def test_quote_rate_must_be_a_positive_server_integer(rate):
+    with pytest.raises(ValueError):
+        h3.quote_units(124, rate)
+
+
+@pytest.mark.asyncio
+async def test_unpriced_h3_cannot_reach_credit_writes(monkeypatch):
+    from grid_api import database
+
+    def forbidden(*_args, **_kwargs):
+        pytest.fail("an unpriced candidate must not open a database transaction")
+
+    monkeypatch.setattr(credits, "charging_enabled_for", lambda *_args: True)
+    monkeypatch.setattr(credits, "_economic_alert", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(database, "new_session", forbidden)
+    result = await credits.authorize_media(uuid4(), h3.MODEL, "video", 1, 124 / 24, uuid4())
+    assert result["ok"] is False
+    assert result["status"] == "unpriced" and result["reserved"] == 0
+
+
+def test_changed_contract_cannot_bypass_matrix_by_model_copy():
+    contract = h3.FastH3Contract.model_validate(json.loads(h3.CANDIDATE.read_text())["_grid"]["videoContract"])
+    invalid = contract.model_copy(update={"max_frames": 625})
+    with pytest.raises(ValueError, match="Invalid FastH3"):
+        h3.normalize(invalid, {"prompt": "Park.", "frames": 124})
+    smaller = contract.model_copy(update={"max_frames": 124, "canvases": ["1344x768"]})
+    with pytest.raises(ValueError):
+        h3.normalize(smaller, {"prompt": "Park.", "frames": 141})
+    with pytest.raises(ValueError):
+        h3.normalize(smaller, {"prompt": "Park.", "width": 768, "height": 768})
+
+
+def test_repeated_compilation_does_not_mutate_template_or_previous_result():
+    before = h3.CANDIDATE.read_bytes()
+    first = candidate(appearance_image=True, audio_mode="reference")
+    frozen = copy.deepcopy(first)
+    second = candidate(opening_image=True, audio_mode="preserve")
+    assert first == frozen
+    assert first["spec"] != second["spec"]
+    assert h3.CANDIDATE.read_bytes() == before
+    assert candidate()["source_bindings"] == []
diff --git a/recipes/AGENTS.md b/recipes/AGENTS.md
index 19a90892..6379870c 100644
--- a/recipes/AGENTS.md
+++ b/recipes/AGENTS.md
@@ -59,5 +59,5 @@ knobs → validate → drop here).
 
 ## Child DOX Index
 
-- [qualification/AGENTS.md](qualification/AGENTS.md) - non-serving LTX-2.5
+- [qualification/AGENTS.md](qualification/AGENTS.md) - non-serving media
   candidate recipes and promotion gates. Startup loading must stay nonrecursive.
diff --git a/recipes/qualification/AGENTS.md b/recipes/qualification/AGENTS.md
index ceaf19d1..4c408dbf 100644
--- a/recipes/qualification/AGENTS.md
+++ b/recipes/qualification/AGENTS.md
@@ -13,6 +13,10 @@ recursively, and inventory alone does not enable job admission.
 
 ## Ownership
 
+- `fasth3/` owns an offline-only FastH3 Omni NVFP4 candidate scaffold and
+  separate17k+5frame/input-role contract. Live recipe parsing rejects its H3
+  schema; startup remains nonrecursive. No model execution, public capability,
+  tariff or activation is authorized by these files. See its leaf contract.
 - `moss-image-sft-bf16-{t2i,i2i}-v1.json` retain the image qualification copies.
   The distinct MOSS Image SFT BF16 identity pins a BF16 diffusion backbone,
   official INT8 Qwen3.5-9B encoder and Flux2 VAE. Img2img uses source pixels
@@ -82,4 +86,4 @@ recursively, and inventory alone does not enable job admission.
 
 ## Child DOX Index
 
-- None - leaf.
+- [fasth3/AGENTS.md](fasth3/AGENTS.md) - offline-only H3 candidate and release gates.
diff --git a/recipes/qualification/fasth3/AGENTS.md b/recipes/qualification/fasth3/AGENTS.md
new file mode 100644
index 00000000..ad6298a3
--- /dev/null
+++ b/recipes/qualification/fasth3/AGENTS.md
@@ -0,0 +1,31 @@
+# FastH3 Qualification Candidates
+
+## Purpose
+
+Offline-only FastH3 Omni graph template and immutable component provenance.
+No recipe here is registered, priced, advertised, or approved for execution.
+Startup loading must remain nonrecursive; live parsing rejects the H3 schema.
+
+## Contracts
+
+- `README.md` owns candidate input semantics and outstanding rights/native/
+  protocol/billing/UI/rollout gates. Keep it explicit that this is not an API.
+- `omni-nvfp4-qad-v1.json` is the fixed Core-owned scaffold for the separate
+  `services/fasth3_contract.py` offline compiler. Preserve eight-step sampling,
+  model/component pins, 24fps and exact17k+5frames.
+- Appearance, opening, audio reference and preserved soundtrack have distinct
+  slots. Native guide combinations remain unqualified. No arbitrary graph,
+  filename, precision or adapter may come from client input.
+- FastVideo's pinned model card selects the MiniMax H3 Community License.
+  The reviewed license excludes US/EU/UK/South Korea, including hosted use.
+  No separate applicable authorization has been supplied. No additional model
+  execution/public serving follows before rights are resolved.
+- Qualification bounds describe experiments, not safe capacity. Promotion
+  requires rights, native quality/capacity, worker/source protocol, approved
+  tariff, funded billing/recovery/refund canaries and reviewed rollout.
+
+## Verification
+
+- `pytest grid_api/services/tests/test_fasth3_contract.py`
+- Existing LTX recipe/contract tests must remain green and unchanged.
+- Keep weights, private media, rig details and legal correspondence out of Git.
diff --git a/recipes/qualification/fasth3/README.md b/recipes/qualification/fasth3/README.md
new file mode 100644
index 00000000..10cb6008
--- /dev/null
+++ b/recipes/qualification/fasth3/README.md
@@ -0,0 +1,68 @@
+# FastH3 Omni Candidate
+
+This directory prepares an offline Core compiler, not a live Grid model. There
+is no public API support, tariff, worker registration or activation switch in
+this change. The current live recipe parser rejects the separate H3 schema;
+startup does not recurse into qualification directories.
+
+## Components And Rights
+
+The JSON records immutable model/runtime revisions and component digests for
+the official ComfyUI-native NVFP4 QAD release. It does not contain weights.
+These are distinct from FastVideo's own checkpoint format and inference kernels.
+No dependency, driver, installed runtime or existing LTX recipe changes here.
+
+The model card selects the
+[MiniMax H3 Community License](https://huggingface.co/MiniMaxAI/MiniMax-H3/blob/42ed227ee7df40d41602854ae760620d6eb651fe/LICENSE),
+which excludes US/EU/UK/South Korea, including hosted use. Separate applicable
+authorization has not been supplied. Further model execution and public serving
+remain gated. Source-code licensing does not grant model-use rights.
+
+## Candidate Contract
+
+`grid_api.services.fasth3_contract.compile_candidate` takes strict compiler
+inputs, not public API requests. It returns the Core-owned graph, normalized
+execution units and explicit unbound source roles. Filenames in the graph are
+placeholders, not proof of validated, uploaded or authorized source bytes.
+
+- Fixed eight-step res_multistep/simple sampling, guidance1, shifts12/3,24fps.
+- Explicit124..362frames on17k+5; invalid lengths reject rather than round.
+- Candidate canvases1344x768,768x1344,768x768. These bounds are not GPU fit or
+  quality qualification. The full range must not be advertised automatically.
+- Separate `appearance_image` and `opening_image` presence flags.
+- `audio_mode=generated|reference|preserve`, never ambiguous reference audio.
+  Both audio-input modes require an explicit image for this initial scope.
+- Appearance/reference audio condition generated output through reference
+  slots; they do not lock frame zero or promise the original waveform.
+- Opening images and preserved soundtracks select explicit guide experiments.
+  The preserved soundtrack is also selected for final mux. This is graph wiring,
+  not native compatibility, timing, identity or lip-sync acceptance.
+- No client filenames, graphs, video references, LoRAs, precision, sampler or
+  schedule overrides. Unknown fields and absent reference labels reject.
+
+`quote_units` computes an integer micro-USD ceiling from exact frames/24 and a
+supplied positive server rate. Its test rates are synthetic. It neither selects
+a tariff nor moves money. Future quotes/reservations must bind one approved,
+versioned rate and the exact same execution duration.
+
+## Remaining Release Gates
+
+Resolve model-use rights and required safeguards. Qualify native text, opening,
+appearance and audio combinations, including clocks, scene/identity quality,
+lip sync, capacity, cold/warm timing and matched LTX controls. Add a reviewed
+H3 source validator/commitment and worker-feature contract; never reinterpret
+LTX's8n+1frame,48kHz output or Ingredients-only image assumptions globally.
+
+Then integrate disabled-by-default recipe/API admission, authoritative pricing,
+private uploads/ownership/retention, output verification and Gallery UI. Reuse
+the existing durable reservation, execution, settlement, refund and GET recovery
+authorities. Require isolated lifecycle tests, bounded purchased canaries,
+reviewed merges, exact release checks, retained recovery and tested rollback
+before a one-worker public canary. Preserve available LTX capacity throughout.
+
+## Verification
+
+The H3 unit suite checks strict units, all candidate source combinations,
+absent live capability/price, public-async denial and unpriced billing denial
+before database access. Existing LTX contracts are unchanged. Tests use no
+model weights, live workers, storage credentials or purchased credits.
diff --git a/recipes/qualification/fasth3/omni-nvfp4-qad-v1.json b/recipes/qualification/fasth3/omni-nvfp4-qad-v1.json
new file mode 100644
index 00000000..78949856
--- /dev/null
+++ b/recipes/qualification/fasth3/omni-nvfp4-qad-v1.json
@@ -0,0 +1,92 @@
+{
+  "_grid": {
+    "name": "FastH3 Omni NVFP4 QAD qualification v1",
+    "modelName": "FastH3 Omni NVFP4 QAD",
+    "engine": "comfyui",
+    "jobType": "video",
+    "deterministic": false,
+    "qualificationOnly": true,
+    "requiredModels": ["FastH3 Omni NVFP4 QAD"],
+    "videoContract": {
+      "schema": "aipg.fasth3-video.v1",
+      "variant": "omni-nvfp4-qad",
+      "min_frames": 124,
+      "max_frames": 362,
+      "canvases": ["1344x768", "768x1344", "768x768"]
+    },
+    "provenance": {
+      "model_repository": "FastVideo/FastH3-Omni-NVFP4-QAD-Comfy",
+      "model_revision": "b1b4d67d79930007a3f73ca8e98739796ded95a0",
+      "comfy_revision": "b1a247c9cf9dbfda4b2c0a0c1d50950667eeaa0c",
+      "shared_repository": "FastVideo/FastH3-Omni-Comfy",
+      "shared_revision": "f519722692a381f5e365a03b4b60e6ea27b6e31d",
+      "components": {
+        "fasth3_omniref_qad_nvfp4_comfy.safetensors": "bf25ee58df65b4dfee349d612f7ac62a2fea0dddd1e68eb37e75c8ce8b166f5a",
+        "qwen3vl_32b_minimax_h3_nvfp4_awq.safetensors": "35a88d51044231fe332301d7a62aa81e3f2cba62febeb446e2c1e3e0ef76f2c6",
+        "minimax_h3_video_vae_fp16.safetensors": "7c1f131492e7eddacaac9069a61b81bdd39de5cc96561e677c5eab1cdce5e522",
+        "minimax_h3_audio_vae_fp32.safetensors": "8e505d95dd1561d47abd43d4238fd40d9bb1ae9e147ed0a4cba778d76ae4db48"
+      },
+      "license_repository": "MiniMaxAI/MiniMax-H3",
+      "license_revision": "42ed227ee7df40d41602854ae760620d6eb651fe",
+      "license_sha256": "59b99642b95ea21630e311198ddbfffbfe05aadba0c2f5d884cbdf4efcc90f44"
+    }
+  },
+  "model": {
+    "class_type": "UNETLoader",
+    "inputs": {"unet_name": "fasth3_omniref_qad_nvfp4_comfy.safetensors", "weight_dtype": "default"}
+  },
+  "encoder": {
+    "class_type": "CLIPLoader",
+    "inputs": {"clip_name": "qwen3vl_32b_minimax_h3_nvfp4_awq.safetensors", "type": "minimax", "device": "default"}
+  },
+  "video_vae": {
+    "class_type": "VAELoader",
+    "inputs": {"vae_name": "minimax_h3_video_vae_fp16.safetensors"}
+  },
+  "audio_vae": {
+    "class_type": "VAELoader",
+    "inputs": {"vae_name": "minimax_h3_audio_vae_fp32.safetensors"}
+  },
+  "conditioning": {
+    "class_type": "MiniMaxH3ReferenceToVideo",
+    "inputs": {
+      "clip": ["encoder", 0], "vae": ["video_vae", 0], "audio_vae": ["audio_vae", 0],
+      "prompt": "", "width": 1344, "height": 768, "length": 124, "ref_image_size": "match"
+    }
+  },
+  "shift": {
+    "class_type": "MiniMaxH3SigmaShift",
+    "inputs": {"model": ["model", 0], "shift_video": 12.0, "shift_audio": 3.0}
+  },
+  "sampler": {"class_type": "KSamplerSelect", "inputs": {"sampler_name": "res_multistep"}},
+  "sigmas": {
+    "class_type": "BasicScheduler",
+    "inputs": {"model": ["shift", 0], "scheduler": "simple", "steps": 8, "denoise": 1.0}
+  },
+  "guider": {
+    "class_type": "BasicGuider",
+    "inputs": {"model": ["shift", 0], "conditioning": ["conditioning", 0]}
+  },
+  "noise": {"class_type": "RandomNoise", "inputs": {"noise_seed": 29}},
+  "sample": {
+    "class_type": "SamplerCustomAdvanced",
+    "inputs": {
+      "noise": ["noise", 0], "guider": ["guider", 0], "sampler": ["sampler", 0],
+      "sigmas": ["sigmas", 0], "latent_image": ["conditioning", 1]
+    }
+  },
+  "decode_video": {
+    "class_type": "VAEDecode", "inputs": {"samples": ["sample", 0], "vae": ["video_vae", 0]}
+  },
+  "decode_audio": {
+    "class_type": "VAEDecodeAudio", "inputs": {"samples": ["sample", 0], "vae": ["audio_vae", 0]}
+  },
+  "video": {
+    "class_type": "CreateVideo",
+    "inputs": {"images": ["decode_video", 0], "fps": 24, "audio": ["decode_audio", 0]}
+  },
+  "save": {
+    "class_type": "SaveVideo",
+    "inputs": {"video": ["video", 0], "filename_prefix": "fasth3_qualification", "format": "auto", "codec": "h264"}
+  }
+}

From 1a037a00055ff6e3f538bba1104d6155abb9f2fa Mon Sep 17 00:00:00 2001
From: halfaipg <justin@techtoaster.com>
Date: Tue, 20 Oct 2026 18:34:34 -0400
Subject: [PATCH 2/5] feat(video): validate offline FastH3 source bundles

Verify bounded PNG and canonical PCM WAV inputs with distinct appearance, opening, voice-reference and soundtrack commitments. Keep preparation outside public admission and storage, with separate generated and preserved audio output clocks.
---
 grid_api/services/AGENTS.md                   |  10 +
 grid_api/services/fasth3_contract.py          |   2 +
 grid_api/services/fasth3_sources.py           | 106 +++++++++
 .../services/tests/test_fasth3_sources.py     | 204 ++++++++++++++++++
 recipes/qualification/fasth3/README.md        |  28 ++-
 5 files changed, 348 insertions(+), 2 deletions(-)
 create mode 100644 grid_api/services/fasth3_sources.py
 create mode 100644 grid_api/services/tests/test_fasth3_sources.py

diff --git a/grid_api/services/AGENTS.md b/grid_api/services/AGENTS.md
index 2f3ce1a3..ee02e5dc 100644
--- a/grid_api/services/AGENTS.md
+++ b/grid_api/services/AGENTS.md
@@ -15,6 +15,16 @@ content sanitization, and reward settlement.
   not native capacity/quality. Model-use rights, source protocol, purchased
   billing and public rollout remain unresolved release gates. Tests prove
   startup/public-async/unpriced-money denial, not license or native execution.
+- `fasth3_sources.py` prepares offline inline source bundles only after strict
+  candidate/role checks. It fully verifies/decodes bounded opaque, unrotated,
+  single-frame PNGs, with exact output dimensions for opening images. Verify
+  PNG structure before metadata reads can close its verification stream.
+  Canonical48kHz stereo PCM16 WAV references may last2..15seconds; preserved
+  soundtracks require exactly frames*2000samples. No trimming, stretching or
+  silent role conversion. Retain exact byte/hash/role/geometry/sample
+  commitments. Generated/reference sound declares32kHz output; preserved
+  sound declares48kHz. No upload, ownership, retention, worker advertisement,
+  native quality or permission proof follows from this offline preparation.
 
 - Live book `2026-10-20-a` adds the owner-approved `MOSS Image SFT BF16`
   image-only launch peg of USD0.010/output on Core0d2ee0b7.
diff --git a/grid_api/services/fasth3_contract.py b/grid_api/services/fasth3_contract.py
index f53ad7d0..3cbb5b9d 100644
--- a/grid_api/services/fasth3_contract.py
+++ b/grid_api/services/fasth3_contract.py
@@ -115,6 +115,8 @@ def normalize(contract: FastH3Contract, values: dict) -> dict:
         "seconds": float(duration),
         "duration_numerator": duration.numerator,
         "duration_denominator": duration.denominator,
+        "output_audio_sample_rate": 48_000 if inputs.audio_mode == "preserve" else 32_000,
+        "output_audio_channels": 2,
     }
 
 
diff --git a/grid_api/services/fasth3_sources.py b/grid_api/services/fasth3_sources.py
new file mode 100644
index 00000000..e0328d8d
--- /dev/null
+++ b/grid_api/services/fasth3_sources.py
@@ -0,0 +1,106 @@
+# SPDX-FileCopyrightText: 2026 AI Power Grid
+# SPDX-License-Identifier: AGPL-3.0-or-later
+
+"""Offline bounded FastH3 source preparation, with no upload or admission effect."""
+
+import io
+import struct
+import wave
+
+from fastapi import HTTPException
+from PIL import Image
+
+from .fasth3_contract import compile_candidate
+from .video_sources import MAX_AUDIO_BYTES, MAX_IMAGE_BYTES, Source, decode_inline
+
+IMAGE_ROLES = {"appearance_image", "opening_image"}
+AUDIO_ROLES = {"conditioning_audio", "soundtrack"}
+MAX_ENCODED_BUNDLE = 4 * (2 * MAX_IMAGE_BYTES + MAX_AUDIO_BYTES) // 3 + 256
+
+
+def _image(raw: bytes, role: str, width: int, height: int) -> Source:
+    if type(raw) is not bytes or not 0 < len(raw) <= MAX_IMAGE_BYTES:
+        raise ValueError("invalid image size")
+    with Image.open(io.BytesIO(raw), formats=["PNG"]) as image:
+        w, h = image.size
+        if (
+            not 32 <= w <= 2048
+            or not 32 <= h <= 2048
+            or w * h > 2048**2
+            or image.mode not in {"RGB", "RGBA"}
+            or (role == "opening_image" and (w, h) != (width, height))
+        ):
+            raise ValueError("invalid image geometry")
+        image.verify()
+    with Image.open(io.BytesIO(raw), formats=["PNG"]) as image:
+        if image.n_frames != 1 or image.getexif().get(274, 1) != 1:
+            raise ValueError("animated or rotated source image")
+        image.load()
+        if image.mode == "RGBA" and image.getchannel("A").getextrema() != (255, 255):
+            raise ValueError("transparent sources require explicit prior flattening")
+    return Source(role, raw, "png", {"width": w, "height": h})
+
+
+def _audio(raw: bytes, role: str, frames: int) -> Source:
+    # Canonical PCM WAV only: no metadata/extended/concatenated RIFF chunks.
+    if (
+        type(raw) is not bytes
+        or not 44 <= len(raw) <= MAX_AUDIO_BYTES
+        or raw[:4] != b"RIFF"
+        or raw[8:12] != b"WAVE"
+        or raw[12:16] != b"fmt "
+        or raw[36:40] != b"data"
+        or struct.unpack_from("<I", raw, 4)[0] != len(raw) - 8
+        or struct.unpack_from("<I", raw, 16)[0] != 16
+        or struct.unpack_from("<I", raw, 40)[0] != len(raw) - 44
+    ):
+        raise ValueError("invalid canonical WAV")
+    with wave.open(io.BytesIO(raw)) as reader:
+        samples = reader.getnframes()
+        if (
+            (reader.getnchannels(), reader.getsampwidth(), reader.getframerate(), reader.getcomptype()) != (2, 2, 48_000, "NONE")
+            or len(raw) != 44 + samples * 4
+            or struct.unpack_from("<I", raw, 28)[0] != 192_000
+            or struct.unpack_from("<H", raw, 32)[0] != 4
+            or len(reader.readframes(samples + 1)) != samples * 4
+        ):
+            raise ValueError("invalid PCM audio")
+    if role == "soundtrack":
+        if samples != frames * 2000:
+            raise ValueError("soundtrack must cover the exact target duration")
+    elif not 2 * 48_000 <= samples <= 15 * 48_000:
+        raise ValueError("audio reference must last between two and fifteen seconds")
+    units = {"sample_rate": 48_000, "channels": 2, "samples": samples}
+    if role == "soundtrack":
+        units.update(frames=frames, fps=24)
+    return Source(role, raw, "wav", units)
+
+
+def prepare_candidate(values: dict, inline_sources: dict) -> tuple[dict, tuple[Source, ...]]:
+    """Validate every named source before handing any bytes to a future uploader.
+
+    Appearance is not frame zero; reference audio is not a frozen soundtrack.
+    Inputs are already-normalized inline PNG/PCM WAV, never URLs or filenames.
+    No user/account ownership, private storage or worker permission is granted.
+    """
+    plan = compile_candidate(values)
+    expected = [binding["role"] for binding in plan["source_bindings"]]
+    try:
+        if (
+            type(inline_sources) is not dict
+            or set(inline_sources) != set(expected)
+            or not all(type(value) is str for value in inline_sources.values())
+            or sum(len(value) for value in inline_sources.values()) > MAX_ENCODED_BUNDLE
+        ):
+            raise ValueError("source roles or bounds disagree")
+        sources = []
+        execution = plan["execution"]
+        for role in expected:
+            image = role in IMAGE_ROLES
+            if not image and role not in AUDIO_ROLES:
+                raise ValueError("unknown source role")
+            raw = decode_inline(inline_sources[role], "reference_image" if image else "reference_audio")
+            sources.append(_image(raw, role, execution["width"], execution["height"]) if image else _audio(raw, role, execution["frames"]))
+    except (ValueError, TypeError, OSError, SyntaxError, EOFError, wave.Error, struct.error, Image.DecompressionBombError, HTTPException):
+        raise ValueError("Invalid FastH3 qualification sources") from None
+    return plan, tuple(sources)
diff --git a/grid_api/services/tests/test_fasth3_sources.py b/grid_api/services/tests/test_fasth3_sources.py
new file mode 100644
index 00000000..6605aad8
--- /dev/null
+++ b/grid_api/services/tests/test_fasth3_sources.py
@@ -0,0 +1,204 @@
+# SPDX-FileCopyrightText: 2026 AI Power Grid
+# SPDX-License-Identifier: AGPL-3.0-or-later
+
+"""Real synthetic PNG/WAV parsing, not native model or public upload proof."""
+
+import base64
+import hashlib
+import io
+import struct
+import wave
+
+import pytest
+from PIL import Image
+
+from grid_api.services import fasth3_sources as sources
+from grid_api.services import storage
+
+
+def inline(raw):
+    return base64.b64encode(raw).decode("ascii")
+
+
+def png(size=(64, 96), **options):
+    stream = io.BytesIO()
+    Image.new("RGB", size, "red").save(stream, "PNG", **options)
+    return stream.getvalue()
+
+
+def wav(samples=248_000, *, channels=2, rate=48_000, width=2):
+    stream = io.BytesIO()
+    with wave.open(stream, "wb") as writer:
+        writer.setparams((channels, width, rate, 0, "NONE", "not compressed"))
+        writer.writeframes(b"\0" * samples * channels * width)
+    return stream.getvalue()
+
+
+@pytest.fixture(autouse=True)
+def forbid_io(monkeypatch):
+    def forbidden(*_args, **_kwargs):
+        pytest.fail("offline source validation must not upload or select storage")
+
+    for name in ("upload_source", "upload_private_source", "private_source_bucket"):
+        monkeypatch.setattr(storage, name, forbidden)
+
+
+def prepare(values=None, bundle=None):
+    return sources.prepare_candidate({"prompt": "A sunny park.", **(values or {})}, bundle or {})
+
+
+def test_generated_plan_needs_no_sources_and_never_uploads():
+    plan, assets = prepare()
+    assert plan["qualification_only"] is True
+    assert assets == ()
+    assert plan["execution"]["output_audio_sample_rate"] == 32_000
+
+
+def test_appearance_opening_and_soundtrack_are_separate_hash_bound_roles():
+    picture, scene, sound = png(), png((1344, 768)), wav()
+    plan, assets = prepare(
+        {"appearance_image": True, "opening_image": True, "audio_mode": "preserve"},
+        {"appearance_image": inline(picture), "opening_image": inline(scene), "soundtrack": inline(sound)},
+    )
+    assert [asset.role for asset in assets] == ["appearance_image", "opening_image", "soundtrack"]
+    assert [asset.data for asset in assets] == [picture, scene, sound]
+    for asset in assets:
+        assert asset.commitment["sha256"] == hashlib.sha256(asset.data).hexdigest()
+        assert asset.commitment["bytes"] == len(asset.data)
+    assert assets[0].metadata == {"width": 64, "height": 96}
+    assert assets[1].metadata == {"width": 1344, "height": 768}
+    assert assets[2].metadata == {"sample_rate": 48_000, "channels": 2, "samples": 248_000, "frames": 124, "fps": 24}
+    assert plan["execution"]["output_audio_sample_rate"] == 48_000
+    assert plan["execution"]["output_audio_channels"] == 2
+
+
+@pytest.mark.parametrize("samples", [96_000, 144_000, 720_000])
+def test_voice_reference_duration_is_independent_of_target_video(samples):
+    _, assets = prepare(
+        {"appearance_image": True, "audio_mode": "reference"},
+        {"appearance_image": inline(png()), "conditioning_audio": inline(wav(samples))},
+    )
+    assert assets[1].metadata["samples"] == samples
+    assert "frames" not in assets[1].metadata
+
+
+@pytest.mark.parametrize("frames", [124, 141, 243, 362])
+def test_preserved_audio_uses_h3_exact_frames_not_ltx_grid(frames):
+    _, assets = prepare(
+        {"appearance_image": True, "audio_mode": "preserve", "frames": frames},
+        {"appearance_image": inline(png()), "soundtrack": inline(wav(frames * 2000))},
+    )
+    assert assets[1].metadata["samples"] == frames * 2000
+
+
+@pytest.mark.parametrize(
+    "bundle",
+    [
+        {"appearance_image": "https://example.invalid/image.png"},
+        {"appearance_image": "local.png"},
+        {"appearance_image": "%%%"},
+        {"appearance_image": "data:audio/wav;base64,YWJj"},
+        {"appearance_image": ""},
+        {"appearance_image": b"abc"},
+        {"opening_image": "YWJj"},
+        {"appearance_image": "YWJj", "soundtrack": "YWJj"},
+        {},
+    ],
+)
+def test_missing_extra_remote_wrong_role_or_malformed_source_rejects(bundle):
+    with pytest.raises(ValueError, match="Invalid FastH3 qualification sources"):
+        prepare({"appearance_image": True}, bundle)
+
+
+@pytest.mark.parametrize("bad", [b"", b"not a PNG", png()[:-8], png((16, 96)), png((2049, 32))])
+def test_image_bounds_and_complete_decode(bad):
+    with pytest.raises(ValueError):
+        prepare({"appearance_image": True}, {"appearance_image": inline(bad)})
+
+
+def test_opening_size_cannot_silently_crop_an_appearance_portrait():
+    with pytest.raises(ValueError):
+        prepare({"opening_image": True}, {"opening_image": inline(png())})
+
+
+@pytest.mark.parametrize("alpha", [0, 128, 255])
+def test_transparency_is_not_silently_dropped_by_the_model_loader(alpha):
+    stream = io.BytesIO()
+    Image.new("RGBA", (64, 96), (255, 0, 0, alpha)).save(stream, "PNG")
+    if alpha == 255:
+        _, assets = prepare({"appearance_image": True}, {"appearance_image": inline(stream.getvalue())})
+        assert len(assets) == 1
+    else:
+        with pytest.raises(ValueError):
+            prepare({"appearance_image": True}, {"appearance_image": inline(stream.getvalue())})
+
+
+@pytest.mark.parametrize("case", ["short", "long", "mono", "rate", "width", "truncated", "trailing", "invalid"])
+def test_target_audio_must_be_complete_canonical_exact_duration_pcm(case):
+    factories = {
+        "short": lambda: wav(247_999),
+        "long": lambda: wav(248_001),
+        "mono": lambda: wav(channels=1),
+        "rate": lambda: wav(rate=32_000),
+        "width": lambda: wav(width=1),
+        "truncated": lambda: wav()[:-1],
+        "trailing": lambda: wav() + b"extra",
+        "invalid": lambda: b"not audio",
+    }
+    bad = factories[case]()
+    with pytest.raises(ValueError):
+        prepare({"appearance_image": True, "audio_mode": "preserve"}, {"appearance_image": inline(png()), "soundtrack": inline(bad)})
+
+
+@pytest.mark.parametrize("samples", [95_999, 720_001])
+def test_reference_audio_bounds_do_not_silently_trim(samples):
+    with pytest.raises(ValueError):
+        prepare(
+            {"appearance_image": True, "audio_mode": "reference"},
+            {"appearance_image": inline(png()), "conditioning_audio": inline(wav(samples))},
+        )
+
+
+def test_wav_transport_fields_cannot_lie_about_byte_rate():
+    bad = bytearray(wav())
+    struct.pack_into("<I", bad, 28, 1)
+    with pytest.raises(ValueError):
+        prepare({"appearance_image": True, "audio_mode": "preserve"}, {"appearance_image": inline(png()), "soundtrack": inline(bytes(bad))})
+
+
+def test_encoded_bundle_bound_precedes_decoder_admission(monkeypatch):
+    monkeypatch.setattr(sources, "MAX_ENCODED_BUNDLE", 64)
+
+    def forbidden(*_args, **_kwargs):
+        pytest.fail("oversized input must reject before decoding")
+
+    monkeypatch.setattr(sources, "decode_inline", forbidden)
+    with pytest.raises(ValueError):
+        prepare({"appearance_image": True}, {"appearance_image": "a" * 65})
+
+
+def test_animated_png_is_not_a_single_reference_frame():
+    stream = io.BytesIO()
+    Image.new("RGB", (64, 96), "red").save(stream, "PNG", save_all=True, append_images=[Image.new("RGB", (64, 96), "blue")], duration=100)
+    with pytest.raises(ValueError):
+        prepare({"appearance_image": True}, {"appearance_image": inline(stream.getvalue())})
+
+
+def test_rotated_metadata_is_not_silently_transformed():
+    exif = Image.Exif()
+    exif[274] = 6
+    with pytest.raises(ValueError):
+        prepare({"appearance_image": True}, {"appearance_image": inline(png(exif=exif))})
+
+
+@pytest.mark.parametrize("role", ["appearance_image", "soundtrack"])
+def test_data_uri_is_accepted_only_with_the_exact_role_mime(role):
+    picture = "data:image/png;base64," + inline(png())
+    sound = "data:audio/wav;base64," + inline(wav())
+    values = {"appearance_image": True, "audio_mode": "preserve"}
+    bundle = {"appearance_image": picture, "soundtrack": sound}
+    _, assets = prepare(values, bundle)
+    assert len(assets) == 2
+    bundle[role] = bundle[role].replace("image/png", "image/jpeg").replace("audio/wav", "audio/mpeg")
+    with pytest.raises(ValueError):
+        prepare(values, bundle)
diff --git a/recipes/qualification/fasth3/README.md b/recipes/qualification/fasth3/README.md
index 10cb6008..56b30d81 100644
--- a/recipes/qualification/fasth3/README.md
+++ b/recipes/qualification/fasth3/README.md
@@ -45,12 +45,36 @@ supplied positive server rate. Its test rates are synthetic. It neither selects
 a tariff nor moves money. Future quotes/reservations must bind one approved,
 versioned rate and the exact same execution duration.
 
+## Offline Source Preparation
+
+`fasth3_sources.prepare_candidate` additionally checks all expected inline roles
+and byte limits before returning validated sources. Each image/audio input has
+a separate SHA-256, byte count and role commitment; nothing is uploaded.
+
+Images must fully decode as unrotated, nonanimated, opaque PNGs,32..2048pixels
+per dimension. Opening images must already match the candidate output canvas;
+appearance images keep their own geometry. Nonopaque images require an explicit
+flattening/preview step upstream, never silent alpha removal. Sources are not
+reencoded, auto-cropped, fitted or otherwise changed after commitment.
+
+Audio inputs are canonical metadata-free48kHz stereo PCM16 WAV. Voice/music
+references are2..15seconds independently of target duration. Preserved sound
+must contain exactly frames*2000samples; short/long audio rejects rather than
+padding or trimming. Output expectations distinguish generated32kHz audio from
+the explicitly muxed48kHz soundtrack. Valid waveform transport is not lip-sync
+or identity proof.
+
+This remains a disabled preparation helper, not a public upload route. Account
+ownership, private-bucket access, source retention/cleanup, worker revalidation,
+process-isolated decoder qualification and purchased end-to-end proofs are still
+required before activation. Keep the legacy LTX validator unchanged.
+
 ## Remaining Release Gates
 
 Resolve model-use rights and required safeguards. Qualify native text, opening,
 appearance and audio combinations, including clocks, scene/identity quality,
-lip sync, capacity, cold/warm timing and matched LTX controls. Add a reviewed
-H3 source validator/commitment and worker-feature contract; never reinterpret
+lip sync, capacity, cold/warm timing and matched LTX controls. Qualify the H3
+source validator/commitments and add a worker-feature contract; never reinterpret
 LTX's8n+1frame,48kHz output or Ingredients-only image assumptions globally.
 
 Then integrate disabled-by-default recipe/API admission, authoritative pricing,

From 1957d1360f3a0a4250da784f62a10b4bf3b8ee19 Mon Sep 17 00:00:00 2001
From: halfaipg <justin@techtoaster.com>
Date: Tue, 20 Oct 2026 18:52:16 -0400
Subject: [PATCH 3/5] test(video): prove FastH3 Core-to-worker source handoff

---
 grid_api/services/AGENTS.md                   |   8 +
 .../tests/test_fasth3_worker_handoff.py       | 138 ++++++++++++++++++
 2 files changed, 146 insertions(+)
 create mode 100644 grid_api/services/tests/test_fasth3_worker_handoff.py

diff --git a/grid_api/services/AGENTS.md b/grid_api/services/AGENTS.md
index ee02e5dc..cb0d6196 100644
--- a/grid_api/services/AGENTS.md
+++ b/grid_api/services/AGENTS.md
@@ -1084,6 +1084,14 @@ content sanitization, and reward settlement.
 
 ## Verification
 
+- `tests/test_fasth3_worker_handoff.py` optionally runs actual offline Core
+  source preparation against `MEDIA_VIDEO_WORKER_SOURCE`'s separate FastH3
+  adapter. `MEDIA_VIDEO_WORKER_PYTHON` may select its reviewed test interpreter.
+  Ten role combinations use synthetic PNG/PCM WAV and mocked credential-free
+  HTTP, checking exact bytes, collision names and graph preservation. No GPU,
+  storage, ownership, native quality or paid lifecycle qualification is implied;
+  the public runtime remains closed and licensing remains a separate gate.
+
 - `tests/test_x402_http_runtime.py` exercises the pinned official SDK and
   production FastAPI route/middleware with synthetic facilitator and worker
   dependencies: challenge, exact-usage override, replay/stream/malformed input,
diff --git a/grid_api/services/tests/test_fasth3_worker_handoff.py b/grid_api/services/tests/test_fasth3_worker_handoff.py
new file mode 100644
index 00000000..0e26aca0
--- /dev/null
+++ b/grid_api/services/tests/test_fasth3_worker_handoff.py
@@ -0,0 +1,138 @@
+# SPDX-FileCopyrightText: 2026 AI Power Grid
+# SPDX-License-Identifier: AGPL-3.0-or-later
+
+"""Opt-in real Core/worker contract check with synthetic media and mocked HTTP."""
+
+import base64
+import io
+import json
+import os
+import subprocess
+import wave
+from pathlib import Path
+
+import pytest
+from PIL import Image
+
+from grid_api.services.fasth3_sources import prepare_candidate
+
+
+@pytest.mark.parametrize(
+    "appearance,opening,mode",
+    [
+        (False, False, "generated"),
+        (True, False, "generated"),
+        (False, True, "generated"),
+        (True, True, "generated"),
+        (True, False, "reference"),
+        (False, True, "reference"),
+        (True, True, "reference"),
+        (True, False, "preserve"),
+        (False, True, "preserve"),
+        (True, True, "preserve"),
+    ],
+)
+def test_actual_core_source_plan_reaches_candidate_worker_without_dispatch(tmp_path, appearance, opening, mode):
+    source = os.environ.get("MEDIA_VIDEO_WORKER_SOURCE")
+    if not source:
+        pytest.skip("reviewed MEDIA_VIDEO_WORKER_SOURCE is required for FastH3 handoff")
+    worker = Path(source).resolve()
+    # Keep the venv entry path: resolving its symlink selects bare Python.
+    python = Path(os.environ.get("MEDIA_VIDEO_WORKER_PYTHON", str(worker / ".venv/bin/python"))).absolute()
+    if not python.is_file() or not (worker / "bridge/fasth3_sources.py").is_file():
+        pytest.fail("FastH3 handoff requires its reviewed worker checkout and test Python")
+
+    def picture(size):
+        stream = io.BytesIO()
+        Image.new("RGB", size, "red").save(stream, "PNG")
+        return base64.b64encode(stream.getvalue()).decode()
+
+    bundle = {}
+    if appearance:
+        bundle["appearance_image"] = picture((64, 96))
+    if opening:
+        bundle["opening_image"] = picture((768, 1344))
+    if mode != "generated":
+        sound = io.BytesIO()
+        with wave.open(sound, "wb") as writer:
+            writer.setparams((2, 2, 48_000, 0, "NONE", "not compressed"))
+            writer.writeframes(b"\0" * (282_000 if mode == "preserve" else 96_000) * 4)
+        bundle["soundtrack" if mode == "preserve" else "conditioning_audio"] = base64.b64encode(sound.getvalue()).decode()
+    plan, sources = prepare_candidate(
+        {
+            "prompt": "A sunny synthetic park.",
+            "seed": 0,
+            "frames": 141,
+            "width": 768,
+            "height": 1344,
+            "appearance_image": appearance,
+            "opening_image": opening,
+            "audio_mode": mode,
+        },
+        bundle,
+    )
+    content, committed = {}, []
+    for source_media in sources:
+        url = "https://" + "a" * 32 + f".r2.cloudflarestorage.com/synthetic/{source_media.role}.{source_media.ext}"
+        content[url] = base64.b64encode(source_media.data).decode()
+        committed.append({**source_media.commitment, "url": url})
+    payload = {
+        "recipe_engine": "comfyui",
+        "recipe_source_schema": "aipg.fasth3-sources.v1",
+        "required_media_features": ["recipe-fasth3-sources-v1"],
+        "n": 1,
+        "execution": plan["execution"],
+        "recipe_spec": plan["spec"],
+        "recipe_source_bindings": plan["source_bindings"],
+        "source_media": committed,
+    }
+    script = """
+import asyncio, base64, copy, json, sys
+import httpx, respx
+sys.path.insert(0, sys.argv[1])
+from bridge import fasth3_sources, source_media
+source_media.Settings.COMFYUI_URL = "http://127.0.0.1:8188"
+request = json.load(sys.stdin)
+body = request["payload"]
+original = copy.deepcopy(body)
+uploaded = []
+def read(request):
+    assert "authorization" not in request.headers
+    return httpx.Response(200, content=base64.b64decode(content[str(request.url)]))
+def upload(request):
+    assert "authorization" not in request.headers
+    index = len(uploaded)
+    source = body["source_media"][index]
+    extension = "png" if source["content_type"] == "image/png" else "wav"
+    name = "collision-" + str(index) + "." + extension
+    uploaded.append(name)
+    return httpx.Response(200, json={"name": name, "type": "input", "subfolder": ""})
+content = request["content"]
+async def run():
+    with respx.mock(assert_all_called=True) as routes:
+        for url in content:
+            routes.get(url).mock(side_effect=read)
+        if content:
+            routes.post("http://127.0.0.1:8188/upload/image").mock(side_effect=upload)
+        identity = "a0666ffe-da9b-4e92-93c1-abf9301e0558"
+        graph = await fasth3_sources.build_candidate_workflow(identity, body)
+        expected = copy.deepcopy(body["recipe_spec"])
+        for index, binding in enumerate(body["recipe_source_bindings"]):
+            name, _, field = binding["path"].split(".")
+            expected[name]["inputs"][field] = uploaded[index]
+        expected["save"]["inputs"]["filename_prefix"] = "grid_" + identity
+        assert graph == expected
+        assert body == original
+        assert len(uploaded) == len(body["source_media"])
+asyncio.run(run())
+"""
+    result = subprocess.run(
+        [str(python), "-c", script, str(worker)],
+        cwd=tmp_path,
+        env={"PATH": os.environ.get("PATH", ""), "PYTHONDONTWRITEBYTECODE": "1", "PYTHONNOUSERSITE": "1", "PYTHON_DOTENV_DISABLED": "1"},
+        input=json.dumps({"payload": payload, "content": content}),
+        text=True,
+        capture_output=True,
+        timeout=30,
+    )
+    assert result.returncode == 0, "synthetic FastH3 Core/worker handoff failed"

From 91abf546f0a1aef7aab753698932d1407b2edb79 Mon Sep 17 00:00:00 2001
From: halfaipg <justin@techtoaster.com>
Date: Tue, 20 Oct 2026 19:20:06 -0400
Subject: [PATCH 4/5] feat(video): inspect offline FastH3 output clocks

Reuse the bounded codec runner for complete frame and audio inspection. Keep AAC presentation timing separate from decoded padding and leave live freeze, settlement, and admission untouched. Give the optional H3 handoff its own worker-source pin rather than consuming the existing LTX fixture.
---
 grid_api/services/AGENTS.md                   |  17 +-
 grid_api/services/fasth3_output.py            | 314 ++++++++++++++
 grid_api/services/tests/test_fasth3_output.py | 402 ++++++++++++++++++
 .../tests/test_fasth3_worker_handoff.py       |   6 +-
 4 files changed, 734 insertions(+), 5 deletions(-)
 create mode 100644 grid_api/services/fasth3_output.py
 create mode 100644 grid_api/services/tests/test_fasth3_output.py

diff --git a/grid_api/services/AGENTS.md b/grid_api/services/AGENTS.md
index cb0d6196..36b2278c 100644
--- a/grid_api/services/AGENTS.md
+++ b/grid_api/services/AGENTS.md
@@ -25,6 +25,17 @@ content sanitization, and reward settlement.
   commitments. Generated/reference sound declares32kHz output; preserved
   sound declares48kHz. No upload, ownership, retention, worker advertisement,
   native quality or permission proof follows from this offline preparation.
+- `fasth3_output.py` inspects offline candidate bytes through the existing
+  bounded/isolation-required codec runner. Require one H.264/AAC pair, exact
+  H3 frame/resolution/track clocks and full contiguous frame/audio decoding.
+  AAC must be LC; at most1023decoded terminal padding samples are separately
+  reported after an independently valid presentation clock. Numeric probe
+  fields are bounded decimal/rational strings, never exponential allocations.
+  Generated/reference audio presents32kHz; preserved audio presents48kHz.
+  Report AAC terminal padding separately, never trim to manufacture agreement.
+  Codec/capacity/runtime failures remain uncertain, not failed-render/refund
+  authority. No live freeze/settlement caller exists; a reviewed immutable-byte
+  handoff, native identity/sound/lip-sync tests and paid lifecycle remain gates.
 
 - Live book `2026-10-20-a` adds the owner-approved `MOSS Image SFT BF16`
   image-only launch peg of USD0.010/output on Core0d2ee0b7.
@@ -1085,8 +1096,10 @@ content sanitization, and reward settlement.
 ## Verification
 
 - `tests/test_fasth3_worker_handoff.py` optionally runs actual offline Core
-  source preparation against `MEDIA_VIDEO_WORKER_SOURCE`'s separate FastH3
-  adapter. `MEDIA_VIDEO_WORKER_PYTHON` may select its reviewed test interpreter.
+  source preparation against `FASTH3_WORKER_SOURCE`'s separate FastH3
+  adapter. `FASTH3_WORKER_PYTHON` may select its reviewed test interpreter.
+  Keep this opt-in independent of the existing LTX worker fixture pin; an old
+  LTX adapter is not an H3 handoff source, and an explicit bad H3 pin must fail.
   Ten role combinations use synthetic PNG/PCM WAV and mocked credential-free
   HTTP, checking exact bytes, collision names and graph preservation. No GPU,
   storage, ownership, native quality or paid lifecycle qualification is implied;
diff --git a/grid_api/services/fasth3_output.py b/grid_api/services/fasth3_output.py
new file mode 100644
index 00000000..76fb1082
--- /dev/null
+++ b/grid_api/services/fasth3_output.py
@@ -0,0 +1,314 @@
+# SPDX-FileCopyrightText: 2026 AI Power Grid
+# SPDX-License-Identifier: AGPL-3.0-or-later
+
+"""Offline H3 output inspection; never admission, model fidelity or money authority.
+
+Call synchronously outside the event loop. This deliberately does not plug into
+the live S3 freeze/settlement path; that needs a reviewed immutable-byte handoff.
+"""
+
+import csv
+import hashlib
+import json
+import re
+import subprocess
+import tempfile
+import threading
+from dataclasses import dataclass
+from fractions import Fraction
+from pathlib import Path
+from typing import Any
+
+from . import media_decoder
+from .fasth3_contract import compile_candidate
+from .video_output import MAX_VIDEO_BYTES
+
+_SLOTS = threading.BoundedSemaphore(2)
+
+
+class OutputRejected(ValueError):
+    """A completed observation violates this candidate's media contract."""
+
+
+class OutputCheckUnavailable(RuntimeError):
+    """Verification cannot establish a result; never infer payment or refund."""
+
+
+@dataclass(frozen=True)
+class OutputEvidence:
+    sha256: str
+    bytes: int
+    frames: int
+    width: int
+    height: int
+    duration_numerator: int
+    duration_denominator: int
+    audio_sample_rate: int
+    audio_presentation_samples: int
+    audio_decoded_samples: int
+    aac_padding_samples: int
+    qualification_only: bool = True
+
+
+def _integer(value: Any) -> int:
+    if type(value) is int and 0 <= value < 2**63:
+        return value
+    if isinstance(value, str) and re.fullmatch(r"0|[1-9][0-9]{0,17}", value):
+        return int(value)
+    raise ValueError("invalid integer")
+
+
+def _rational(value: Any) -> Fraction:
+    if not isinstance(value, str) or not re.fullmatch(r"-?(?:0|[1-9][0-9]{0,17})(?:\.[0-9]{1,12}|/[1-9][0-9]{0,17})?", value):
+        raise ValueError("invalid rational")
+    return Fraction(value)
+
+
+def _run(tool: str, args: list[str], directory: Path, *, timeout: int, limit: int) -> bytes:
+    try:
+        data = media_decoder.run_tool(tool, args, directory, timeout=timeout, limit=limit)
+        if not isinstance(data, bytes) or len(data) > limit:
+            raise ValueError("invalid decoder response")
+        return data
+    except (OSError, ValueError, subprocess.SubprocessError):
+        # Decoder exit can also mean namespace, resource or runtime failure.
+        # Do not misclassify it as definitive evidence to refund a held job.
+        raise OutputCheckUnavailable("FastH3 output verification unavailable") from None
+
+
+def _probe(raw: bytes, execution: dict[str, Any]) -> int:
+    try:
+        metadata = json.loads(raw)
+        streams = metadata["streams"]
+        if not isinstance(streams, list) or len(streams) != 2:
+            raise ValueError
+        videos = [stream for stream in streams if stream["codec_type"] == "video"]
+        audios = [stream for stream in streams if stream["codec_type"] == "audio"]
+        if len(videos) != 1 or len(audios) != 1:
+            raise ValueError
+        video, audio = videos[0], audios[0]
+        frames, width, height = (execution[key] for key in ("frames", "width", "height"))
+        duration, rate = Fraction(frames, 24), execution["output_audio_sample_rate"]
+        if (
+            video["codec_name"] != "h264"
+            or type(video["width"]) is not int
+            or video["width"] != width
+            or type(video["height"]) is not int
+            or video["height"] != height
+            or video["pix_fmt"] != "yuv420p"
+            or _integer(video["nb_frames"]) != frames
+            or _rational(video["avg_frame_rate"]) != 24
+            or _rational(video["r_frame_rate"]) != 24
+            or _integer(video["start_pts"]) != 0
+            or _rational(video["start_time"]) != 0
+            or _rational(video["time_base"]) <= 0
+            or _integer(video["duration_ts"]) * _rational(video["time_base"]) != duration
+            or video.get("sample_aspect_ratio") not in (None, "N/A", "1:1")
+            or video.get("disposition", {}).get("attached_pic", 0)
+            or any(item.get("rotation", 0) != 0 for item in video.get("side_data_list", []))
+            or audio["codec_name"] != "aac"
+            or audio.get("profile") != "LC"
+            or _integer(audio["sample_rate"]) != rate
+            or type(audio["channels"]) is not int
+            or audio["channels"] != 2
+            or _integer(audio["start_pts"]) != 0
+            or _rational(audio["start_time"]) != 0
+            or _rational(audio["time_base"]) != Fraction(1, rate)
+        ):
+            raise ValueError
+        samples = _integer(audio["duration_ts"])
+        if abs(Fraction(samples, rate) - duration) > Fraction(1, rate):
+            raise ValueError
+        container = metadata["format"]
+        if (
+            "mp4" not in container["format_name"].split(",")
+            or _rational(container["start_time"]) != 0
+            # MP4's movie timescale may round the overall duration to a ms.
+            # Track clocks above, not this display field, prove synchronization.
+            or abs(_rational(container["duration"]) - max(duration, Fraction(samples, rate))) > Fraction(1, 1000)
+        ):
+            raise ValueError
+    except (KeyError, TypeError, ValueError, AttributeError, ZeroDivisionError):
+        raise OutputRejected("FastH3 output metadata violates the candidate contract") from None
+    return samples
+
+
+def _rows(raw: bytes, headers: set[str]) -> list[list[str]]:
+    try:
+        lines = raw.decode("ascii").splitlines()
+        if not headers.issubset(set(lines)):
+            raise ValueError
+        rows = list(csv.reader(line for line in lines if line and not line.startswith("#")))
+        if not rows or len(rows) > 400:
+            raise ValueError
+        for row in rows:
+            if len(row) != 6 or not re.fullmatch(r"[0-9a-f]{32}", row[5].strip()):
+                raise ValueError
+    except (UnicodeError, ValueError, csv.Error):
+        raise OutputRejected("FastH3 output decode observation is invalid") from None
+    return rows
+
+
+def _video_frames(raw: bytes, execution: dict[str, Any]) -> None:
+    width, height, frames = (execution[key] for key in ("width", "height", "frames"))
+    rows = _rows(raw, {"#tb 0: 1/24", "#media_type 0: video", "#codec_id 0: rawvideo", f"#dimensions 0: {width}x{height}"})
+    try:
+        if len(rows) != frames:
+            raise ValueError
+        for index, row in enumerate(rows):
+            if [_integer(item.strip()) for item in row[:5]] != [0, index, index, 1, width * height * 3 // 2]:
+                raise ValueError
+    except ValueError:
+        raise OutputRejected("FastH3 decoded video violates the candidate frame clock") from None
+
+
+def _audio_samples(raw: bytes, rate: int, presented: int) -> int:
+    rows = _rows(
+        raw,
+        {f"#tb 0: 1/{rate}", "#media_type 0: audio", "#codec_id 0: pcm_s16le", f"#sample_rate 0: {rate}", "#channel_layout_name 0: stereo"},
+    )
+    total = 0
+    try:
+        for row in rows:
+            stream, dts, pts, count, size = (_integer(item.strip()) for item in row[:5])
+            if stream != 0 or dts != total or pts != total or not 0 < count <= 4096 or size != count * 4:
+                raise ValueError
+            total += count
+        # Decode everything without trimming. AAC can pad one terminal 1024-
+        # sample block; its presentation clock must independently pass _probe.
+        if not presented <= total <= presented + 1023:
+            raise ValueError
+    except ValueError:
+        raise OutputRejected("FastH3 decoded audio violates the candidate sample clock") from None
+    return total
+
+
+def inspect_candidate(raw: bytes, *, inputs: dict[str, Any]) -> OutputEvidence:
+    """Inspect exact local bytes; no upload, model, settlement or admission calls.
+
+    Missing/failed codecs or capacity are uncertain, not failed-render evidence.
+    Passing proves bounded media conformance only, never requested identity,
+    soundtrack fidelity, lip sync, model weights or production readiness.
+    """
+    execution = compile_candidate(inputs)["execution"]
+    if (
+        type(raw) is not bytes
+        or not 16 <= len(raw) <= MAX_VIDEO_BYTES
+        or raw[4:8] != b"ftyp"
+        or not 16 <= int.from_bytes(raw[:4], "big") <= len(raw)
+        or int.from_bytes(raw[:4], "big") % 4
+    ):
+        raise OutputRejected("FastH3 output is outside its byte contract")
+    if not media_decoder.available():
+        raise OutputCheckUnavailable("FastH3 output verification unavailable")
+    if not _SLOTS.acquire(timeout=5):
+        raise OutputCheckUnavailable("FastH3 output verification capacity unavailable")
+    try:
+        with tempfile.TemporaryDirectory(prefix="grid-fasth3-output-") as directory:
+            workspace = Path(directory)
+            path = workspace / "output.mp4"
+            path.write_bytes(raw)
+            path.chmod(0o600)
+            base = [
+                "-v",
+                "error",
+                "-max_alloc",
+                "67108864",
+                "-threads",
+                "1",
+                "-protocol_whitelist",
+                "file,pipe",
+                "-enable_drefs",
+                "0",
+                "-use_absolute_path",
+                "0",
+                "-f",
+                "mov",
+            ]
+            metadata = _run(
+                "ffprobe",
+                [
+                    *base,
+                    "-show_entries",
+                    "stream=codec_type,codec_name,profile,width,height,pix_fmt,sample_aspect_ratio,avg_frame_rate,r_frame_rate,"
+                    "nb_frames,time_base,start_pts,start_time,duration_ts,sample_rate,channels:stream_side_data=rotation:"
+                    "stream_disposition=attached_pic:format=format_name,duration,start_time",
+                    "-of",
+                    "json",
+                    str(path),
+                ],
+                workspace,
+                timeout=10,
+                limit=32768,
+            )
+            presented = _probe(metadata, execution)
+            decode = ["-nostdin", "-xerror", "-err_detect", "explode", *base, "-i", str(path)]
+            frames = _run(
+                "ffmpeg",
+                [
+                    *decode,
+                    "-map",
+                    "0:v:0",
+                    "-an",
+                    "-sn",
+                    "-dn",
+                    "-threads",
+                    "1",
+                    "-frames:v",
+                    str(execution["frames"] + 1),
+                    "-fps_mode",
+                    "passthrough",
+                    "-c:v",
+                    "rawvideo",
+                    "-pix_fmt",
+                    "yuv420p",
+                    "-f",
+                    "framemd5",
+                    "-",
+                ],
+                workspace,
+                timeout=30,
+                limit=65536,
+            )
+            _video_frames(frames, execution)
+            sound = _run(
+                "ffmpeg",
+                [
+                    *decode,
+                    "-map",
+                    "0:a:0",
+                    "-vn",
+                    "-sn",
+                    "-dn",
+                    "-threads",
+                    "1",
+                    "-af",
+                    "asetnsamples=n=4096:p=0",
+                    "-c:a",
+                    "pcm_s16le",
+                    "-f",
+                    "framemd5",
+                    "-",
+                ],
+                workspace,
+                timeout=30,
+                limit=65536,
+            )
+            decoded = _audio_samples(sound, execution["output_audio_sample_rate"], presented)
+    except OSError:
+        raise OutputCheckUnavailable("FastH3 output verification unavailable") from None
+    finally:
+        _SLOTS.release()
+    return OutputEvidence(
+        sha256=hashlib.sha256(raw).hexdigest(),
+        bytes=len(raw),
+        frames=execution["frames"],
+        width=execution["width"],
+        height=execution["height"],
+        duration_numerator=execution["duration_numerator"],
+        duration_denominator=execution["duration_denominator"],
+        audio_sample_rate=execution["output_audio_sample_rate"],
+        audio_presentation_samples=presented,
+        audio_decoded_samples=decoded,
+        aac_padding_samples=decoded - presented,
+    )
diff --git a/grid_api/services/tests/test_fasth3_output.py b/grid_api/services/tests/test_fasth3_output.py
new file mode 100644
index 00000000..553c86cb
--- /dev/null
+++ b/grid_api/services/tests/test_fasth3_output.py
@@ -0,0 +1,402 @@
+# SPDX-FileCopyrightText: 2026 AI Power Grid
+# SPDX-License-Identifier: AGPL-3.0-or-later
+
+"""Synthetic output-clock faults and real CPU decode, never native generation."""
+
+import hashlib
+import json
+import shutil
+import subprocess
+from fractions import Fraction
+from unittest.mock import Mock
+
+import pytest
+
+from grid_api.services import fasth3_output as output
+from grid_api.services import storage
+
+RAW = bytes.fromhex("000000186674797069736f6d0000020069736f6d69736f32")
+
+
+def metadata(frames=124, rate=32_000, size=(1344, 768)):
+    duration = Fraction(frames, 24)
+    samples = (frames * rate + 23) // 24
+    return {
+        "streams": [
+            {
+                "codec_type": "video",
+                "codec_name": "h264",
+                "width": size[0],
+                "height": size[1],
+                "pix_fmt": "yuv420p",
+                "sample_aspect_ratio": "1:1",
+                "avg_frame_rate": "24/1",
+                "r_frame_rate": "24/1",
+                "nb_frames": str(frames),
+                "time_base": "1/12288",
+                "start_pts": 0,
+                "start_time": "0.000000",
+                "duration_ts": frames * 512,
+            },
+            {
+                "codec_type": "audio",
+                "codec_name": "aac",
+                "profile": "LC",
+                "sample_rate": str(rate),
+                "channels": 2,
+                "time_base": f"1/{rate}",
+                "start_pts": 0,
+                "start_time": "0.000000",
+                "duration_ts": samples,
+            },
+        ],
+        "format": {
+            "format_name": "mov,mp4,m4a,3gp,3g2,mj2",
+            "start_time": "0.000000",
+            "duration": f"{float(max(duration, Fraction(samples, rate))):.6f}",
+        },
+    }
+
+
+def video_rows(frames=124, size=(1344, 768)):
+    headers = ["#tb 0: 1/24", "#media_type 0: video", "#codec_id 0: rawvideo", f"#dimensions 0: {size[0]}x{size[1]}"]
+    return ("\n".join(headers + [f"0, {i}, {i}, 1, {size[0] * size[1] * 3 // 2}, " + "a" * 32 for i in range(frames)]) + "\n").encode()
+
+
+def audio_rows(frames=124, rate=32_000, *, samples=None):
+    if samples is None:
+        presented = (frames * rate + 23) // 24
+        samples = ((presented + 1023) // 1024) * 1024
+    headers = [
+        f"#tb 0: 1/{rate}",
+        "#media_type 0: audio",
+        "#codec_id 0: pcm_s16le",
+        f"#sample_rate 0: {rate}",
+        "#channel_layout_name 0: stereo",
+    ]
+    rows = [f"0, {i}, {i}, {min(4096, samples - i)}, {min(4096, samples - i) * 4}, " + "b" * 32 for i in range(0, samples, 4096)]
+    return ("\n".join(headers + rows) + "\n").encode()
+
+
+@pytest.fixture
+def codecs(monkeypatch):
+    calls = []
+    values = {"metadata": metadata(), "video": video_rows(), "audio": audio_rows()}
+
+    def run(tool, args, directory, **bounds):
+        calls.append((tool, args, directory, bounds))
+        assert directory.stat().st_mode & 0o777 == 0o700
+        assert (directory / "output.mp4").stat().st_mode & 0o777 == 0o600
+        assert bounds["timeout"] <= 30 and bounds["limit"] <= 65536
+        assert args[args.index("-protocol_whitelist") + 1] == "file,pipe"
+        assert args[args.index("-enable_drefs") + 1] == "0"
+        assert args[args.index("-use_absolute_path") + 1] == "0"
+        return json.dumps(values["metadata"]).encode() if tool == "ffprobe" else values["video" if "-an" in args else "audio"]
+
+    monkeypatch.setattr(output.media_decoder, "available", lambda: True)
+    monkeypatch.setattr(output.media_decoder, "run_tool", run)
+
+    def no_storage():
+        pytest.fail("offline output inspection must never use storage")
+
+    monkeypatch.setattr(storage, "_client", no_storage)
+    return values, calls
+
+
+def inspect(raw=RAW, **inputs):
+    return output.inspect_candidate(raw, inputs={"prompt": "Synthetic output check.", **inputs})
+
+
+@pytest.mark.parametrize("frames,rate,size", [(124, 32_000, (1344, 768)), (141, 48_000, (768, 1344)), (362, 32_000, (768, 768))])
+def test_exact_presentation_clock_and_full_untrimmed_decode_are_separate(codecs, frames, rate, size):
+    values, calls = codecs
+    values.update(metadata=metadata(frames, rate, size), video=video_rows(frames, size), audio=audio_rows(frames, rate))
+    options = {"appearance_image": True, "audio_mode": "preserve"} if rate == 48_000 else {}
+    result = inspect(frames=frames, width=size[0], height=size[1], **options)
+    assert result.sha256 == hashlib.sha256(RAW).hexdigest() and result.bytes == len(RAW)
+    assert result.frames == frames and result.audio_sample_rate == rate
+    assert Fraction(result.duration_numerator, result.duration_denominator) == Fraction(frames, 24)
+    assert result.audio_presentation_samples == (frames * rate + 23) // 24
+    assert 0 <= result.aac_padding_samples <= 1023
+    assert result.audio_decoded_samples == result.audio_presentation_samples + result.aac_padding_samples
+    assert result.qualification_only is True and len(calls) == 3
+    assert all(not path.exists() for _, _, path, _ in calls)
+    audio_args = calls[-1][1]
+    assert "atrim" not in " ".join(audio_args) and "-t" not in audio_args
+    assert audio_args[audio_args.index("-af") + 1] == "asetnsamples=n=4096:p=0"
+
+
+@pytest.mark.parametrize(
+    "path,value",
+    [
+        (("streams",), []),
+        (("streams", 0, "codec_name"), "hevc"),
+        (("streams", 0, "width"), 768),
+        (("streams", 0, "pix_fmt"), "yuv444p"),
+        (("streams", 0, "nb_frames"), "125"),
+        (("streams", 0, "avg_frame_rate"), "30/1"),
+        (("streams", 0, "r_frame_rate"), "30/1"),
+        (("streams", 0, "start_pts"), 1),
+        (("streams", 0, "start_time"), "0.1"),
+        (("streams", 0, "duration_ts"), 63489),
+        (("streams", 0, "time_base"), "0/1"),
+        (("streams", 0, "sample_aspect_ratio"), "2:1"),
+        (("streams", 0, "side_data_list"), [{"rotation": 90}]),
+        (("streams", 0, "disposition"), {"attached_pic": 1}),
+        (("streams", 1, "codec_name"), "opus"),
+        (("streams", 1, "profile"), "HE-AAC"),
+        (("streams", 1, "sample_rate"), "48000"),
+        (("streams", 1, "channels"), 1),
+        (("streams", 1, "start_pts"), 1024),
+        (("streams", 1, "duration_ts"), 165335),
+        (("streams", 1, "time_base"), "1/48000"),
+        (("streams", 1, "duration_ts"), True),
+        (("format", "duration"), "5.2"),
+        (("format", "duration"), "nan"),
+        (("format", "duration"), "1e999999999"),
+        (("streams", 0, "time_base"), "1/" + "9" * 60),
+        (("format", "start_time"), "1"),
+        (("format", "format_name"), "matroska,webm"),
+    ],
+)
+def test_inconsistent_probe_rejects_before_any_full_decode(codecs, path, value):
+    values, calls = codecs
+    target = values["metadata"]
+    for key in path[:-1]:
+        target = target[key]
+    target[path[-1]] = value
+    with pytest.raises(output.OutputRejected):
+        inspect()
+    assert len(calls) == 1
+
+
+@pytest.mark.parametrize("case", ["count", "gap", "size", "hash", "clock", "geometry"])
+def test_full_frame_observations_cannot_be_replaced_by_probe_counts(codecs, case):
+    values, calls = codecs
+    changes = {
+        "count": lambda: video_rows(123),
+        "gap": lambda: video_rows().replace(b"0, 1, 1,", b"0, 2, 2,", 1),
+        "size": lambda: video_rows().replace(b"1548288", b"1548287", 1),
+        "hash": lambda: video_rows().replace(b"a" * 32, b"not a digest", 1),
+        "clock": lambda: video_rows().replace(b"1/24", b"1/30", 1),
+        "geometry": lambda: video_rows(size=(768, 768)),
+    }
+    values["video"] = changes[case]()
+    with pytest.raises(output.OutputRejected):
+        inspect()
+    assert len(calls) == 2
+
+
+@pytest.mark.parametrize("samples", [165333, 166358])
+def test_audio_decode_cannot_underrun_presentation_or_hide_multiple_padding_blocks(codecs, samples):
+    values, _ = codecs
+    values["audio"] = audio_rows(samples=samples)
+    with pytest.raises(output.OutputRejected):
+        inspect()
+
+
+def test_audio_requires_contiguous_full_decode_not_just_correct_metadata(codecs):
+    values, _ = codecs
+    values["audio"] = audio_rows().replace(b"0, 4096, 4096,", b"0, 8192, 8192,", 1)
+    with pytest.raises(output.OutputRejected):
+        inspect()
+
+
+@pytest.mark.parametrize(
+    "fault",
+    [
+        OSError("private path"),
+        subprocess.TimeoutExpired("private command", 30),
+        subprocess.CalledProcessError(1, "private path"),
+        ValueError("private diagnostics"),
+    ],
+)
+def test_tool_failures_remain_uncertain_and_redacted(codecs, monkeypatch, fault):
+    tool = Mock(side_effect=fault)
+    gate = Mock()
+    gate.acquire.return_value = True
+    monkeypatch.setattr(output, "_SLOTS", gate)
+    monkeypatch.setattr(output.media_decoder, "run_tool", tool)
+    with pytest.raises(output.OutputCheckUnavailable) as error:
+        inspect()
+    assert "private" not in str(error.value)
+    assert tool.call_count == 1
+    gate.release.assert_called_once_with()
+
+
+def test_missing_codecs_or_busy_capacity_never_decode(codecs, monkeypatch):
+    _, calls = codecs
+    monkeypatch.setattr(output.media_decoder, "available", lambda: False)
+    with pytest.raises(output.OutputCheckUnavailable):
+        inspect()
+    monkeypatch.setattr(output.media_decoder, "available", lambda: True)
+    gate = Mock()
+    gate.acquire.return_value = False
+    monkeypatch.setattr(output, "_SLOTS", gate)
+    with pytest.raises(output.OutputCheckUnavailable):
+        inspect()
+    assert not calls
+    gate.release.assert_not_called()
+
+
+@pytest.mark.parametrize("raw", [b"", b"not a video", RAW[:16], b"\0\0\0\x11" + RAW[4:]])
+def test_invalid_bytes_reject_without_probe(codecs, raw):
+    _, calls = codecs
+    with pytest.raises(output.OutputRejected):
+        inspect(raw)
+    assert not calls
+
+
+def test_input_contract_rejects_before_codec_capacity(codecs):
+    _, calls = codecs
+    with pytest.raises(ValueError):
+        inspect(frames=125)
+    assert not calls
+
+
+@pytest.fixture(scope="module")
+def synthetic_clips(tmp_path_factory):
+    if not output.media_decoder.available():
+        pytest.skip("bounded CPU codec verification unavailable")
+    decoder = shutil.which("ffmpeg")
+    assert decoder
+    root, clips = tmp_path_factory.mktemp("h3-output"), {}
+    for rate in (32_000, 48_000):
+        path = root / f"synthetic-{rate}.mp4"
+        subprocess.run(
+            [
+                decoder,
+                "-v",
+                "error",
+                "-nostdin",
+                "-f",
+                "lavfi",
+                "-i",
+                "color=c=red:size=768x768:rate=24",
+                "-f",
+                "lavfi",
+                "-i",
+                f"sine=frequency=440:sample_rate={rate}",
+                "-map",
+                "0:v:0",
+                "-map",
+                "1:a:0",
+                "-frames:v",
+                "124",
+                "-t",
+                str(124 / 24),
+                "-c:v",
+                "libx264",
+                "-preset",
+                "ultrafast",
+                "-threads",
+                "1",
+                "-pix_fmt",
+                "yuv420p",
+                "-c:a",
+                "aac",
+                "-ac",
+                "2",
+                "-movie_timescale",
+                "96000",
+                str(path),
+            ],
+            capture_output=True,
+            timeout=30,
+            check=True,
+        )
+        clips[rate] = path.read_bytes()
+    return clips
+
+
+@pytest.mark.parametrize("rate", [32_000, 48_000])
+def test_actual_cpu_probe_and_full_decode_of_both_output_clocks(synthetic_clips, rate):
+    options = {"appearance_image": True, "audio_mode": "preserve"} if rate == 48_000 else {}
+    result = inspect(synthetic_clips[rate], width=768, height=768, **options)
+    assert result.frames == 124 and result.audio_sample_rate == rate
+    assert result.sha256 == hashlib.sha256(synthetic_clips[rate]).hexdigest()
+    assert abs(Fraction(result.audio_presentation_samples, rate) - Fraction(124, 24)) <= Fraction(1, rate)
+
+
+def test_real_decoder_corruption_never_becomes_a_passing_receipt(synthetic_clips):
+    with pytest.raises((output.OutputRejected, output.OutputCheckUnavailable)):
+        inspect(synthetic_clips[32_000][:1024], width=768, height=768)
+
+
+@pytest.mark.parametrize("count", [1, 3])
+def test_missing_or_extra_media_streams_are_not_silent_downgrades(codecs, count):
+    values, calls = codecs
+    streams = values["metadata"]["streams"]
+    values["metadata"]["streams"] = streams[:1] if count == 1 else [*streams, dict(streams[1])]
+    with pytest.raises(output.OutputRejected):
+        inspect()
+    assert len(calls) == 1
+
+
+def test_raw_byte_bound_rejects_without_allocating_a_large_fixture(codecs, monkeypatch):
+    _, calls = codecs
+    monkeypatch.setattr(output, "MAX_VIDEO_BYTES", 16)
+    with pytest.raises(output.OutputRejected):
+        inspect()
+    assert not calls
+
+
+def test_invalid_codec_response_type_is_unavailable_not_a_paid_result(codecs, monkeypatch):
+    monkeypatch.setattr(output.media_decoder, "run_tool", lambda *_args, **_kwargs: "not bytes")
+    with pytest.raises(output.OutputCheckUnavailable):
+        inspect()
+
+
+def test_actual_maximum_portrait_frame_and_audio_observations_fit_the_codec_bounds(tmp_path):
+    if not output.media_decoder.available():
+        pytest.skip("bounded CPU codec verification unavailable")
+    decoder = shutil.which("ffmpeg")
+    assert decoder
+    path = tmp_path / "synthetic-boundary.mp4"
+    subprocess.run(
+        [
+            decoder,
+            "-v",
+            "error",
+            "-nostdin",
+            "-f",
+            "lavfi",
+            "-i",
+            "color=c=red:size=768x1344:rate=24",
+            "-f",
+            "lavfi",
+            "-i",
+            "sine=frequency=440:sample_rate=48000",
+            "-map",
+            "0:v:0",
+            "-map",
+            "1:a:0",
+            "-frames:v",
+            "362",
+            "-t",
+            str(362 / 24),
+            "-c:v",
+            "libx264",
+            "-preset",
+            "ultrafast",
+            "-threads",
+            "1",
+            "-pix_fmt",
+            "yuv420p",
+            "-c:a",
+            "aac",
+            "-ac",
+            "2",
+            "-movie_timescale",
+            "96000",
+            str(path),
+        ],
+        capture_output=True,
+        timeout=30,
+        check=True,
+    )
+    result = inspect(path.read_bytes(), frames=362, width=768, height=1344, appearance_image=True, audio_mode="preserve")
+    assert result.frames == 362 and result.audio_sample_rate == 48_000
+    assert Fraction(result.duration_numerator, result.duration_denominator) == Fraction(181, 12)
+    assert abs(result.audio_presentation_samples - 724_000) <= 1
+    assert 0 <= result.aac_padding_samples <= 1023
diff --git a/grid_api/services/tests/test_fasth3_worker_handoff.py b/grid_api/services/tests/test_fasth3_worker_handoff.py
index 0e26aca0..19d92c35 100644
--- a/grid_api/services/tests/test_fasth3_worker_handoff.py
+++ b/grid_api/services/tests/test_fasth3_worker_handoff.py
@@ -33,12 +33,12 @@
     ],
 )
 def test_actual_core_source_plan_reaches_candidate_worker_without_dispatch(tmp_path, appearance, opening, mode):
-    source = os.environ.get("MEDIA_VIDEO_WORKER_SOURCE")
+    source = os.environ.get("FASTH3_WORKER_SOURCE")
     if not source:
-        pytest.skip("reviewed MEDIA_VIDEO_WORKER_SOURCE is required for FastH3 handoff")
+        pytest.skip("reviewed FASTH3_WORKER_SOURCE is required for FastH3 handoff")
     worker = Path(source).resolve()
     # Keep the venv entry path: resolving its symlink selects bare Python.
-    python = Path(os.environ.get("MEDIA_VIDEO_WORKER_PYTHON", str(worker / ".venv/bin/python"))).absolute()
+    python = Path(os.environ.get("FASTH3_WORKER_PYTHON", str(worker / ".venv/bin/python"))).absolute()
     if not python.is_file() or not (worker / "bridge/fasth3_sources.py").is_file():
         pytest.fail("FastH3 handoff requires its reviewed worker checkout and test Python")
 

From 1105e974e5e128249499dea9b020efbb8325c882 Mon Sep 17 00:00:00 2001
From: halfaipg <justin@techtoaster.com>
Date: Tue, 20 Oct 2026 19:42:44 -0400
Subject: [PATCH 5/5] feat(video): freeze private FastH3 candidate plans

Reuse the existing private upload and frozen-plan lifecycle while keeping H3 outside public and economic admission. Recheck upload privacy, primitive identity, and staged commitments. Share exact storage-location inspection without widening live source-role or deletion authority.
---
 grid_api/services/AGENTS.md                   |  20 +-
 grid_api/services/fasth3_plans.py             | 103 ++++++++
 grid_api/services/media_source_retention.py   |  19 +-
 grid_api/services/tests/test_fasth3_plans.py  | 219 ++++++++++++++++++
 .../tests/test_fasth3_worker_handoff.py       |  37 +--
 .../tests/test_media_source_retention.py      |  10 +-
 grid_api/services/tests/test_video_plans.py   |  19 ++
 grid_api/services/video_plans.py              |  17 +-
 8 files changed, 414 insertions(+), 30 deletions(-)
 create mode 100644 grid_api/services/fasth3_plans.py
 create mode 100644 grid_api/services/tests/test_fasth3_plans.py

diff --git a/grid_api/services/AGENTS.md b/grid_api/services/AGENTS.md
index 36b2278c..a591a774 100644
--- a/grid_api/services/AGENTS.md
+++ b/grid_api/services/AGENTS.md
@@ -36,6 +36,13 @@ content sanitization, and reward settlement.
   Codec/capacity/runtime failures remain uncertain, not failed-render/refund
   authority. No live freeze/settlement caller exists; a reviewed immutable-byte
   handoff, native identity/sound/lip-sync tests and paid lifecycle remain gates.
+- `fasth3_plans.py` composes the existing frozen-video planner and private-source
+  uploader with H3's validated role/graph snapshot. Only Core-issued private
+  storage locations can replace placeholders; never re-resolve or re-decode
+  after compilation. Keep raw bytes and prompts out of reprs. The candidate
+  deliberately has no governed recipe root and cannot enter the live journal,
+  reserve, dispatch or authorize deletion. It is not authenticated/account-bound,
+  qualified or priced. Native/ownership/financial integration remains a gate.
 
 - Live book `2026-10-20-a` adds the owner-approved `MOSS Image SFT BF16`
   image-only launch peg of USD0.010/output on Core0d2ee0b7.
@@ -162,6 +169,11 @@ content sanitization, and reward settlement.
   partial failure. Cancellation retains its single-flight guard until storage
   work finishes. Failures rotate behind other pending jobs, not starving them.
   Cleanup markers are operational metadata; no credits, results or ledgers move.
+  `capture_urls` shares exact Core-origin/private-bucket/key validation between
+  compiled protocols without granting ownership or deletion authority. The
+  live collector retains its LTX/Ingredients role restrictions. Preserve the
+  first-pin duplicate: nineteen URL references may represent eighteen unique
+  objects. H3 candidate inspection cannot create a cleanup journal row.
   It continues after admission rollback. Retain old bucket permissions until
   their recorded jobs drain; changing the S3 account or making a recorded bucket
   the output bucket refuses deletion. ComfyUI files, output/attempt orphans and
@@ -238,8 +250,12 @@ content sanitization, and reward settlement.
   before quota or storage. Planning uses non-routable source slots; only their
   locations change during staging. Keep source bytes out of serialized snapshots
   and reprs. Recheck commitments before uploads and copy mutable metadata before
-  yielding. Returned dictionaries cannot mutate the snapshot. This is not a
-  durable quote, frozen tariff, payment authorization, recovery selector or
+  yielding. Returned dictionaries cannot mutate the snapshot.
+  Before staging, recheck the private-upload policy and exact canonical JSON
+  types/seeds as well as byte commitments. A replaced private flag must never
+  turn a frozen private image into a legacy public-bucket upload. Model-specific
+  staging validation is a compiler-selected method, never a client field.
+  This is not a durable quote, frozen tariff, payment authorization, recovery selector or
   public graph interface. Core still prices and reserves through the existing
   live-credit authority; x402 quote/payment/refund implementation remains absent.
 - `media.compile_dispatch` is the non-I/O governed compiler; the existing
diff --git a/grid_api/services/fasth3_plans.py b/grid_api/services/fasth3_plans.py
new file mode 100644
index 00000000..b61b7691
--- /dev/null
+++ b/grid_api/services/fasth3_plans.py
@@ -0,0 +1,103 @@
+# SPDX-FileCopyrightText: 2026 AI Power Grid
+# SPDX-License-Identifier: AGPL-3.0-or-later
+
+"""Private H3 candidate planning/staging, outside public or financial admission."""
+
+import hashlib
+from dataclasses import dataclass
+from typing import Any
+
+from fastapi import HTTPException
+
+from . import fasth3_sources, media_source_retention, video_plans
+from .fasth3_contract import MODEL
+from .media_jobs import MAX_PAYLOAD_BYTES
+
+SCHEMA = "aipg.fasth3-plan.v1"
+SOURCE_SCHEMA = "aipg.fasth3-sources.v1"
+FEATURE = "recipe-fasth3-sources-v1"
+
+
+@dataclass(frozen=True)
+class CandidateH3Plan(video_plans.FrozenVideoPlan):
+    """Reuse frozen source transport, never synthesize a governed recipe root."""
+
+    def check_inputs(self) -> None:
+        super().check_inputs()
+        payload = self.payload
+        if (
+            self.snapshot.get("schema") != SCHEMA
+            or self.model != MODEL
+            or payload.get("qualification_only") is not True
+            or "recipe_root" in payload
+            or payload.get("recipe_source_schema") != SOURCE_SCHEMA
+            or payload.get("required_media_features") != [FEATURE]
+        ):
+            raise HTTPException(422, detail="Invalid FastH3 qualification plan")
+
+    def _validate_staged_sources(self, payload: dict[str, Any]) -> None:
+        # FrozenVideoPlan.stage independently compares every non-URL source
+        # commitment with the fully decoded Core inputs after this check.
+        expected = self.snapshot["payload"]["recipe_source_bindings"]
+        sources = payload.get("source_media", [])
+        if (
+            payload.get("recipe_source_bindings") != expected
+            or not isinstance(sources, list)
+            or any(not isinstance(source, dict) for source in sources)
+            or [source.get("role") for source in sources] != [binding["role"] for binding in expected]
+        ):
+            raise HTTPException(422, detail="FastH3 source roles changed during staging")
+        try:
+            media_source_retention.capture_urls([source["url"] for source in sources])
+        except (ValueError, KeyError, TypeError, AttributeError):
+            raise HTTPException(503, detail="Private FastH3 source staging failed") from None
+
+
+def compile_candidate(inputs: dict[str, Any], inline_sources: dict[str, str]) -> CandidateH3Plan:
+    """Freeze validated sources and exact graph without quota/storage/money I/O.
+
+    The plan is not authenticated, account-owned, priced or runtime-qualified.
+    Ingress must provide all those authorities before any eventual admission.
+    Stage uses existing private uploads; no source URL may enter compilation.
+    """
+    candidate, sources = fasth3_sources.prepare_candidate(inputs, inline_sources)
+    execution = candidate["execution"]
+    seed = execution["seed"]
+    validated = video_plans.VideoInputs(
+        model=MODEL,
+        payload_json=video_plans._json({**execution, "n": 1}),
+        seeds=(seed,),
+        images=(),
+        sources=sources,
+        timed_images=False,
+        private_images=True,
+    )
+    uncompiled = validated.planning_payload()
+    payload = {
+        **uncompiled,
+        "model": MODEL,
+        "execution": execution,
+        "recipe_engine": "comfyui",
+        "recipe_spec": candidate["spec"],
+        "recipe_source_schema": SOURCE_SCHEMA,
+        "recipe_source_bindings": candidate["source_bindings"],
+        "source_media": uncompiled.get("source_media", []),
+        "required_media_features": [FEATURE],
+        "qualification_only": True,
+    }
+    if len(video_plans._json(payload)) > MAX_PAYLOAD_BYTES:
+        raise HTTPException(413, detail="FastH3 dispatch exceeds its candidate limit")
+    snapshot = video_plans._json(
+        {
+            "schema": SCHEMA,
+            "model": MODEL,
+            "payload": payload,
+            "route_models": [MODEL],
+            "execution": execution,
+            "inputs": validated.commitments,
+            "uncompiled": uncompiled,
+        },
+    )
+    result = CandidateH3Plan(MODEL, snapshot, hashlib.sha256(snapshot.encode("ascii")).hexdigest(), validated)
+    result.check_inputs()
+    return result
diff --git a/grid_api/services/media_source_retention.py b/grid_api/services/media_source_retention.py
index 2c71c04a..6d4d2401 100644
--- a/grid_api/services/media_source_retention.py
+++ b/grid_api/services/media_source_retention.py
@@ -66,6 +66,19 @@ def capture(payload: dict[str, Any]) -> dict[str, Any] | None:
         raise ValueError("three private sources require the separate Ingredients profile")
     urls.extend(images)
     urls.extend(item.get("url") for item in media)
+    return capture_urls(urls)
+
+
+def capture_urls(urls: list[Any]) -> dict[str, Any] | None:
+    """Inspect exact Core storage locations, not owner or deletion authority.
+
+    Each compiler must independently validate its input roles first. Only the
+    existing financial hold/journal transaction may persist deletion authority.
+    """
+    # Legacy timed images repeat their first URL in source_image_url. Preserve
+    # the eighteen-unique-object bound, including that one duplicate reference.
+    if not isinstance(urls, list) or len(urls) > MAX_OBJECTS + 1:
+        raise ValueError("private source URL collection is invalid")
     if not urls:
         return None
     endpoint, bucket = _endpoint(), storage.private_source_bucket()
@@ -162,7 +175,7 @@ async def sweep(*, limit: int = 2) -> dict[str, int]:
                             media_jobs.c.job_id.in_(_closed()),
                         )
                         .order_by(sa.func.coalesce(media_jobs.c.source_cleanup_attempted_at, media_jobs.c.created), media_jobs.c.job_id)
-                        .limit(limit)
+                        .limit(limit),
                     )
                 )
                 .mappings()
@@ -178,7 +191,7 @@ async def sweep(*, limit: int = 2) -> dict[str, int]:
                         media_jobs.c.source_cleanup_at.is_(None),
                         media_jobs.c.job_id.in_(_closed()),
                     )
-                    .values(source_cleanup_attempted_at=datetime.now(UTC))
+                    .values(source_cleanup_attempted_at=datetime.now(UTC)),
                 )
                 await session.commit()
             task = asyncio.create_task(asyncio.to_thread(delete, row["source_objects"]))
@@ -199,7 +212,7 @@ async def sweep(*, limit: int = 2) -> dict[str, int]:
                         media_jobs.c.source_cleanup_at.is_(None),
                         media_jobs.c.job_id.in_(_closed()),
                     )
-                    .values(source_cleanup_at=datetime.now(UTC))
+                    .values(source_cleanup_at=datetime.now(UTC)),
                 )
                 await session.commit()
                 counts["deleted"] += result.rowcount
diff --git a/grid_api/services/tests/test_fasth3_plans.py b/grid_api/services/tests/test_fasth3_plans.py
new file mode 100644
index 00000000..73c50cd3
--- /dev/null
+++ b/grid_api/services/tests/test_fasth3_plans.py
@@ -0,0 +1,219 @@
+# SPDX-FileCopyrightText: 2026 AI Power Grid
+# SPDX-License-Identifier: AGPL-3.0-or-later
+
+"""Frozen candidate staging against existing storage mechanics, not admission."""
+
+import dataclasses
+import hashlib
+import json
+from unittest.mock import AsyncMock, Mock
+from uuid import uuid4
+
+import pytest
+from fastapi import HTTPException
+
+from grid_api.services import credits, fasth3_sources, job_queue, media_jobs, quota, recipes, storage
+from grid_api.services import fasth3_plans as plans
+from grid_api.services import media_source_retention as retention
+from grid_api.services.tests.test_fasth3_sources import inline, png, wav
+
+ENDPOINT = "https://" + "a" * 32 + ".r2.cloudflarestorage.com"
+BUCKET = "synthetic-private-sources"
+MODES = [
+    (a, o, m) for a in (False, True) for o in (False, True) for m in ("generated", "reference", "preserve") if m == "generated" or a or o
+]
+
+
+def forbidden(*_args, **_kwargs):
+    pytest.fail("candidate staging must not authorize, dispatch, delete or use real storage")
+
+
+@pytest.fixture(autouse=True)
+def boundaries(monkeypatch):
+    monkeypatch.setenv("R2_TRANSIENT_ACCOUNT", ENDPOINT)
+    monkeypatch.setenv("R2_TRANSIENT_BUCKET", "synthetic-public-outputs")
+    preflight = Mock(return_value=BUCKET)
+    monkeypatch.setattr(storage, "private_source_bucket", preflight)
+    for owner, name in [
+        (storage, "_client"),
+        (storage, "upload_private_source"),
+        (storage, "upload_source"),
+        (credits, "authorize_media"),
+        (job_queue, "submit_job"),
+        (quota, "check_and_consume"),
+        (media_jobs, "reserve"),
+        (retention, "delete"),
+        (retention, "sweep"),
+        (recipes, "resolve_for_model"),
+    ]:
+        monkeypatch.setattr(owner, name, forbidden)
+    return preflight
+
+
+def candidate(appearance=True, opening=True, mode="preserve", **changes):
+    values = {
+        "prompt": "Private synthetic scene.",
+        "seed": 0,
+        "appearance_image": appearance,
+        "opening_image": opening,
+        "audio_mode": mode,
+        **changes,
+    }
+    bundle = {}
+    if appearance:
+        bundle["appearance_image"] = inline(png())
+    if opening:
+        bundle["opening_image"] = inline(png((1344, 768)))
+    if mode != "generated":
+        bundle["soundtrack" if mode == "preserve" else "conditioning_audio"] = inline(wav(248_000 if mode == "preserve" else 96_000))
+    return plans.compile_candidate(values, bundle)
+
+
+def mock_uploads(monkeypatch, callback=None):
+    uploaded = []
+
+    def upload(raw, ext):
+        uploaded.append((raw, ext))
+        if callback:
+            callback()
+        return f"{ENDPOINT}/{BUCKET}/source/{len(uploaded):032x}.{ext}?signature=synthetic"
+
+    monkeypatch.setattr(storage, "upload_private_source", upload)
+    return uploaded
+
+
+@pytest.mark.parametrize("appearance,opening,mode", MODES)
+@pytest.mark.asyncio
+async def test_all_roles_use_shared_frozen_staging_and_only_core_private_urls(boundaries, monkeypatch, appearance, opening, mode):
+    plan = candidate(appearance, opening, mode)
+    boundaries.assert_not_called()
+    assert isinstance(plan, plans.video_plans.FrozenVideoPlan)
+    assert plan.sha256 == hashlib.sha256(plan.snapshot_json.encode("ascii")).hexdigest()
+    assert plan.snapshot["schema"] == plans.SCHEMA
+    assert plan.payload["qualification_only"] is True
+    assert "recipe_root" not in plan.payload
+    assert "Private synthetic scene." not in repr(plan) and "Private synthetic scene." not in repr(plan.inputs)
+    assert "data:" not in plan.snapshot_json
+    assert plan.route_models == [plans.MODEL]
+    saved = plan.snapshot_json
+    graph = plan.payload["recipe_spec"]
+    uploads = mock_uploads(monkeypatch)
+    # Staging must consume the frozen graph, not current recipes or decoders.
+    monkeypatch.setattr(fasth3_sources, "prepare_candidate", forbidden)
+    view = plan.payload
+    view["recipe_spec"].clear()
+    view["execution"]["seed"] = 99
+    staged = await plan.stage()
+    assert staged["recipe_spec"] == graph and staged["seed"] == staged["execution"]["seed"] == 0
+    assert plan.snapshot_json == saved
+    assert len(uploads) == len(plan.inputs.sources)
+    for observed, expected, uploaded in zip(staged["source_media"], plan.inputs.sources, uploads):
+        assert {key: value for key, value in observed.items() if key != "url"} == expected.commitment
+        assert uploaded == (expected.data, expected.ext)
+        assert observed["url"].startswith(f"{ENDPOINT}/{BUCKET}/source/")
+    record = retention.capture_urls([source["url"] for source in staged["source_media"]])
+    assert (record is not None) == bool(plan.inputs.sources)
+    if record:
+        assert "signature" not in json.dumps(record) and len(record["keys"]) == len(uploads)
+        retention.validate(record)
+    else:
+        boundaries.assert_not_called()
+
+
+@pytest.mark.parametrize("mutation", ["private", "seed-bool", "seeds", "metadata", "data", "graph"])
+@pytest.mark.asyncio
+async def test_changed_private_inputs_reject_before_even_private_preflight(boundaries, mutation):
+    plan = candidate()
+    if mutation == "private":
+        plan = dataclasses.replace(plan, inputs=dataclasses.replace(plan.inputs, private_images=False))
+    elif mutation == "seed-bool":
+        payload = plan.inputs.payload
+        payload["seed"] = False
+        plan = dataclasses.replace(plan, inputs=dataclasses.replace(plan.inputs, payload_json=json.dumps(payload)))
+    elif mutation == "seeds":
+        plan = dataclasses.replace(plan, inputs=dataclasses.replace(plan.inputs, seeds=(False,)))
+    elif mutation == "metadata":
+        plan.inputs.sources[0].metadata["width"] = float(plan.inputs.sources[0].metadata["width"])
+    elif mutation == "data":
+        first = dataclasses.replace(plan.inputs.sources[0], data=b"changed")
+        plan = dataclasses.replace(plan, inputs=dataclasses.replace(plan.inputs, sources=(first, *plan.inputs.sources[1:])))
+    else:
+        plan = dataclasses.replace(plan, snapshot_json=plan.snapshot_json.replace('"noise_seed":0', '"noise_seed":99'))
+    with pytest.raises(HTTPException) as error:
+        await plan.stage()
+    assert error.value.status_code == 422
+    boundaries.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_inflight_mutation_does_not_change_copied_source_metadata(monkeypatch):
+    plan = candidate()
+    mock_uploads(monkeypatch, lambda: plan.inputs.sources[-1].metadata.update(samples=1))
+    staged = await plan.stage()
+    assert staged["source_media"][-1]["samples"] == 248_000
+    assert plan.snapshot["inputs"]["sources"][-1]["samples"] == 248_000
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        f"{ENDPOINT}/synthetic-public-outputs/source/" + "a" * 32 + ".png?signature=synthetic",
+        f"{ENDPOINT}/{BUCKET}/video/" + "a" * 32 + ".png?signature=synthetic",
+        f"{ENDPOINT}/{BUCKET}/source/" + "a" * 32 + ".png",
+        "https://other.invalid/private.png?signature=synthetic",
+        "https://video-plan.invalid/appearance_image",
+    ],
+)
+@pytest.mark.asyncio
+async def test_foreign_public_unsigned_or_placeholder_locations_never_become_a_staged_plan(monkeypatch, url):
+    plan = candidate(True, False, "generated")
+    monkeypatch.setattr(storage, "upload_private_source", lambda *_args: url)
+    with pytest.raises(HTTPException) as error:
+        await plan.stage()
+    assert error.value.status_code == 503 and url not in error.value.detail
+
+
+@pytest.mark.asyncio
+async def test_candidate_staging_cannot_grant_a_recipe_root_or_live_journal_authority(monkeypatch):
+    # 209 lies on both H3 and LTX grids: the absence of governance must still reject.
+    plan = candidate(False, False, "generated", frames=209)
+    payload = await plan.stage()
+    with pytest.raises(media_jobs.MediaJobError, match="governed recipe commitment"):
+        media_jobs.prepare(
+            account_id=uuid4(),
+            idempotency_key=str(uuid4()),
+            request_hash="a" * 64,
+            model=plans.MODEL,
+            payload=payload,
+            route_models=plan.route_models,
+            max_cost_micro=1000,
+        )
+
+
+@pytest.mark.asyncio
+async def test_three_h3_sources_do_not_masquerade_as_live_ingredients_retention(monkeypatch):
+    plan = candidate()
+    mock_uploads(monkeypatch)
+    payload = await plan.stage()
+    with pytest.raises(ValueError, match="Ingredients"):
+        retention.capture(payload)
+
+
+def test_dispatch_bound_rejects_before_staging_or_upload(boundaries, monkeypatch):
+    monkeypatch.setattr(plans, "MAX_PAYLOAD_BYTES", 1)
+    with pytest.raises(HTTPException) as error:
+        candidate()
+    assert error.value.status_code == 413
+    boundaries.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_staged_commitment_types_cannot_change_without_changing_numeric_values(monkeypatch):
+    plan = candidate(True, False, "generated")
+    sources = plan.payload["source_media"]
+    sources[0]["url"] = f"{ENDPOINT}/{BUCKET}/source/" + "a" * 32 + ".png?signature=synthetic"
+    sources[0]["bytes"] = float(sources[0]["bytes"])
+    monkeypatch.setattr(plans.video_plans, "stage_inputs", AsyncMock(return_value={"source_media": sources}))
+    with pytest.raises(HTTPException) as error:
+        await plan.stage()
+    assert error.value.status_code == 422
diff --git a/grid_api/services/tests/test_fasth3_worker_handoff.py b/grid_api/services/tests/test_fasth3_worker_handoff.py
index 19d92c35..213637fe 100644
--- a/grid_api/services/tests/test_fasth3_worker_handoff.py
+++ b/grid_api/services/tests/test_fasth3_worker_handoff.py
@@ -14,7 +14,8 @@
 import pytest
 from PIL import Image
 
-from grid_api.services.fasth3_sources import prepare_candidate
+from grid_api.services import storage
+from grid_api.services.fasth3_plans import compile_candidate
 
 
 @pytest.mark.parametrize(
@@ -32,7 +33,8 @@
         (True, True, "preserve"),
     ],
 )
-def test_actual_core_source_plan_reaches_candidate_worker_without_dispatch(tmp_path, appearance, opening, mode):
+@pytest.mark.asyncio
+async def test_actual_core_source_plan_reaches_candidate_worker_without_dispatch(tmp_path, monkeypatch, appearance, opening, mode):
     source = os.environ.get("FASTH3_WORKER_SOURCE")
     if not source:
         pytest.skip("reviewed FASTH3_WORKER_SOURCE is required for FastH3 handoff")
@@ -58,7 +60,7 @@ def picture(size):
             writer.setparams((2, 2, 48_000, 0, "NONE", "not compressed"))
             writer.writeframes(b"\0" * (282_000 if mode == "preserve" else 96_000) * 4)
         bundle["soundtrack" if mode == "preserve" else "conditioning_audio"] = base64.b64encode(sound.getvalue()).decode()
-    plan, sources = prepare_candidate(
+    plan = compile_candidate(
         {
             "prompt": "A sunny synthetic park.",
             "seed": 0,
@@ -71,21 +73,20 @@ def picture(size):
         },
         bundle,
     )
-    content, committed = {}, []
-    for source_media in sources:
-        url = "https://" + "a" * 32 + f".r2.cloudflarestorage.com/synthetic/{source_media.role}.{source_media.ext}"
-        content[url] = base64.b64encode(source_media.data).decode()
-        committed.append({**source_media.commitment, "url": url})
-    payload = {
-        "recipe_engine": "comfyui",
-        "recipe_source_schema": "aipg.fasth3-sources.v1",
-        "required_media_features": ["recipe-fasth3-sources-v1"],
-        "n": 1,
-        "execution": plan["execution"],
-        "recipe_spec": plan["spec"],
-        "recipe_source_bindings": plan["source_bindings"],
-        "source_media": committed,
-    }
+    content = {}
+    endpoint = "https://" + "a" * 32 + ".r2.cloudflarestorage.com"
+    bucket = "synthetic-private-sources"
+    monkeypatch.setenv("R2_TRANSIENT_ACCOUNT", endpoint)
+    monkeypatch.setenv("R2_TRANSIENT_BUCKET", "synthetic-public-outputs")
+    monkeypatch.setattr(storage, "private_source_bucket", lambda: bucket)
+
+    def upload(raw, extension):
+        url = f"{endpoint}/{bucket}/source/{len(content):032x}.{extension}?signature=synthetic"
+        content[url] = base64.b64encode(raw).decode()
+        return url
+
+    monkeypatch.setattr(storage, "upload_private_source", upload)
+    payload = await plan.stage()
     script = """
 import asyncio, base64, copy, json, sys
 import httpx, respx
diff --git a/grid_api/services/tests/test_media_source_retention.py b/grid_api/services/tests/test_media_source_retention.py
index 7cfee7b6..2a29d322 100644
--- a/grid_api/services/tests/test_media_source_retention.py
+++ b/grid_api/services/tests/test_media_source_retention.py
@@ -53,10 +53,18 @@ def test_capture_records_only_own_unique_inputs_without_signed_queries():
     retention.validate(RECORD)
 
 
+def test_url_capture_reuses_origin_and_key_bounds_without_changing_live_roles():
+    assert retention.capture_urls([URL]) == RECORD
+    assert retention.capture_urls([]) is None
+    for values in ([URL] * 20, None, {"url": URL}):
+        with pytest.raises(ValueError):
+            retention.capture_urls(values)
+
+
 def test_all_sixteen_pins_and_two_media_sources_have_bounded_authority():
     urls = [f"{ENDPOINT}/{BUCKET}/source/{index:032x}.png?synthetic=1" for index in range(16)]
     record = retention.capture(
-        {"source_image_url": urls[0], "source_image_urls": urls, "source_media": [{"url": URL}, {"url": URL.replace(".wav", ".mp4")}]}
+        {"source_image_url": urls[0], "source_image_urls": urls, "source_media": [{"url": URL}, {"url": URL.replace(".wav", ".mp4")}]},
     )
     assert len(record["keys"]) == 18
     retention.validate(record)
diff --git a/grid_api/services/tests/test_video_plans.py b/grid_api/services/tests/test_video_plans.py
index 1c53521d..8397254f 100644
--- a/grid_api/services/tests/test_video_plans.py
+++ b/grid_api/services/tests/test_video_plans.py
@@ -351,3 +351,22 @@ def upload(_raw, ext):
     payload = await plan.stage()
     assert payload["source_media"][0]["samples"] == 18_000
     assert plan.snapshot["inputs"]["sources"][0]["samples"] == 18_000
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("change", ["public_upload", "seed_bool", "seed_tuple"])
+async def test_frozen_private_plan_cannot_switch_upload_policy_or_primitive_identity(admission, change):
+    inputs = await videos.validate_video(request("generated", pin=True), {}, async_submission=True)
+    plan = video_plans.compile_video(inputs)
+    if change == "public_upload":
+        inputs = dataclasses.replace(inputs, private_images=False)
+    elif change == "seed_tuple":
+        inputs = dataclasses.replace(inputs, seeds=(False,))
+    else:
+        payload = inputs.payload
+        payload["seed"] = False
+        inputs = dataclasses.replace(inputs, payload_json=json.dumps(payload))
+    plan = dataclasses.replace(plan, inputs=inputs)
+    with pytest.raises(HTTPException) as error:
+        await plan.stage()
+    assert error.value.status_code == 422
diff --git a/grid_api/services/video_plans.py b/grid_api/services/video_plans.py
index 0c5630f5..320b0d78 100644
--- a/grid_api/services/video_plans.py
+++ b/grid_api/services/video_plans.py
@@ -97,11 +97,17 @@ def check_inputs(self) -> None:
             hashlib.sha256(self.snapshot_json.encode("ascii")).hexdigest() != self.sha256
             or snapshot["model"] != self.model
             or self.inputs.model != self.model
-            or snapshot["inputs"] != json.loads(_json(self.inputs.commitments))
-            or snapshot["uncompiled"] != self.inputs.planning_payload()
+            or self.inputs.private_images is not True
+            or _json(list(self.inputs.seeds)) != _json([snapshot["payload"]["seed"]])
+            or _json(snapshot["inputs"]) != _json(self.inputs.commitments)
+            or _json(snapshot["uncompiled"]) != _json(self.inputs.planning_payload())
         ):
             raise HTTPException(422, detail="Video inputs changed after planning")
 
+    def _validate_staged_sources(self, payload: dict[str, Any]) -> None:
+        if payload.get("source_media"):
+            video_sources.validate_commitments(payload["source_media"], payload["recipe_source_bindings"], self.snapshot["execution"])
+
     async def stage(self) -> dict[str, Any]:
         """Bind private uploads to the frozen graph, without re-resolving it."""
         self.check_inputs()
@@ -133,10 +139,9 @@ async def stage(self) -> dict[str, Any]:
                     raise ValueError
             except (ValueError, TypeError, AttributeError):
                 raise HTTPException(503, detail="Private video source staging failed") from None
-        if payload.get("source_media"):
-            video_sources.validate_commitments(payload["source_media"], payload["recipe_source_bindings"], self.snapshot["execution"])
+        self._validate_staged_sources(payload)
         observed = [{key: value for key, value in source.items() if key != "url"} for source in payload.get("source_media", [])]
-        if observed != self.snapshot["inputs"]["sources"]:
+        if _json(observed) != _json(self.snapshot["inputs"]["sources"]):
             raise HTTPException(422, detail="Video sources changed during staging")
         return payload
 
@@ -171,7 +176,7 @@ def compile_video(inputs: VideoInputs) -> FrozenVideoPlan:
             "execution": recipe["execution"],
             "inputs": inputs.commitments,
             "uncompiled": uncompiled,
-        }
+        },
     )
     return FrozenVideoPlan(inputs.model, snapshot, hashlib.sha256(snapshot.encode("ascii")).hexdigest(), inputs)