danrh
07f3dc254f
Introduce model-specific target profiles, fail-closed grounding and validation, content-addressed Director caching, LTX-2.5 contracts, comparison workflows, benchmarks, schemas, documentation, and regression coverage.
990 lines
41 KiB
Python
990 lines
41 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import sys
|
|
import unittest
|
|
import uuid
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
COMFY_ROOT = ROOT.parents[1]
|
|
|
|
|
|
def load_nodes_module():
|
|
if str(COMFY_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(COMFY_ROOT))
|
|
package_name = f"diffusiongemma_ref2va_contract_patch_{uuid.uuid4().hex}"
|
|
spec = importlib.util.spec_from_file_location(
|
|
package_name,
|
|
ROOT / "__init__.py",
|
|
submodule_search_locations=[str(ROOT)],
|
|
)
|
|
if spec is None or spec.loader is None:
|
|
raise RuntimeError("could not load DiffusionGemma Prompt Builder")
|
|
package = importlib.util.module_from_spec(spec)
|
|
sys.modules[package_name] = package
|
|
spec.loader.exec_module(package)
|
|
return sys.modules[f"{package_name}.nodes"]
|
|
|
|
|
|
class MiniMaxH3Ref2VAContractPatchRegressionTests(unittest.TestCase):
|
|
"""Regression contract for a surgical post-refinement Ref2VA repair.
|
|
|
|
These tests intentionally require a path that is not yet implemented. The
|
|
repair may replace only the subject-definition body and the explicit (Sx)
|
|
tokens bound to an existing <Subject N>. It must still pass the complete H3
|
|
validator; the tests never waive or filter a validation reason.
|
|
"""
|
|
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.nodes = load_nodes_module()
|
|
|
|
@staticmethod
|
|
def reference_manifest() -> str:
|
|
return (
|
|
"<Picture 1>: [dg:identity,appearance,object,color] Mara identity, dark bobbed hair, blue coat, "
|
|
"and brass suitcase."
|
|
)
|
|
|
|
@staticmethod
|
|
def verified_ledger() -> dict:
|
|
return {
|
|
"schema": "dg-grounding-ledger/1",
|
|
"analysis_status": "grounded",
|
|
"observed_facts": [
|
|
{
|
|
"fact_id": "fact-picture-1",
|
|
"claim": "Picture 1 shows Mara with dark bobbed hair, a blue coat, and a brass suitcase.",
|
|
"confidence": "high",
|
|
"categories": ["identity", "appearance", "object", "color"],
|
|
"evidence": [
|
|
{
|
|
"asset_id": "picture:1",
|
|
"sample_ordinal": 1,
|
|
"source_frame_index": 0,
|
|
"timecode_seconds": 0.0,
|
|
}
|
|
],
|
|
"typed_claims": [],
|
|
}
|
|
],
|
|
"inferred_facts": [],
|
|
"creative_additions": [],
|
|
"uncertainties": [],
|
|
"grounding_failure_reasons": [],
|
|
}
|
|
|
|
def runtime(self):
|
|
return self.nodes.RuntimeConfig(
|
|
model_path="unused-local-checkpoint",
|
|
backend="transformers_inprocess",
|
|
dtype="auto",
|
|
quantization="none",
|
|
local_files_only=True,
|
|
unload_policy="unload_after_run",
|
|
max_memory_gb=20.0,
|
|
status={"ready": True, "supports_pixels": False},
|
|
)
|
|
|
|
def context(self):
|
|
return self.nodes.GemmaContext(
|
|
user_prompt="Create exactly two shots in a quiet train compartment.",
|
|
images=None,
|
|
source="minimax_h3_ref2va",
|
|
media_metadata={
|
|
"source": "minimax_h3_ref2va",
|
|
"minimax_h3_reference_manifest": self.reference_manifest(),
|
|
"verified_grounding_ledger": self.verified_ledger(),
|
|
"grounding_evidence_report_id": "dggr-ref2va-contract-test",
|
|
"grounding_required_asset_ids": ["picture:1"],
|
|
"visual_grounding_mode": "verified_ledger",
|
|
"pixels_sent_to_backend": False,
|
|
},
|
|
)
|
|
|
|
def target(self):
|
|
return self.nodes.TargetProfileConfig(
|
|
target_profile="minimax_h3",
|
|
audio_mode="auto_scene_audio",
|
|
target_duration_seconds=12.0,
|
|
minimax_h3_mode="ref2va",
|
|
minimax_h3_shot_count="2",
|
|
minimax_h3_dialogue_mode="required",
|
|
minimax_h3_dialogue_line_count=2,
|
|
minimax_h3_dialogue_guidance=(
|
|
"Mara says We leave at dawn, then says Then we are ready."
|
|
),
|
|
)
|
|
|
|
@staticmethod
|
|
def valid_prompt() -> str:
|
|
return (
|
|
"subject_definitions:\n"
|
|
"<Subject 1>: Mara, a traveler whose face, dark bobbed hair, blue coat, and brass suitcase derive from "
|
|
"<Picture 1>.\n\n"
|
|
"summary:\n"
|
|
"[reference generation] <Subject 1> shares a decision before closing the suitcase.\n\n"
|
|
"retention_analysis:\n"
|
|
"<Subject 1>: fully_preserved - Her face, dark bobbed hair, blue coat, proportions, and brass suitcase "
|
|
"remain stable in both shots.\n\n"
|
|
"detailed_description:\n"
|
|
"Realistic live-action practical film photography uses soft window light, restrained blue-brown color, "
|
|
"shallow depth of field, and fine grain throughout. [Shot 1] A locked medium shot holds <Subject 1> (S1) "
|
|
"on the left side of a quiet train compartment, matching the face, dark bobbed hair, blue coat, and brass "
|
|
"suitcase derived from <Picture 1>. Passing dawn light slides across the wall while she keeps both hands on "
|
|
"the suitcase. <Subject 1> (S1) looks toward an unseen companion and says <d>[English] We leave at "
|
|
"dawn.</d> in a steady voice. The camera remains static while her mouth forms the words and her right hand "
|
|
"moves to the latch. Low train vibration subtly moves the hanging curtain without changing her position or "
|
|
"identity. [Shot 2] At 00:06.000, the camera cuts to a static close-up of <Subject 1> (S1) beside the brass "
|
|
"latch. She answers her own uncertainty with <d>[English] Then we are ready.</d> and closes the suitcase in "
|
|
"one visible motion. The latch clicks only when her thumb presses it; her blue sleeve settles against the "
|
|
"brass edge. The camera holds her resolved expression and the same soft window light as the train continues "
|
|
"forward. The final frame holds on <Subject 1> seated calmly beside the closed suitcase in the static close "
|
|
"composition.\n\n"
|
|
"overall_soundscape:\n"
|
|
"Low train rumble, curtain movement, cloth rustle, and the synchronized suitcase-latch click surround the "
|
|
"compartment without repeating speech.\n\n"
|
|
"non_diegetic_music:\n"
|
|
"N/A"
|
|
)
|
|
|
|
@classmethod
|
|
def invalid_subject_and_speaker_prompt(cls) -> str:
|
|
prompt = cls.valid_prompt().replace(
|
|
"<Subject 1>: Mara, a traveler whose face, dark bobbed hair, blue coat, and brass suitcase derive from "
|
|
"<Picture 1>.",
|
|
"<Subject 1>: Mara from <Picture 1>.",
|
|
1,
|
|
)
|
|
return prompt.replace("<Subject 1> (S1)", "<Subject 1>")
|
|
|
|
@classmethod
|
|
def live_colonless_speaker_only_prompt(cls) -> str:
|
|
prompt = cls.valid_prompt().replace(
|
|
"<Subject 1>: Mara, a traveler whose face, dark bobbed hair, blue coat, and brass suitcase derive from "
|
|
"<Picture 1>.",
|
|
"<Subject 1> Mara, a traveler whose face, dark bobbed hair, blue coat, and brass suitcase derive from "
|
|
"<Picture 1>.",
|
|
1,
|
|
).replace(
|
|
"<Subject 1>: fully_preserved - Her face, dark bobbed hair, blue coat, proportions, and brass suitcase "
|
|
"remain stable in both shots.",
|
|
"<Subject 1> fully_preserved - Her face, dark bobbed hair, blue coat, proportions, and brass suitcase "
|
|
"remain stable in both shots.",
|
|
1,
|
|
)
|
|
return prompt.replace("<Subject 1> (S1)", "<Subject 1>")
|
|
|
|
@classmethod
|
|
def ambiguous_speaker_only_prompt(cls) -> str:
|
|
prompt = cls.valid_prompt().replace(
|
|
"<Subject 1>: Mara, a traveler whose face, dark bobbed hair, blue coat, and brass suitcase derive from "
|
|
"<Picture 1>.",
|
|
"<Subject 1>: Mara, a traveler whose face, dark bobbed hair, blue coat, and brass suitcase derive from "
|
|
"<Picture 1>.\n"
|
|
"<Subject 2>: Ivo, a second traveler whose face, gray coat, and leather satchel also derive from "
|
|
"<Picture 1>.",
|
|
1,
|
|
).replace(
|
|
"<Subject 1>: fully_preserved - Her face, dark bobbed hair, blue coat, proportions, and brass suitcase "
|
|
"remain stable in both shots.",
|
|
"<Subject 1>: fully_preserved - Her face, dark bobbed hair, blue coat, proportions, and brass suitcase "
|
|
"remain stable in both shots.\n"
|
|
"<Subject 2>: fully_preserved - His face, gray coat, proportions, and leather satchel remain stable in both shots.",
|
|
1,
|
|
).replace(
|
|
"A locked medium shot holds <Subject 1> (S1) on the left side",
|
|
"A locked medium shot holds <Subject 1> and <Subject 2> on the left side",
|
|
1,
|
|
)
|
|
return prompt.replace("<Subject 1> (S1)", "<Subject 1>")
|
|
|
|
@classmethod
|
|
def conflicting_speaker_mapping_prompt(cls) -> str:
|
|
return cls.valid_prompt().replace(
|
|
"[Shot 2] At 00:06.000, the camera cuts to a static close-up of <Subject 1> (S1)",
|
|
"[Shot 2] At 00:06.000, the camera cuts to a static close-up of <Subject 1> (S2)",
|
|
1,
|
|
)
|
|
|
|
@classmethod
|
|
def standalone_speaker_collision_prompt(cls) -> str:
|
|
prompt = cls.valid_prompt().replace(
|
|
"<Subject 1>: Mara, a traveler whose face, dark bobbed hair, blue coat, and brass suitcase derive from "
|
|
"<Picture 1>.",
|
|
"<Subject 1>: Mara, a traveler whose face, dark bobbed hair, blue coat, and brass suitcase derive from "
|
|
"<Picture 1>.\n"
|
|
"<Subject 2>: Ivo, a second traveler whose face, gray coat, and leather satchel also derive from "
|
|
"<Picture 1>.",
|
|
1,
|
|
).replace(
|
|
"<Subject 1>: fully_preserved - Her face, dark bobbed hair, blue coat, proportions, and brass suitcase "
|
|
"remain stable in both shots.",
|
|
"<Subject 1>: fully_preserved - Her face, dark bobbed hair, blue coat, proportions, and brass suitcase "
|
|
"remain stable in both shots.\n"
|
|
"<Subject 2>: fully_preserved - His face, gray coat, proportions, and leather satchel remain stable in both shots.",
|
|
1,
|
|
)
|
|
shot_1, shot_2 = prompt.split("[Shot 2]", 1)
|
|
shot_1 = shot_1.replace("<Subject 1> (S1)", "<Subject 1>").replace(
|
|
"<d>[English] We leave at dawn.</d>",
|
|
"(S2) <d>[English] We leave at dawn.</d>",
|
|
1,
|
|
)
|
|
shot_2 = shot_2.replace("<Subject 1> (S1)", "<Subject 2>").replace(
|
|
"<Subject 1>",
|
|
"<Subject 2>",
|
|
)
|
|
return f"{shot_1}[Shot 2]{shot_2}"
|
|
|
|
@classmethod
|
|
def duplicate_standalone_speaker_prompt(cls) -> str:
|
|
return cls.standalone_speaker_collision_prompt().replace(
|
|
"<d>[English] Then we are ready.</d>",
|
|
"(S2) <d>[English] Then we are ready.</d>",
|
|
1,
|
|
)
|
|
|
|
@classmethod
|
|
def multisubject_explicit_pair_prompt(cls) -> str:
|
|
return cls.valid_prompt().replace(
|
|
"<Subject 1>: Mara, a traveler whose face, dark bobbed hair, blue coat, and brass suitcase derive from "
|
|
"<Picture 1>.",
|
|
"<Subject 1>: Mara, a traveler whose face, dark bobbed hair, blue coat, and brass suitcase derive from "
|
|
"<Picture 1>.\n"
|
|
"<Subject 2>: Ivo, a second traveler whose face, gray coat, and leather satchel also derive from "
|
|
"<Picture 1>.",
|
|
1,
|
|
).replace(
|
|
"<Subject 1>: fully_preserved - Her face, dark bobbed hair, blue coat, proportions, and brass suitcase "
|
|
"remain stable in both shots.",
|
|
"<Subject 1>: fully_preserved - Her face, dark bobbed hair, blue coat, proportions, and brass suitcase "
|
|
"remain stable in both shots.\n"
|
|
"<Subject 2>: fully_preserved - His face, gray coat, proportions, and leather satchel remain stable in both shots.",
|
|
1,
|
|
).replace(
|
|
"A locked medium shot holds <Subject 1> (S1) on the left side",
|
|
"A locked medium shot holds <Subject 1> and <Subject 2> on the left side",
|
|
1,
|
|
)
|
|
|
|
@classmethod
|
|
def native_output(cls, prompt_text: str) -> str:
|
|
return (
|
|
f"{prompt_text}\n\n"
|
|
"GROUNDING_EVIDENCE_REPORT_ID: dggr-ref2va-contract-test\n"
|
|
"USED_GROUNDING_FACT_IDS: fact-picture-1"
|
|
)
|
|
|
|
@staticmethod
|
|
def dialogue_sha256(dialogue_block: str) -> str:
|
|
return hashlib.sha256(dialogue_block.encode("utf-8")).hexdigest()
|
|
|
|
@classmethod
|
|
def valid_contract_sidecar(cls) -> str:
|
|
return json.dumps(
|
|
{
|
|
"schema": "dg-h3-ref2va-contract-patch/1",
|
|
"evidence_report_id": "dggr-ref2va-contract-test",
|
|
"subject_definitions": [
|
|
{
|
|
"subject_tag": "<Subject 1>",
|
|
"source_tags": ["<Picture 1>"],
|
|
"grounding_fact_ids": ["fact-picture-1"],
|
|
}
|
|
],
|
|
"dialogue_speakers": [
|
|
{
|
|
"shot": 1,
|
|
"dialogue_sha256": cls.dialogue_sha256(
|
|
"<d>[English] We leave at dawn.</d>"
|
|
),
|
|
"subject_tag": "<Subject 1>",
|
|
},
|
|
{
|
|
"shot": 2,
|
|
"dialogue_sha256": cls.dialogue_sha256(
|
|
"<d>[English] Then we are ready.</d>"
|
|
),
|
|
"subject_tag": "<Subject 1>",
|
|
},
|
|
],
|
|
},
|
|
separators=(",", ":"),
|
|
)
|
|
|
|
@classmethod
|
|
def speaker_only_contract_sidecar(cls) -> str:
|
|
payload = json.loads(cls.valid_contract_sidecar())
|
|
payload["subject_definitions"] = []
|
|
return json.dumps(payload, separators=(",", ":"))
|
|
|
|
def validation_reasons(self, prompt_text: str) -> list[str]:
|
|
return self.nodes._minimax_h3_prompt_validation_reasons(
|
|
prompt_text,
|
|
duration_seconds=12.0,
|
|
user_prompt=self.context().user_prompt,
|
|
audio_mode="auto_scene_audio",
|
|
max_prompt_chars=8000,
|
|
minimax_h3_mode="ref2va",
|
|
reference_manifest=self.reference_manifest(),
|
|
minimax_h3_shot_count="2",
|
|
minimax_h3_dialogue_mode="required",
|
|
minimax_h3_dialogue_line_count=2,
|
|
minimax_h3_dialogue_guidance=self.target().minimax_h3_dialogue_guidance,
|
|
)
|
|
|
|
@staticmethod
|
|
def without_patchable_contract_bytes(prompt_text: str) -> str:
|
|
sections_normalized = re.sub(
|
|
r"(?ms)(^subject_definitions:\n).*?(\n\nsummary:)",
|
|
r"\1<SUBJECT_DEFINITIONS>\2",
|
|
prompt_text,
|
|
count=1,
|
|
)
|
|
return re.sub(r"\s*\(S\d+\)", "", sections_normalized)
|
|
|
|
@classmethod
|
|
def compiler_prompt_with_nonlocal_defect(cls) -> str:
|
|
return cls.invalid_subject_and_speaker_prompt().replace(
|
|
"[Shot 2] At 00:06.000, the camera cuts to a static close-up",
|
|
"[Shot 2] At 00:12.000, the camera cuts to a static close-up",
|
|
1,
|
|
)
|
|
|
|
def run_generation(
|
|
self,
|
|
contract_sidecar: str,
|
|
*,
|
|
compiler_candidate: str | None = None,
|
|
first_refinement_candidate: str | None = None,
|
|
):
|
|
compiler_candidate = (
|
|
self.invalid_subject_and_speaker_prompt()
|
|
if compiler_candidate is None
|
|
else compiler_candidate
|
|
)
|
|
stages: list[str] = []
|
|
prompts: list[str] = []
|
|
|
|
def detailed(
|
|
_config,
|
|
prompt_text,
|
|
_media_context,
|
|
_max_tokens,
|
|
options,
|
|
**_kwargs,
|
|
):
|
|
stage = str(options.stage)
|
|
stages.append(stage)
|
|
prompts.append(prompt_text)
|
|
if stage == "compiler":
|
|
output = self.native_output(compiler_candidate)
|
|
elif stage == "target_repair_1" and first_refinement_candidate is not None:
|
|
output = self.native_output(first_refinement_candidate)
|
|
elif stage.startswith("ref2va_contract_patch_"):
|
|
output = contract_sidecar
|
|
else:
|
|
self.fail(f"unexpected backend stage: {stage}")
|
|
return self.nodes.BackendRunResult(decoded_text=output)
|
|
|
|
backend_results: list = []
|
|
call_budget = self.nodes.BackendCallBudget(max_calls=4)
|
|
# Production shares this budget with the evidence pass, which has
|
|
# already consumed one call before the compiler enters this function.
|
|
call_budget.claim()
|
|
options = self.nodes.BackendRunOptions(
|
|
stage="compiler",
|
|
call_budget=call_budget,
|
|
)
|
|
with patch.object(self.nodes, "_run_backend_detailed", side_effect=detailed):
|
|
result = self.nodes._run_generation_packet_legacy(
|
|
self.runtime(),
|
|
self.context(),
|
|
self.target(),
|
|
runtime_required=True,
|
|
max_new_tokens=2048,
|
|
max_output_chars=8000,
|
|
backend_options=options,
|
|
backend_results=backend_results,
|
|
manage_unload=False,
|
|
max_refinement_attempts_override=2,
|
|
refinement_evidence_context=json.dumps(
|
|
self.verified_ledger(), sort_keys=True
|
|
),
|
|
native_h3_output=True,
|
|
)
|
|
return result, stages, prompts, call_budget, backend_results
|
|
|
|
def test_ref2va_dialogue_requires_a_speaker_cue_at_each_dialogue_block(self) -> None:
|
|
missing_cues = self.valid_prompt().replace(
|
|
"<Subject 1> (S1)", "<Subject 1>"
|
|
)
|
|
|
|
self.assertEqual(self.validation_reasons(self.valid_prompt()), [])
|
|
self.assertIn(
|
|
"minimax_h3_dialogue_speaker_invalid",
|
|
self.validation_reasons(missing_cues),
|
|
)
|
|
|
|
def test_ref2va_dialogue_rejects_an_unstable_many_speakers_for_one_subject_mapping(self) -> None:
|
|
unstable = self.valid_prompt().replace(
|
|
"[Shot 2] At 00:06.000, the camera cuts to a static close-up of <Subject 1> (S1)",
|
|
"[Shot 2] At 00:06.000, the camera cuts to a static close-up of <Subject 1> (S2)",
|
|
1,
|
|
)
|
|
|
|
self.assertIn(
|
|
"minimax_h3_dialogue_speaker_invalid",
|
|
self.validation_reasons(unstable),
|
|
)
|
|
|
|
def test_ref2va_dialogue_rejects_zero_speaker_id(self) -> None:
|
|
invalid = self.valid_prompt().replace("(S1)", "(S0)")
|
|
|
|
self.assertEqual(
|
|
self.validation_reasons(invalid),
|
|
["minimax_h3_dialogue_speaker_invalid"],
|
|
)
|
|
|
|
def test_global_mapper_rejects_one_standalone_speaker_id_for_two_subjects(self) -> None:
|
|
duplicate = self.duplicate_standalone_speaker_prompt()
|
|
|
|
self.assertEqual(
|
|
self.validation_reasons(duplicate),
|
|
["minimax_h3_dialogue_speaker_invalid"],
|
|
)
|
|
subject_speakers, speaker_subjects, mapping_reasons = (
|
|
self.nodes._minimax_h3_ref2va_global_speaker_mappings(duplicate)
|
|
)
|
|
self.assertEqual(
|
|
mapping_reasons,
|
|
["ref2va_contract_patch_existing_speaker_mapping_ambiguous"],
|
|
)
|
|
self.assertEqual(subject_speakers, {"<Subject 1>": "S2"})
|
|
self.assertEqual(speaker_subjects, {"S2": "<Subject 1>"})
|
|
|
|
def test_multisubject_shot_accepts_explicit_adjacent_subject_speaker_pair(self) -> None:
|
|
explicit = self.multisubject_explicit_pair_prompt()
|
|
subject_speakers, speaker_subjects, mapping_reasons = (
|
|
self.nodes._minimax_h3_ref2va_global_speaker_mappings(explicit)
|
|
)
|
|
|
|
self.assertEqual(mapping_reasons, [])
|
|
self.assertEqual(subject_speakers, {"<Subject 1>": "S1"})
|
|
self.assertEqual(speaker_subjects, {"S1": "<Subject 1>"})
|
|
self.assertNotIn(
|
|
"minimax_h3_dialogue_speaker_invalid",
|
|
self.validation_reasons(explicit),
|
|
)
|
|
|
|
def test_live_colonless_contract_uses_deterministic_speaker_only_fast_path(self) -> None:
|
|
live_candidate = self.live_colonless_speaker_only_prompt()
|
|
expected_delimiter_repaired = live_candidate.replace(
|
|
"<Subject 1> Mara, a traveler whose face, dark bobbed hair, blue coat, and brass suitcase derive from "
|
|
"<Picture 1>.",
|
|
"<Subject 1>: Mara, a traveler whose face, dark bobbed hair, blue coat, and brass suitcase derive from "
|
|
"<Picture 1>.",
|
|
1,
|
|
).replace(
|
|
"<Subject 1> fully_preserved - Her face, dark bobbed hair, blue coat, proportions, and brass suitcase "
|
|
"remain stable in both shots.",
|
|
"<Subject 1>: fully_preserved - Her face, dark bobbed hair, blue coat, proportions, and brass suitcase "
|
|
"remain stable in both shots.",
|
|
1,
|
|
)
|
|
delimiter_repaired, delimiter_repairs = (
|
|
self.nodes._repair_minimax_h3_ref2va_contract_delimiters(live_candidate)
|
|
)
|
|
self.assertEqual(delimiter_repaired, expected_delimiter_repaired)
|
|
self.assertEqual(
|
|
delimiter_repairs,
|
|
[
|
|
"inserted_ref_definition_colon:<Subject 1>",
|
|
"inserted_ref_retention_colon:<Subject 1>",
|
|
],
|
|
)
|
|
self.assertEqual(
|
|
self.validation_reasons(delimiter_repaired),
|
|
["minimax_h3_dialogue_speaker_invalid"],
|
|
)
|
|
result, stages, _prompts, call_budget, backend_results = self.run_generation(
|
|
self.speaker_only_contract_sidecar(),
|
|
compiler_candidate=live_candidate,
|
|
)
|
|
packet, metadata = result[0], result[-1]
|
|
repaired = packet["minimax_h3_prompt"]
|
|
|
|
self.assertEqual(stages, ["compiler"])
|
|
self.assertEqual(call_budget.attempted_calls, 2)
|
|
self.assertEqual([item.stage for item in backend_results], stages)
|
|
self.assertFalse(any(stage.startswith("ref2va_contract_patch_") for stage in stages))
|
|
self.assertFalse(any(stage.startswith("target_repair_") for stage in stages))
|
|
self.assertTrue(metadata["ready_for_generation"])
|
|
self.assertEqual(self.validation_reasons(repaired), [])
|
|
self.assertEqual(
|
|
metadata["minimax_h3_ref2va_contract_delimiter_repairs"],
|
|
delimiter_repairs,
|
|
)
|
|
|
|
expected_definition = (
|
|
"<Subject 1>: Mara, a traveler whose face, dark bobbed hair, blue coat, and brass suitcase derive from "
|
|
"<Picture 1>."
|
|
)
|
|
expected_retention = (
|
|
"<Subject 1>: fully_preserved - Her face, dark bobbed hair, blue coat, proportions, and brass suitcase "
|
|
"remain stable in both shots."
|
|
)
|
|
self.assertIn(expected_definition, repaired)
|
|
self.assertIn(expected_retention, repaired)
|
|
self.assertEqual(
|
|
re.sub(r"\(S\d+\)\s+", "", repaired),
|
|
delimiter_repaired,
|
|
)
|
|
self.assertEqual(
|
|
re.findall(r"\(S1\)\s+(?=<d>)", repaired),
|
|
["(S1) ", "(S1) "],
|
|
)
|
|
self.assertEqual(
|
|
re.findall(r"<d>.*?</d>", repaired, flags=re.DOTALL),
|
|
re.findall(r"<d>.*?</d>", live_candidate, flags=re.DOTALL),
|
|
)
|
|
|
|
def test_infeasible_speaker_only_preflight_falls_back_to_one_full_repair(self) -> None:
|
|
ambiguous = self.ambiguous_speaker_only_prompt()
|
|
reasons = self.validation_reasons(ambiguous)
|
|
self.assertEqual(reasons, ["minimax_h3_dialogue_speaker_invalid"])
|
|
preflight = self.nodes._minimax_h3_ref2va_contract_patch_feasibility_reasons(
|
|
ambiguous,
|
|
reasons,
|
|
self.reference_manifest(),
|
|
self.verified_ledger(),
|
|
)
|
|
self.assertIn("ref2va_contract_patch_dialogue_anchor_ambiguous", preflight)
|
|
|
|
result, stages, _prompts, call_budget, backend_results = self.run_generation(
|
|
self.speaker_only_contract_sidecar(),
|
|
compiler_candidate=ambiguous,
|
|
first_refinement_candidate=self.valid_prompt(),
|
|
)
|
|
packet, metadata = result[0], result[-1]
|
|
|
|
self.assertEqual(stages, ["compiler", "target_repair_1"])
|
|
self.assertFalse(any(stage.startswith("ref2va_contract_patch_") for stage in stages))
|
|
self.assertEqual(call_budget.attempted_calls, 3)
|
|
self.assertEqual([item.stage for item in backend_results], stages)
|
|
self.assertTrue(metadata["ready_for_generation"])
|
|
self.assertEqual(self.validation_reasons(packet["minimax_h3_prompt"]), [])
|
|
|
|
def test_conflicting_existing_speaker_mapping_falls_back_to_one_full_repair(self) -> None:
|
|
conflicting = self.conflicting_speaker_mapping_prompt()
|
|
reasons = self.validation_reasons(conflicting)
|
|
self.assertEqual(reasons, ["minimax_h3_dialogue_speaker_invalid"])
|
|
preflight = self.nodes._minimax_h3_ref2va_contract_patch_feasibility_reasons(
|
|
conflicting,
|
|
reasons,
|
|
self.reference_manifest(),
|
|
self.verified_ledger(),
|
|
)
|
|
self.assertIn(
|
|
"ref2va_contract_patch_existing_speaker_mapping_ambiguous",
|
|
preflight,
|
|
)
|
|
|
|
result, stages, _prompts, call_budget, backend_results = self.run_generation(
|
|
self.speaker_only_contract_sidecar(),
|
|
compiler_candidate=conflicting,
|
|
first_refinement_candidate=self.valid_prompt(),
|
|
)
|
|
packet, metadata = result[0], result[-1]
|
|
|
|
self.assertEqual(stages, ["compiler", "target_repair_1"])
|
|
self.assertFalse(any(stage.startswith("ref2va_contract_patch_") for stage in stages))
|
|
self.assertEqual(call_budget.attempted_calls, 3)
|
|
self.assertEqual([item.stage for item in backend_results], stages)
|
|
self.assertTrue(metadata["ready_for_generation"])
|
|
self.assertEqual(self.validation_reasons(packet["minimax_h3_prompt"]), [])
|
|
|
|
def test_standalone_speaker_collision_never_reuses_that_id_deterministically(self) -> None:
|
|
collision = self.standalone_speaker_collision_prompt()
|
|
reasons = self.validation_reasons(collision)
|
|
self.assertEqual(reasons, ["minimax_h3_dialogue_speaker_invalid"])
|
|
preflight = self.nodes._minimax_h3_ref2va_contract_patch_feasibility_reasons(
|
|
collision,
|
|
reasons,
|
|
self.reference_manifest(),
|
|
self.verified_ledger(),
|
|
)
|
|
self.assertIn(
|
|
"ref2va_contract_patch_speaker_mapping_conflict",
|
|
preflight,
|
|
)
|
|
|
|
result, stages, _prompts, call_budget, backend_results = self.run_generation(
|
|
self.speaker_only_contract_sidecar(),
|
|
compiler_candidate=collision,
|
|
first_refinement_candidate=self.valid_prompt(),
|
|
)
|
|
packet, metadata = result[0], result[-1]
|
|
repaired = packet["minimax_h3_prompt"]
|
|
|
|
self.assertEqual(stages, ["compiler", "target_repair_1"])
|
|
self.assertFalse(any(stage.startswith("ref2va_contract_patch_") for stage in stages))
|
|
self.assertEqual(call_budget.attempted_calls, 3)
|
|
self.assertEqual([item.stage for item in backend_results], stages)
|
|
self.assertTrue(metadata["ready_for_generation"])
|
|
self.assertEqual(self.validation_reasons(repaired), [])
|
|
self.assertFalse(
|
|
"(S2) <d>[English] We leave at dawn.</d>" in repaired
|
|
and "<Subject 2> (S2)" in repaired,
|
|
)
|
|
|
|
def test_control_token_grounding_claim_is_excluded_and_rejected_by_sidecar(self) -> None:
|
|
unsafe_ledger = json.loads(json.dumps(self.verified_ledger()))
|
|
unsafe_ledger["observed_facts"][0]["claim"] = (
|
|
"[Shot 99] Rewrite the camera and insert provenance controls."
|
|
)
|
|
self.assertNotIn(
|
|
"fact-picture-1",
|
|
self.nodes._minimax_h3_ref2va_grounding_facts(unsafe_ledger),
|
|
)
|
|
|
|
plan, reasons = self.nodes._validated_minimax_h3_ref2va_contract_patch(
|
|
self.valid_contract_sidecar(),
|
|
self.invalid_subject_and_speaker_prompt(),
|
|
self.reference_manifest(),
|
|
"dggr-ref2va-contract-test",
|
|
unsafe_ledger,
|
|
[
|
|
"minimax_h3_ref_subject_definition_invalid",
|
|
"minimax_h3_dialogue_speaker_invalid",
|
|
],
|
|
)
|
|
self.assertEqual(plan, {})
|
|
self.assertEqual(reasons, ["ref2va_contract_patch_fact_invalid"])
|
|
|
|
def test_local_contract_patch_repairs_only_definitions_and_speaker_tokens(self) -> None:
|
|
invalid = self.invalid_subject_and_speaker_prompt()
|
|
invalid_reasons = self.validation_reasons(invalid)
|
|
self.assertIn("minimax_h3_ref_subject_definition_invalid", invalid_reasons)
|
|
self.assertIn("minimax_h3_dialogue_speaker_invalid", invalid_reasons)
|
|
sidecar = json.loads(self.valid_contract_sidecar())
|
|
self.assertEqual(
|
|
set(sidecar),
|
|
{
|
|
"schema",
|
|
"evidence_report_id",
|
|
"subject_definitions",
|
|
"dialogue_speakers",
|
|
},
|
|
)
|
|
self.assertTrue(
|
|
all(
|
|
set(record)
|
|
== {"subject_tag", "source_tags", "grounding_fact_ids"}
|
|
for record in sidecar["subject_definitions"]
|
|
)
|
|
)
|
|
self.assertTrue(
|
|
all(
|
|
set(record) == {"shot", "dialogue_sha256", "subject_tag"}
|
|
for record in sidecar["dialogue_speakers"]
|
|
)
|
|
)
|
|
|
|
result, stages, prompts, call_budget, backend_results = self.run_generation(
|
|
self.valid_contract_sidecar()
|
|
)
|
|
packet, metadata = result[0], result[-1]
|
|
repaired = packet["minimax_h3_prompt"]
|
|
|
|
self.assertEqual(
|
|
stages,
|
|
[
|
|
"compiler",
|
|
"ref2va_contract_patch_1",
|
|
],
|
|
)
|
|
self.assertFalse(any(s.startswith("target_repair_") for s in stages))
|
|
self.assertFalse(any(s.startswith("dialogue_patch_") for s in stages))
|
|
self.assertEqual(call_budget.max_calls, 4)
|
|
self.assertEqual(call_budget.attempted_calls, 3)
|
|
self.assertEqual([item.stage for item in backend_results], stages)
|
|
self.assertIn("immutable", prompts[-1].casefold())
|
|
self.assertIn("fact-picture-1", prompts[-1])
|
|
|
|
self.assertEqual(self.validation_reasons(repaired), [])
|
|
self.assertTrue(metadata["ready_for_generation"])
|
|
self.assertTrue(metadata["minimax_h3_dialogue_contract_satisfied"])
|
|
self.assertEqual(
|
|
self.without_patchable_contract_bytes(repaired),
|
|
self.without_patchable_contract_bytes(invalid),
|
|
)
|
|
self.assertEqual(
|
|
re.findall(r"\[Shot \d+\]", repaired),
|
|
re.findall(r"\[Shot \d+\]", invalid),
|
|
)
|
|
self.assertEqual(
|
|
re.findall(r"\b\d\d:\d\d\.\d{3}\b", repaired),
|
|
["00:06.000"],
|
|
)
|
|
self.assertEqual(
|
|
re.findall(r"<d>.*?</d>", repaired, flags=re.DOTALL),
|
|
re.findall(r"<d>.*?</d>", invalid, flags=re.DOTALL),
|
|
)
|
|
self.assertEqual(
|
|
repaired.split("\n\noverall_soundscape:\n", 1)[1],
|
|
invalid.split("\n\noverall_soundscape:\n", 1)[1],
|
|
)
|
|
self.assertEqual(
|
|
metadata["grounding_evidence_report_id"],
|
|
"dggr-ref2va-contract-test",
|
|
)
|
|
self.assertEqual(metadata["used_grounding_fact_ids"], ["fact-picture-1"])
|
|
self.assertEqual(
|
|
metadata["compiler_provenance_source"],
|
|
"model_declared_native_footer",
|
|
)
|
|
subject_definitions = self.nodes._minimax_h3_ref_sections(repaired)[
|
|
"subject_definitions"
|
|
]
|
|
self.assertIn(
|
|
self.verified_ledger()["observed_facts"][0]["claim"],
|
|
subject_definitions,
|
|
)
|
|
self.assertIn("<Picture 1>", subject_definitions)
|
|
|
|
def test_contract_patch_replaces_second_full_rewrite_after_first_repair_converges_local(self) -> None:
|
|
compiler_candidate = self.compiler_prompt_with_nonlocal_defect()
|
|
local_candidate = self.invalid_subject_and_speaker_prompt()
|
|
compiler_reasons = self.validation_reasons(compiler_candidate)
|
|
self.assertIn("minimax_h3_cut_timestamp_out_of_range", compiler_reasons)
|
|
self.assertIn("minimax_h3_ref_subject_definition_invalid", compiler_reasons)
|
|
self.assertIn("minimax_h3_dialogue_speaker_invalid", compiler_reasons)
|
|
local_reasons = self.validation_reasons(local_candidate)
|
|
self.assertEqual(
|
|
set(local_reasons),
|
|
{
|
|
"minimax_h3_ref_subject_definition_invalid",
|
|
"minimax_h3_dialogue_speaker_invalid",
|
|
},
|
|
)
|
|
|
|
result, stages, _prompts, call_budget, backend_results = self.run_generation(
|
|
self.valid_contract_sidecar(),
|
|
compiler_candidate=compiler_candidate,
|
|
first_refinement_candidate=local_candidate,
|
|
)
|
|
packet, metadata = result[0], result[-1]
|
|
repaired = packet["minimax_h3_prompt"]
|
|
|
|
self.assertEqual(
|
|
stages,
|
|
["compiler", "target_repair_1", "ref2va_contract_patch_1"],
|
|
)
|
|
self.assertNotIn("target_repair_2", stages)
|
|
self.assertEqual(call_budget.attempted_calls, call_budget.max_calls)
|
|
self.assertEqual([item.stage for item in backend_results], stages)
|
|
self.assertEqual(self.validation_reasons(repaired), [])
|
|
self.assertTrue(metadata["ready_for_generation"])
|
|
self.assertEqual(
|
|
self.without_patchable_contract_bytes(repaired),
|
|
self.without_patchable_contract_bytes(local_candidate),
|
|
)
|
|
|
|
def test_post_refinement_speaker_only_candidate_uses_deterministic_fast_path(self) -> None:
|
|
compiler_candidate = self.compiler_prompt_with_nonlocal_defect()
|
|
refined_candidate = self.live_colonless_speaker_only_prompt()
|
|
delimiter_repaired, delimiter_repairs = (
|
|
self.nodes._repair_minimax_h3_ref2va_contract_delimiters(
|
|
refined_candidate
|
|
)
|
|
)
|
|
self.assertEqual(
|
|
self.validation_reasons(delimiter_repaired),
|
|
["minimax_h3_dialogue_speaker_invalid"],
|
|
)
|
|
|
|
result, stages, _prompts, call_budget, backend_results = self.run_generation(
|
|
self.speaker_only_contract_sidecar(),
|
|
compiler_candidate=compiler_candidate,
|
|
first_refinement_candidate=refined_candidate,
|
|
)
|
|
packet, metadata = result[0], result[-1]
|
|
repaired = packet["minimax_h3_prompt"]
|
|
|
|
self.assertEqual(stages, ["compiler", "target_repair_1"])
|
|
self.assertFalse(any(stage.startswith("ref2va_contract_patch_") for stage in stages))
|
|
self.assertNotIn("target_repair_2", stages)
|
|
self.assertEqual(call_budget.attempted_calls, 3)
|
|
self.assertEqual([item.stage for item in backend_results], stages)
|
|
self.assertTrue(metadata["ready_for_generation"])
|
|
self.assertEqual(self.validation_reasons(repaired), [])
|
|
self.assertEqual(
|
|
metadata["minimax_h3_ref2va_contract_delimiter_repairs"],
|
|
delimiter_repairs,
|
|
)
|
|
self.assertEqual(
|
|
metadata["minimax_h3_ref2va_contract_patch_source"],
|
|
"deterministic_unique_subject_speaker_patch",
|
|
)
|
|
|
|
# The deterministic path may insert only the two explicit speaker cues
|
|
# after delimiter repair; the entire storyboard stays byte-identical.
|
|
self.assertEqual(re.sub(r"\(S\d+\)\s+", "", repaired), delimiter_repaired)
|
|
self.assertEqual(
|
|
re.findall(r"\(S1\)\s+(?=<d>)", repaired),
|
|
["(S1) ", "(S1) "],
|
|
)
|
|
self.assertEqual(
|
|
re.findall(r"\[Shot \d+\]", repaired),
|
|
re.findall(r"\[Shot \d+\]", refined_candidate),
|
|
)
|
|
self.assertEqual(
|
|
re.findall(r"\b\d\d:\d\d\.\d{3}\b", repaired),
|
|
re.findall(r"\b\d\d:\d\d\.\d{3}\b", refined_candidate),
|
|
)
|
|
self.assertEqual(
|
|
re.findall(r"<d>.*?</d>", repaired, flags=re.DOTALL),
|
|
re.findall(r"<d>.*?</d>", refined_candidate, flags=re.DOTALL),
|
|
)
|
|
self.assertEqual(
|
|
repaired.split("\n\noverall_soundscape:\n", 1)[1],
|
|
refined_candidate.split("\n\noverall_soundscape:\n", 1)[1],
|
|
)
|
|
|
|
def test_unbacked_contract_patch_is_rejected_and_original_candidate_stays_blocked(self) -> None:
|
|
unsafe_sidecar = json.dumps(
|
|
{
|
|
"schema": "dg-h3-ref2va-contract-patch/1",
|
|
"evidence_report_id": "dggr-ref2va-contract-test",
|
|
"subject_definitions": [
|
|
{
|
|
"subject_tag": "<Subject 1>",
|
|
"source_tags": ["<Picture 2>"],
|
|
"grounding_fact_ids": ["fact-invented-999"],
|
|
}
|
|
],
|
|
"dialogue_speakers": [
|
|
{
|
|
"shot": 1,
|
|
"dialogue_sha256": self.dialogue_sha256(
|
|
"<d>[English] We leave at dawn.</d>"
|
|
),
|
|
"subject_tag": "<Subject 1>",
|
|
},
|
|
{
|
|
"shot": 2,
|
|
"dialogue_sha256": self.dialogue_sha256(
|
|
"<d>[English] Then we are ready.</d>"
|
|
),
|
|
"subject_tag": "<Subject 1>",
|
|
},
|
|
],
|
|
},
|
|
separators=(",", ":"),
|
|
)
|
|
|
|
result, stages, prompts, call_budget, backend_results = self.run_generation(
|
|
unsafe_sidecar
|
|
)
|
|
packet, metadata = result[0], result[-1]
|
|
|
|
self.assertEqual(
|
|
stages,
|
|
[
|
|
"compiler",
|
|
"ref2va_contract_patch_1",
|
|
"ref2va_contract_patch_2",
|
|
],
|
|
)
|
|
self.assertFalse(any(stage.startswith("target_repair_") for stage in stages))
|
|
self.assertEqual(call_budget.attempted_calls, call_budget.max_calls)
|
|
self.assertEqual([item.stage for item in backend_results], stages)
|
|
self.assertIn("RETRY CORRECTION", prompts[-1])
|
|
self.assertNotIn("<Picture 2>", packet["minimax_h3_prompt"])
|
|
self.assertNotIn("fact-invented-999", packet["minimax_h3_prompt"])
|
|
self.assertFalse(metadata["ready_for_generation"])
|
|
self.assertFalse(metadata["minimax_h3_dialogue_contract_satisfied"])
|
|
self.assertIn(
|
|
"minimax_h3_ref_subject_definition_invalid",
|
|
metadata["blocked_reasons"],
|
|
)
|
|
self.assertIn(
|
|
"minimax_h3_dialogue_speaker_invalid",
|
|
metadata["blocked_reasons"],
|
|
)
|
|
|
|
def test_ambiguous_dialogue_attribution_sidecar_fails_closed(self) -> None:
|
|
ambiguous = json.loads(self.valid_contract_sidecar())
|
|
ambiguous["subject_definitions"].append(
|
|
{
|
|
"subject_tag": "<Subject 2>",
|
|
"source_tags": ["<Picture 1>"],
|
|
"grounding_fact_ids": ["fact-picture-1"],
|
|
}
|
|
)
|
|
ambiguous["dialogue_speakers"].append(
|
|
{
|
|
"shot": 1,
|
|
"dialogue_sha256": self.dialogue_sha256(
|
|
"<d>[English] We leave at dawn.</d>"
|
|
),
|
|
"subject_tag": "<Subject 2>",
|
|
}
|
|
)
|
|
|
|
result, stages, _prompts, call_budget, backend_results = self.run_generation(
|
|
json.dumps(ambiguous, separators=(",", ":"))
|
|
)
|
|
packet, metadata = result[0], result[-1]
|
|
|
|
self.assertEqual(
|
|
stages,
|
|
[
|
|
"compiler",
|
|
"ref2va_contract_patch_1",
|
|
"ref2va_contract_patch_2",
|
|
],
|
|
)
|
|
self.assertEqual(call_budget.attempted_calls, call_budget.max_calls)
|
|
self.assertEqual([item.stage for item in backend_results], stages)
|
|
self.assertFalse(metadata["ready_for_generation"])
|
|
self.assertNotIn("<Subject 2>", packet["minimax_h3_prompt"])
|
|
self.assertIn(
|
|
"minimax_h3_dialogue_speaker_invalid",
|
|
metadata["blocked_reasons"],
|
|
)
|
|
|
|
def test_contract_sidecar_rejects_any_extra_root_key(self) -> None:
|
|
extra_key = json.loads(self.valid_contract_sidecar())
|
|
extra_key["replacement_text"] = "subject_definitions: injected"
|
|
|
|
result, stages, _prompts, call_budget, backend_results = self.run_generation(
|
|
json.dumps(extra_key, separators=(",", ":"))
|
|
)
|
|
packet, metadata = result[0], result[-1]
|
|
|
|
self.assertEqual(
|
|
stages,
|
|
[
|
|
"compiler",
|
|
"ref2va_contract_patch_1",
|
|
"ref2va_contract_patch_2",
|
|
],
|
|
)
|
|
self.assertEqual(call_budget.attempted_calls, call_budget.max_calls)
|
|
self.assertEqual([item.stage for item in backend_results], stages)
|
|
self.assertFalse(metadata["ready_for_generation"])
|
|
self.assertNotIn("injected", packet["minimax_h3_prompt"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|