danrh
8f23c95546
Add measured-audio, timed-lyrics, H3 lane planning, identity relay, upload-song routing, model-specific camera contracts, LTX long-horizon planning, workflow migrations, the validated V6 example, regression coverage, and an Unreleased changelog.
2273 lines
94 KiB
Python
2273 lines
94 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
import importlib.util
|
|
import json
|
|
import sys
|
|
import unittest
|
|
import uuid
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
import torch
|
|
|
|
|
|
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_grounding_integration_{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 GroundingGuardIntegrationTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.nodes = load_nodes_module()
|
|
|
|
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": True},
|
|
)
|
|
|
|
def target(self):
|
|
return self.nodes.TargetProfileConfig(target_profile="ltx")
|
|
|
|
def image_context(self):
|
|
return self.nodes.GemmaContext(
|
|
user_prompt="Describe the reference faithfully.",
|
|
images=torch.zeros((1, 8, 8, 3), dtype=torch.float32),
|
|
source="image",
|
|
media_metadata={
|
|
"source": "image",
|
|
"reference_image_count": 1,
|
|
"image_role": "primary_image_source",
|
|
},
|
|
)
|
|
|
|
def video_context(self, frame_count: int = 5):
|
|
return self.nodes.GemmaContext(
|
|
user_prompt="Describe the video faithfully.",
|
|
images=torch.zeros((frame_count, 8, 8, 3), dtype=torch.float32),
|
|
source="video",
|
|
media_metadata={
|
|
"source": "video",
|
|
"sampled_indices": [index * 10 for index in range(frame_count)],
|
|
"sampled_frame_count": frame_count,
|
|
"video_sampled_frame_count": frame_count,
|
|
"source_fps": 10.0,
|
|
"sample_fps": 1.0,
|
|
"video_role": "primary_video_recreation_source",
|
|
},
|
|
)
|
|
|
|
@staticmethod
|
|
def uncertain_ledger(status: str = "uncertain") -> dict:
|
|
return {
|
|
"schema": "dg-grounding-ledger/1",
|
|
"analysis_status": status,
|
|
"observed_facts": [],
|
|
"inferred_facts": [],
|
|
"creative_additions": [],
|
|
"uncertainties": ["Visual evidence could not be verified."],
|
|
"grounding_failure_reasons": ["insufficient visual certainty"],
|
|
}
|
|
|
|
@staticmethod
|
|
def grounded_ledger(registry: dict) -> dict:
|
|
facts: list[dict] = []
|
|
fact_index = 1
|
|
for asset in registry.get("assets", []):
|
|
samples = asset.get("samples", [])
|
|
chosen_samples = samples
|
|
if asset.get("kind") == "video" and len(samples) > 1:
|
|
chosen_samples = [samples[0], samples[-1]]
|
|
for sample in chosen_samples:
|
|
evidence = {
|
|
"asset_id": asset["asset_id"],
|
|
"sample_ordinal": sample["sample_ordinal"],
|
|
"source_frame_index": sample["source_frame_index"],
|
|
"timecode_seconds": sample["timecode_seconds"],
|
|
}
|
|
category = "action" if asset.get("kind") == "video" else "object"
|
|
facts.append(
|
|
{
|
|
"fact_id": f"fact-{fact_index}",
|
|
"claim": f"A directly visible fact from {asset['asset_id']}.",
|
|
"confidence": "high",
|
|
"categories": [category],
|
|
"evidence": [evidence],
|
|
"typed_claims": [],
|
|
}
|
|
)
|
|
fact_index += 1
|
|
return {
|
|
"schema": "dg-grounding-ledger/1",
|
|
"analysis_status": "grounded",
|
|
"observed_facts": facts,
|
|
"inferred_facts": [],
|
|
"creative_additions": [],
|
|
"uncertainties": [],
|
|
"grounding_failure_reasons": [],
|
|
}
|
|
|
|
@staticmethod
|
|
def packet(prompt: str = "A verified generator prompt.", ledger: dict | None = None) -> dict:
|
|
packet = {
|
|
"ltx_prompt": prompt,
|
|
"ideogram_prompt": "",
|
|
"minimax_h3_prompt": "",
|
|
"negative_prompt": "",
|
|
"scene_segments": [],
|
|
"metadata": {
|
|
"target_profile": "ltx",
|
|
"ready_for_generation": True,
|
|
"blocked_reasons": [],
|
|
},
|
|
}
|
|
if ledger is not None:
|
|
packet["grounding_ledger"] = copy.deepcopy(ledger)
|
|
return packet
|
|
|
|
def backend_result(
|
|
self,
|
|
ledger: dict,
|
|
*,
|
|
transport_confirmed: bool = True,
|
|
telemetry_refusal: bool = False,
|
|
):
|
|
return self.nodes.BackendRunResult(
|
|
decoded_text=json.dumps(ledger),
|
|
transport={
|
|
"pixel_transport_confirmed": transport_confirmed,
|
|
"not_applicable": False,
|
|
"mismatch_reasons": [],
|
|
},
|
|
telemetry={
|
|
"schema_version": "dg-denoising-telemetry/1",
|
|
"forward_count": 2,
|
|
"snapshots": [{"canvas": 1, "step": 1}],
|
|
"stream_ended": True,
|
|
"final_output": {"text": [], "token_count": [1]},
|
|
"errors": [],
|
|
"refusal": {"detected": telemetry_refusal},
|
|
},
|
|
effective_sampling={"profile": "checkpoint_defaults"},
|
|
timing={"total_seconds": 0.01},
|
|
)
|
|
|
|
def truncated_evidence_result(self, registry: dict, complete_fact_count: int = 5):
|
|
ledger = self.grounded_ledger(registry)
|
|
template = ledger["observed_facts"][0]
|
|
facts: list[dict] = []
|
|
for fact_index in range(1, complete_fact_count + 2):
|
|
fact = copy.deepcopy(template)
|
|
fact["fact_id"] = f"fact-{fact_index}"
|
|
fact["claim"] = f"Visible detail {fact_index} deliberately described."
|
|
facts.append(fact)
|
|
ledger["observed_facts"] = facts
|
|
decoded_text = json.dumps(ledger)
|
|
partial_claim = f"Visible detail {complete_fact_count + 1} deliberately"
|
|
decoded_text = decoded_text[: decoded_text.index(partial_claim) + len(partial_claim)]
|
|
result = self.backend_result(ledger)
|
|
result.decoded_text = decoded_text
|
|
return result
|
|
|
|
@staticmethod
|
|
def legacy_result(packet: dict):
|
|
return packet, "compiler raw", "", "", True, "", packet["metadata"]
|
|
|
|
def test_settings_and_cot_public_contract_preserve_existing_outputs(self) -> None:
|
|
settings_inputs = self.nodes.DiffusionGemmaGroundingGuardSettings.INPUT_TYPES()
|
|
required = settings_inputs["required"]
|
|
optional = settings_inputs["optional"]
|
|
self.assertEqual(required["mode"][0], ["off", "audit", "strict"])
|
|
self.assertEqual(required["mode"][1]["default"], "audit")
|
|
self.assertTrue(required["retry_on_uncertain"][1]["default"])
|
|
self.assertEqual(required["sampling_profile"][1]["default"], "checkpoint_defaults")
|
|
self.assertEqual(required["seed"][1]["max"], (1 << 64) - 1)
|
|
self.assertTrue(optional["external_evidence_json"][1]["forceInput"])
|
|
self.assertEqual(
|
|
optional["evidence_token_budget"][0],
|
|
["auto", "768", "1024", "1280"],
|
|
)
|
|
self.assertEqual(optional["evidence_token_budget"][1]["default"], "auto")
|
|
self.assertTrue(optional["evidence_token_budget"][1]["advanced"])
|
|
self.assertEqual(
|
|
self.nodes.DiffusionGemmaGroundingGuardSettings.RETURN_TYPES,
|
|
("DG_GROUNDING_GUARD_CONFIG", "STRING"),
|
|
)
|
|
|
|
cot_inputs = self.nodes.DiffusionGemmaCoTGenerator.INPUT_TYPES()
|
|
self.assertIn("grounding_guard_config", cot_inputs["optional"])
|
|
self.assertEqual(
|
|
self.nodes.DiffusionGemmaCoTGenerator.RETURN_NAMES[:4],
|
|
("final_json", "reasoning_text", "raw_response", "metadata_json"),
|
|
)
|
|
self.assertEqual(
|
|
self.nodes.DiffusionGemmaCoTGenerator.RETURN_NAMES[4:],
|
|
("grounding_status", "grounding_report_json"),
|
|
)
|
|
|
|
config, config_json = self.nodes.DiffusionGemmaGroundingGuardSettings().build()
|
|
self.assertEqual(config.mode, "audit")
|
|
self.assertEqual(config.evidence_token_budget, "auto")
|
|
self.assertEqual(json.loads(config_json)["sampling_settings"]["max_denoising_steps"], 48)
|
|
self.assertEqual(json.loads(config_json)["evidence_token_budget"], "auto")
|
|
|
|
legacy_config, _ = self.nodes.DiffusionGemmaGroundingGuardSettings().build(
|
|
"strict",
|
|
False,
|
|
"checkpoint_defaults",
|
|
7,
|
|
False,
|
|
"legacy_trace",
|
|
'{"schema":"legacy-external"}',
|
|
)
|
|
self.assertEqual(legacy_config.evidence_token_budget, "auto")
|
|
self.assertEqual(
|
|
json.loads(legacy_config.external_evidence_json)["schema"],
|
|
"legacy-external",
|
|
)
|
|
|
|
selected_config, selected_json = self.nodes.DiffusionGemmaGroundingGuardSettings().build(
|
|
evidence_token_budget="1280"
|
|
)
|
|
self.assertEqual(selected_config.evidence_token_budget, "1280")
|
|
self.assertEqual(json.loads(selected_json)["evidence_token_budget"], "1280")
|
|
|
|
def test_strict_guard_block_survives_splitter_and_generation_gate(self) -> None:
|
|
guard_reasons = [
|
|
"visual_grounding_unverified",
|
|
"visual_grounding_insufficient_coverage",
|
|
]
|
|
guard_report = {
|
|
"schema": "dg-grounding-report/1",
|
|
"analysis_status": "uncertain",
|
|
"decision": "block",
|
|
"would_block": True,
|
|
"grounding_guard_would_block": True,
|
|
"blocked_reasons": guard_reasons,
|
|
"retry_reasons": ["role coverage remained incomplete after retry"],
|
|
"validation": {
|
|
"errors": ["ledger:role_specific_category_missing:picture:2"]
|
|
},
|
|
}
|
|
context = self.image_context()
|
|
splitter = self.nodes.DiffusionGemmaJSONSplitter()
|
|
gate = self.nodes.DiffusionGemmaGenerationGate()
|
|
prompt_index = {"ltx": 0, "ideogram4": 1, "minimax_h3": 13}
|
|
|
|
for profile in prompt_index:
|
|
with self.subTest(profile=profile):
|
|
target = self.nodes.TargetProfileConfig(
|
|
target_profile=profile,
|
|
minimax_h3_mode="t2va",
|
|
)
|
|
blocked_packet = self.nodes._blocked_grounding_packet(
|
|
target,
|
|
copy.deepcopy(guard_report),
|
|
)
|
|
|
|
split = splitter.split(json.dumps(blocked_packet), context, target)
|
|
metadata = json.loads(split[4])
|
|
|
|
self.assertFalse(split[12])
|
|
self.assertEqual(split[0], "")
|
|
self.assertEqual(split[1], "")
|
|
self.assertEqual(split[13], "")
|
|
self.assertEqual(metadata["blocked_reasons"], guard_reasons)
|
|
self.assertFalse(metadata["claim_verification_passed"])
|
|
|
|
with self.assertRaises(ValueError) as raised:
|
|
gate.gate(split[prompt_index[profile]], split[12], split[4])
|
|
message = str(raised.exception)
|
|
self.assertIn("Grounding Guard blocked generation", message)
|
|
self.assertIn("visual_grounding_insufficient_coverage", message)
|
|
self.assertIn(
|
|
"ledger:role_specific_category_missing:picture:2", message
|
|
)
|
|
self.assertNotIn("minimax_h3_ref_missing", message)
|
|
self.assertNotIn("empty_generation_prompt", message)
|
|
|
|
stale_metadata = {
|
|
"blocked_reasons": ["empty_generation_prompt"],
|
|
"grounding_guard": guard_report,
|
|
}
|
|
with self.assertRaisesRegex(
|
|
ValueError, "visual_grounding_insufficient_coverage"
|
|
):
|
|
gate.gate(
|
|
"A stale splitter incorrectly supplied this fallback prompt.",
|
|
True,
|
|
json.dumps(stale_metadata),
|
|
)
|
|
|
|
for broken_decision in (None, "blok"):
|
|
with self.subTest(broken_decision=broken_decision):
|
|
malformed_report = copy.deepcopy(guard_report)
|
|
malformed_report["config"] = {"mode": "strict"}
|
|
if broken_decision is None:
|
|
malformed_report.pop("decision")
|
|
else:
|
|
malformed_report["decision"] = broken_decision
|
|
target = self.nodes.TargetProfileConfig(target_profile="ltx")
|
|
blocked_packet = self.nodes._blocked_grounding_packet(
|
|
target,
|
|
malformed_report,
|
|
)
|
|
split = splitter.split(json.dumps(blocked_packet), context, target)
|
|
self.assertFalse(split[12])
|
|
self.assertEqual(split[0], "")
|
|
self.assertEqual(
|
|
json.loads(split[4])["blocked_reasons"], guard_reasons
|
|
)
|
|
with self.assertRaisesRegex(
|
|
ValueError, "visual_grounding_insufficient_coverage"
|
|
):
|
|
gate.gate(split[0], split[12], split[4])
|
|
|
|
audit_metadata = {
|
|
"grounding_guard": {
|
|
**guard_report,
|
|
"decision": "warn",
|
|
"config": {"mode": "audit"},
|
|
}
|
|
}
|
|
self.assertEqual(
|
|
gate.gate("Audit mode remains non-blocking.", True, json.dumps(audit_metadata)),
|
|
("Audit mode remains non-blocking.",),
|
|
)
|
|
|
|
def test_branch_generation_gate_blocks_only_its_prompt_output(self) -> None:
|
|
gate = self.nodes.DiffusionGemmaBranchGenerationGate()
|
|
metadata = {
|
|
"blocked_reasons": [
|
|
"json_parse_invalid",
|
|
"minimax_h3_ref_missing_summary",
|
|
]
|
|
}
|
|
|
|
blocked_prompt, status, ready = gate.gate(
|
|
"An invalid H3 prompt that must not reach the sampler.",
|
|
False,
|
|
json.dumps(metadata),
|
|
)
|
|
|
|
self.assertIsInstance(blocked_prompt, self.nodes.ExecutionBlocker)
|
|
self.assertIsNone(blocked_prompt.message)
|
|
self.assertFalse(ready)
|
|
self.assertIn("did not pass validation", status)
|
|
self.assertIn("json_parse_invalid", status)
|
|
self.assertIn("minimax_h3_ref_missing_summary", status)
|
|
self.assertIn("produced no media output", status)
|
|
self.assertIn("select refresh once", status)
|
|
|
|
valid_prompt = "A validated prompt for an independent comparison branch."
|
|
self.assertEqual(
|
|
gate.gate(valid_prompt, True, json.dumps({"blocked_reasons": []})),
|
|
(
|
|
valid_prompt,
|
|
"DiffusionGemma branch is ready for generation.",
|
|
True,
|
|
),
|
|
)
|
|
|
|
def test_declared_h3_asset_without_tensor_blocks_strict_before_generation(self) -> None:
|
|
runtime = self.runtime()
|
|
context = self.nodes.GemmaContext(
|
|
user_prompt="Use the declared character reference.",
|
|
images=None,
|
|
source="image",
|
|
media_metadata={
|
|
"source": "image",
|
|
"minimax_h3_mode": "ref2va",
|
|
"minimax_h3_reference_image_batch_count": 1,
|
|
"minimax_h3_reference_manifest": (
|
|
"<Picture 1>: [dg:identity,appearance] protagonist identity"
|
|
),
|
|
},
|
|
)
|
|
target = self.nodes.TargetProfileConfig(
|
|
target_profile="minimax_h3",
|
|
minimax_h3_mode="ref2va",
|
|
)
|
|
with patch.object(
|
|
self.nodes, "_run_generation_packet_legacy"
|
|
) as legacy, patch.object(self.nodes, "_maybe_unload") as unload:
|
|
result = self.nodes._run_generation_packet(
|
|
runtime,
|
|
context,
|
|
target,
|
|
grounding_guard_config={"mode": "strict"},
|
|
)
|
|
legacy.assert_not_called()
|
|
unload.assert_called_once_with(runtime)
|
|
self.assertEqual(result[0]["minimax_h3_prompt"], "")
|
|
self.assertFalse(result[-1]["ready_for_generation"])
|
|
self.assertEqual(
|
|
result[-1]["blocked_reasons"][:2],
|
|
["visual_grounding_unverified", "visual_transport_error"],
|
|
)
|
|
self.assertEqual(result[-1]["grounding_guard"]["model_call_count"], 0)
|
|
self.assertIn(
|
|
"picture:1",
|
|
result[-1]["grounding_guard"]["asset_registry"][
|
|
"unattached_declared_visual_asset_ids"
|
|
],
|
|
)
|
|
|
|
def test_declared_h3_asset_without_tensor_only_warns_in_audit(self) -> None:
|
|
runtime = self.runtime()
|
|
context = self.nodes.GemmaContext(
|
|
user_prompt="Use the declared character reference.",
|
|
images=None,
|
|
source="image",
|
|
media_metadata={
|
|
"source": "image",
|
|
"minimax_h3_mode": "ref2va",
|
|
"minimax_h3_reference_image_batch_count": 1,
|
|
"minimax_h3_reference_manifest": (
|
|
"<Picture 1>: [dg:identity,appearance] protagonist identity"
|
|
),
|
|
},
|
|
)
|
|
target = self.nodes.TargetProfileConfig(
|
|
target_profile="minimax_h3",
|
|
minimax_h3_mode="ref2va",
|
|
)
|
|
packet = self.packet("")
|
|
packet["ltx_prompt"] = ""
|
|
packet["minimax_h3_prompt"] = "legacy H3 prompt retained in audit"
|
|
packet["metadata"]["target_profile"] = "minimax_h3"
|
|
with patch.object(
|
|
self.nodes,
|
|
"_run_generation_packet_legacy",
|
|
return_value=self.legacy_result(packet),
|
|
) as legacy, patch.object(self.nodes, "_maybe_unload"):
|
|
result = self.nodes._run_generation_packet(
|
|
runtime,
|
|
context,
|
|
target,
|
|
grounding_guard_config={"mode": "audit"},
|
|
)
|
|
legacy.assert_called_once()
|
|
self.assertEqual(
|
|
result[0]["minimax_h3_prompt"],
|
|
"legacy H3 prompt retained in audit",
|
|
)
|
|
self.assertTrue(result[-1]["ready_for_generation"])
|
|
self.assertEqual(result[-1]["grounding_guard"]["decision"], "warn")
|
|
self.assertTrue(
|
|
result[-1]["grounding_guard"]["grounding_guard_would_block"]
|
|
)
|
|
self.assertEqual(
|
|
result[-1]["grounding_guard"]["blocked_reasons"][:2],
|
|
["visual_grounding_unverified", "visual_transport_error"],
|
|
)
|
|
|
|
def test_strict_evidence_rejects_prose_prefix_but_accepts_native_thinking(self) -> None:
|
|
ledger = self.grounded_ledger(
|
|
self.nodes._grounding_asset_registry(self.image_context())
|
|
)
|
|
exact = json.dumps(ledger)
|
|
self.assertEqual(
|
|
self.nodes._strict_evidence_json_payload(f"<think>check citations</think>\n{exact}"),
|
|
exact,
|
|
)
|
|
with self.assertRaisesRegex(
|
|
self.nodes.GroundingLedgerError,
|
|
"without a conversational prefix",
|
|
):
|
|
self.nodes._strict_evidence_json_payload(
|
|
f"<|thought|>check citations<|end_thought|><|final_json|>{exact}<|end_final_json|>"
|
|
)
|
|
with self.assertRaisesRegex(
|
|
self.nodes.GroundingLedgerError,
|
|
"without a conversational prefix",
|
|
):
|
|
self.nodes._strict_evidence_json_payload(f"Here is the evidence:\n{exact}")
|
|
with self.assertRaisesRegex(
|
|
self.nodes.GroundingLedgerError,
|
|
"root must be the ledger",
|
|
):
|
|
self.nodes._strict_evidence_json_payload(
|
|
json.dumps({"grounding_ledger": ledger, "ltx_prompt": "must not pass"})
|
|
)
|
|
|
|
def test_strict_evidence_example_uses_the_authoritative_h3_asset_id(self) -> None:
|
|
registry = self.nodes.build_asset_registry(
|
|
{
|
|
"source": "image",
|
|
"minimax_h3_mode": "ref2va",
|
|
"minimax_h3_reference_image_batch_count": 1,
|
|
"minimax_h3_reference_manifest": (
|
|
"<Picture 1>: [dg:identity,appearance] protagonist identity"
|
|
),
|
|
},
|
|
1,
|
|
)
|
|
prompt = self.nodes._strict_evidence_prompt(registry)
|
|
self.assertIn('"asset_id": "picture:1"', prompt)
|
|
self.assertNotIn('"asset_id": "image:1"', prompt)
|
|
self.assertIn("Use no more than 4 observed facts total", prompt)
|
|
self.assertIn("Completing and closing the strict JSON object takes priority", prompt)
|
|
retry_prompt = self.nodes._strict_evidence_prompt(registry, retry=True)
|
|
self.assertIn("previous ledger did not pass validation", retry_prompt)
|
|
self.assertIn("Keep every JSON object and array closed", retry_prompt)
|
|
self.assertEqual(self.nodes._strict_evidence_max_new_tokens(2048), 768)
|
|
self.assertEqual(self.nodes._strict_evidence_max_new_tokens(2048, retry=True), 1024)
|
|
for budget, expected in (("768", 768), ("1024", 1024), ("1280", 1280)):
|
|
with self.subTest(evidence_token_budget=budget):
|
|
self.assertEqual(
|
|
self.nodes._strict_evidence_max_new_tokens(
|
|
512,
|
|
evidence_token_budget=budget,
|
|
),
|
|
expected,
|
|
)
|
|
self.assertEqual(
|
|
self.nodes._strict_evidence_max_new_tokens(
|
|
512,
|
|
retry=True,
|
|
evidence_token_budget=budget,
|
|
),
|
|
expected,
|
|
)
|
|
|
|
def test_h3_grounding_annotations_are_host_only(self) -> None:
|
|
manifest = (
|
|
"<Picture 1>: [dg:identity,appearance] protagonist identity\n"
|
|
"<Video 1>: [dg:action,motion,camera,temporal] movement reference"
|
|
)
|
|
context = self.nodes.GemmaContext(
|
|
user_prompt="Use the references faithfully.",
|
|
images=torch.zeros((3, 8, 8, 3), dtype=torch.float32),
|
|
source="image+video",
|
|
media_metadata={
|
|
"source": "image+video",
|
|
"minimax_h3_mode": "ref2va",
|
|
"minimax_h3_reference_image_batch_count": 1,
|
|
"minimax_h3_reference_manifest": manifest,
|
|
"sampled_indices": [0, 10],
|
|
"source_fps": 10.0,
|
|
},
|
|
)
|
|
registry = self.nodes._grounding_asset_registry(context)
|
|
compiler_metadata = self.nodes._strict_compiler_metadata(
|
|
context,
|
|
self.grounded_ledger(registry),
|
|
"dggr-test",
|
|
{"pixel_transport_confirmed": True},
|
|
registry,
|
|
)
|
|
self.assertNotIn("[dg:", json.dumps(compiler_metadata).lower())
|
|
|
|
model_prompt = self.nodes._build_model_prompt(
|
|
context.user_prompt,
|
|
self.nodes.DEFAULT_MASTER_PROMPT,
|
|
"minimax_h3",
|
|
context.media_metadata,
|
|
minimax_h3_mode="ref2va",
|
|
minimax_h3_reference_manifest=manifest,
|
|
)
|
|
self.assertNotIn("[dg:", model_prompt.lower())
|
|
self.assertIn("<Picture 1>: protagonist identity", model_prompt)
|
|
self.assertIn(
|
|
"minimax_h3_prompt_contains_grounding_role_annotation",
|
|
self.nodes._minimax_h3_ref2va_validation_reasons(
|
|
"subject_definitions: [dg:identity] leaked annotation",
|
|
5.0,
|
|
reference_manifest=manifest,
|
|
),
|
|
)
|
|
refinement_prompt = self.nodes._build_minimax_h3_refinement_prompt(
|
|
context.user_prompt,
|
|
"candidate prompt",
|
|
5.0,
|
|
"auto_scene_audio",
|
|
["minimax_h3_ref_missing_asset_tag"],
|
|
"ref2va",
|
|
manifest,
|
|
)
|
|
self.assertNotIn("[dg:", refinement_prompt.lower())
|
|
self.assertIn("<Picture 1>: protagonist identity", refinement_prompt)
|
|
|
|
def test_strict_prose_prefixed_ledger_blocks_without_compiler(self) -> None:
|
|
runtime = self.runtime()
|
|
ledger = self.grounded_ledger(
|
|
self.nodes._grounding_asset_registry(self.image_context())
|
|
)
|
|
result_with_prefix = self.backend_result(ledger)
|
|
result_with_prefix.decoded_text = "Here is the evidence:\n" + json.dumps(ledger)
|
|
with patch.object(
|
|
self.nodes,
|
|
"_run_backend_detailed",
|
|
return_value=result_with_prefix,
|
|
), patch.object(
|
|
self.nodes, "_run_generation_packet_legacy"
|
|
) as compiler, patch.object(self.nodes, "_maybe_unload"):
|
|
result = self.nodes._run_generation_packet(
|
|
runtime,
|
|
self.image_context(),
|
|
self.target(),
|
|
grounding_guard_config={"mode": "strict", "retry_on_uncertain": False},
|
|
)
|
|
compiler.assert_not_called()
|
|
self.assertEqual(result[0]["ltx_prompt"], "")
|
|
self.assertEqual(
|
|
result[-1]["blocked_reasons"][:2],
|
|
["visual_grounding_unverified", "visual_grounding_schema_invalid"],
|
|
)
|
|
|
|
def test_unattached_identity_still_never_consumes_the_first_video_frame(self) -> None:
|
|
frame_markers = [object() for _ in range(6)]
|
|
context = self.nodes.GemmaContext(
|
|
user_prompt="Use the still identity and video action.",
|
|
images=torch.zeros((6, 8, 8, 3), dtype=torch.float32),
|
|
source="video",
|
|
media_metadata={
|
|
"source": "video",
|
|
"media_synthesis_mode": "image_identity_video_control",
|
|
"reference_image_count": 1,
|
|
"reference_image_backend_attached": False,
|
|
"sampled_indices": [0, 10, 20, 30, 40, 50],
|
|
"video_sampled_frame_count": 6,
|
|
"source_fps": 10.0,
|
|
},
|
|
)
|
|
|
|
with patch.object(self.nodes, "_image_batch_to_pil", return_value=frame_markers):
|
|
messages, _kwargs = self.nodes._transformers_messages_and_processor_kwargs(
|
|
"Describe only verified evidence.",
|
|
self.nodes.MediaContext(
|
|
images=context.images,
|
|
source=context.source,
|
|
metadata=context.media_metadata,
|
|
),
|
|
)
|
|
content = messages[0]["content"]
|
|
rendered_text = "\n".join(
|
|
item["text"] for item in content if item.get("type") == "text"
|
|
)
|
|
transported_images = [item["image"] for item in content if item.get("type") == "image"]
|
|
self.assertIn("IDENTITY REFERENCE MISSING", rendered_text)
|
|
self.assertNotIn("IDENTITY REFERENCE IMAGE(S)", rendered_text)
|
|
self.assertEqual(transported_images, frame_markers)
|
|
|
|
retry = self.nodes._focused_retry_context(context)
|
|
self.assertEqual(int(retry.images.shape[0]), 3)
|
|
self.assertEqual(retry.media_metadata["sampled_indices"], [0, 30, 50])
|
|
self.assertEqual(
|
|
retry.media_metadata["grounding_retry_frame_reduction"]["original_video_sample_count"],
|
|
6,
|
|
)
|
|
|
|
def test_rank_four_native_video_transport_uses_processor_token_fallback(self) -> None:
|
|
registry = self.nodes.build_asset_registry(
|
|
{
|
|
"source": "video",
|
|
"sampled_indices": [0, 10, 20],
|
|
"source_fps": 10.0,
|
|
"video_role": "primary_video_source",
|
|
},
|
|
3,
|
|
)
|
|
inputs = {
|
|
"input_ids": torch.tensor([[1, 222, 2]], dtype=torch.long),
|
|
"pixel_values_videos": torch.zeros((1, 3, 2, 4)),
|
|
"video_position_ids": torch.zeros((1, 3, 2, 2), dtype=torch.long),
|
|
}
|
|
model = SimpleNamespace(
|
|
config=SimpleNamespace(image_token_id=111, video_token_id=None)
|
|
)
|
|
processor = SimpleNamespace(image_token_id=111, video_token_id=222)
|
|
messages = [
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{"type": "video", "video": [object(), object(), object()]},
|
|
{"type": "text", "text": "Describe."},
|
|
],
|
|
}
|
|
]
|
|
|
|
proof = self.nodes._processor_transport_proof(
|
|
inputs, model, processor, messages, registry
|
|
)
|
|
self.assertTrue(proof["pixel_transport_confirmed"])
|
|
self.assertEqual(proof["processor_observed_sample_count"], 3)
|
|
self.assertEqual(proof["video_media_token_count"], 1)
|
|
self.assertEqual(proof["processor_observed_position_sample_count"], 3)
|
|
self.assertEqual(proof["effective_visual_token_budget"], 1)
|
|
self.assertEqual(proof["processor_pixel_patch_count"], 6)
|
|
|
|
def test_transport_rejects_incomplete_tokens_and_position_sample_counts(self) -> None:
|
|
registry = self.nodes.build_asset_registry(
|
|
{"source": "video", "sampled_indices": [0, 1, 2]},
|
|
3,
|
|
)
|
|
messages = [
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{"type": "image", "image": object()},
|
|
{"type": "image", "image": object()},
|
|
{"type": "image", "image": object()},
|
|
{"type": "text", "text": "Describe."},
|
|
],
|
|
}
|
|
]
|
|
processor = SimpleNamespace(image_token_id=111, video_token_id=222)
|
|
model = SimpleNamespace(config=SimpleNamespace(image_token_id=111, video_token_id=222))
|
|
inputs = {
|
|
"input_ids": torch.tensor([[1, 111, 2]], dtype=torch.long),
|
|
"pixel_values": torch.zeros((3, 2, 4)),
|
|
"image_position_ids": torch.zeros((2, 2, 2), dtype=torch.long),
|
|
}
|
|
proof = self.nodes._processor_transport_proof(
|
|
inputs, model, processor, messages, registry
|
|
)
|
|
self.assertFalse(proof["pixel_transport_confirmed"])
|
|
self.assertTrue(
|
|
any("media_token_coverage_mismatch" in reason for reason in proof["mismatch_reasons"])
|
|
)
|
|
self.assertTrue(
|
|
any("position_sample_count_mismatch" in reason for reason in proof["mismatch_reasons"])
|
|
)
|
|
|
|
def test_missing_telemetry_callbacks_are_detected_as_integrity_errors(self) -> None:
|
|
result = self.nodes.BackendRunResult(
|
|
decoded_text="{}",
|
|
telemetry={
|
|
"schema_version": "dg-denoising-telemetry/1",
|
|
"forward_count": 0,
|
|
"snapshots": [],
|
|
"stream_ended": False,
|
|
"final_output": {},
|
|
"errors": [],
|
|
},
|
|
)
|
|
error = self.nodes._telemetry_error_text(result)
|
|
self.assertIn("logits_processor_callback_not_observed", error)
|
|
self.assertIn("draft_streamer_snapshot_not_observed", error)
|
|
self.assertIn("draft_streamer_end_not_observed", error)
|
|
|
|
def test_off_delegates_to_frozen_legacy_path(self) -> None:
|
|
runtime = self.runtime()
|
|
context = self.image_context()
|
|
expected_packet = self.packet("legacy prompt")
|
|
expected_packet["metadata"]["media"] = {
|
|
"pixels_sent_to_backend": True,
|
|
"pixel_transport_confirmed": True,
|
|
}
|
|
expected_result = self.legacy_result(expected_packet)
|
|
with patch.object(
|
|
self.nodes, "_run_generation_packet_legacy", return_value=expected_result
|
|
) as legacy, patch.object(self.nodes, "_run_backend_detailed") as detailed:
|
|
result = self.nodes._run_generation_packet(
|
|
runtime,
|
|
context,
|
|
self.target(),
|
|
grounding_guard_config={"mode": "off"},
|
|
)
|
|
|
|
legacy.assert_called_once()
|
|
detailed.assert_not_called()
|
|
self.assertEqual(result[0]["ltx_prompt"], "legacy prompt")
|
|
self.assertEqual(result[0]["metadata"]["grounding_guard"]["decision"], "disabled")
|
|
self.assertTrue(result[0]["metadata"]["media"]["pixels_sent_to_backend"])
|
|
self.assertTrue(result[0]["metadata"]["media"]["pixel_transport_confirmed"])
|
|
self.assertEqual(
|
|
result[0]["metadata"]["grounding_guard"]["effective_sampling"]["profile"],
|
|
"legacy_scalar_temperature",
|
|
)
|
|
self.assertNotIn("backend_options", legacy.call_args.kwargs)
|
|
|
|
def test_audit_warning_does_not_change_readiness_or_prompt(self) -> None:
|
|
runtime = self.runtime()
|
|
context = self.image_context()
|
|
ledger = self.uncertain_ledger()
|
|
initial_packet = self.packet("keep this prompt", ledger)
|
|
|
|
captured_kwargs: dict = {}
|
|
|
|
def audit_legacy(*_args, **kwargs):
|
|
captured_kwargs.update(kwargs)
|
|
kwargs["backend_results"].append(
|
|
self.backend_result(ledger, transport_confirmed=True)
|
|
)
|
|
return self.legacy_result(initial_packet)
|
|
|
|
with patch.object(
|
|
self.nodes, "_run_generation_packet_legacy", side_effect=audit_legacy
|
|
), patch.object(self.nodes, "_maybe_unload") as unload:
|
|
result = self.nodes._run_generation_packet(
|
|
runtime,
|
|
context,
|
|
self.target(),
|
|
grounding_guard_config={"mode": "audit", "seed": 9},
|
|
)
|
|
|
|
packet, metadata = result[0], result[-1]
|
|
self.assertEqual(packet["ltx_prompt"], "keep this prompt")
|
|
self.assertTrue(metadata["ready_for_generation"])
|
|
self.assertEqual(metadata["blocked_reasons"], [])
|
|
self.assertEqual(metadata["grounding_guard"]["decision"], "warn")
|
|
self.assertTrue(metadata["grounding_guard"]["grounding_guard_would_block"])
|
|
self.assertNotIn("grounding_ledger", packet)
|
|
self.assertNotIn("max_refinement_attempts_override", captured_kwargs)
|
|
unload.assert_called_once_with(runtime)
|
|
|
|
def test_audit_h3_parse_invalid_retries_once_and_uses_retry_result(self) -> None:
|
|
runtime = self.runtime()
|
|
context = self.image_context()
|
|
registry = self.nodes._grounding_asset_registry(context)
|
|
ledger = self.grounded_ledger(registry)
|
|
target = self.nodes.TargetProfileConfig(
|
|
target_profile="minimax_h3",
|
|
minimax_h3_mode="ref2va",
|
|
)
|
|
|
|
invalid_packet = self.packet("")
|
|
invalid_packet["ltx_prompt"] = ""
|
|
invalid_packet["minimax_h3_prompt"] = "salvaged first attempt"
|
|
invalid_packet["metadata"]["target_profile"] = "minimax_h3"
|
|
invalid_packet["metadata"]["ready_for_generation"] = False
|
|
|
|
valid_packet = copy.deepcopy(invalid_packet)
|
|
valid_packet["minimax_h3_prompt"] = "strict JSON retry result"
|
|
valid_packet["metadata"]["ready_for_generation"] = True
|
|
valid_packet["metadata"]["blocked_reasons"] = []
|
|
|
|
observed_stages: list[str] = []
|
|
|
|
def audit_legacy(*_args, **kwargs):
|
|
backend_options = kwargs["backend_options"]
|
|
backend_options.call_budget.claim()
|
|
stage = backend_options.stage
|
|
observed_stages.append(stage)
|
|
result = self.backend_result(ledger, transport_confirmed=True)
|
|
result.stage = stage
|
|
kwargs["backend_results"].append(result)
|
|
if stage == "audit_director":
|
|
return (
|
|
copy.deepcopy(invalid_packet),
|
|
"not strict JSON",
|
|
"",
|
|
"",
|
|
False,
|
|
"salvaged malformed output",
|
|
copy.deepcopy(invalid_packet["metadata"]),
|
|
)
|
|
return self.legacy_result(copy.deepcopy(valid_packet))
|
|
|
|
with patch.object(
|
|
self.nodes, "_run_generation_packet_legacy", side_effect=audit_legacy
|
|
) as legacy, patch.object(self.nodes, "_maybe_unload") as unload:
|
|
result = self.nodes._run_generation_packet(
|
|
runtime,
|
|
context,
|
|
target,
|
|
grounding_guard_config={
|
|
"mode": "audit",
|
|
"retry_on_uncertain": True,
|
|
},
|
|
)
|
|
|
|
packet, metadata = result[0], result[-1]
|
|
report = metadata["grounding_guard"]
|
|
self.assertEqual(legacy.call_count, 2)
|
|
self.assertEqual(observed_stages, ["audit_director", "audit_director_retry"])
|
|
self.assertEqual(packet["minimax_h3_prompt"], "strict JSON retry result")
|
|
self.assertTrue(metadata["ready_for_generation"])
|
|
self.assertEqual(report["attempt_count"], 2)
|
|
self.assertEqual(report["model_call_count"], 2)
|
|
self.assertEqual(
|
|
report["retry_reasons"],
|
|
["audit_retry:prompt_packet_was_not_strict_json"],
|
|
)
|
|
self.assertEqual(
|
|
[call["stage"] for call in report["trajectory_summary"]["calls"]],
|
|
["audit_director", "audit_director_retry"],
|
|
)
|
|
unload.assert_called_once_with(runtime)
|
|
|
|
def test_audit_h3_parse_invalid_does_not_retry_when_retry_is_disabled(self) -> None:
|
|
runtime = self.runtime()
|
|
context = self.image_context()
|
|
ledger = self.grounded_ledger(
|
|
self.nodes._grounding_asset_registry(context)
|
|
)
|
|
target = self.nodes.TargetProfileConfig(
|
|
target_profile="minimax_h3",
|
|
minimax_h3_mode="ref2va",
|
|
)
|
|
invalid_packet = self.packet("")
|
|
invalid_packet["ltx_prompt"] = ""
|
|
invalid_packet["minimax_h3_prompt"] = "salvaged invalid H3 output"
|
|
invalid_packet["metadata"].update(
|
|
{
|
|
"target_profile": "minimax_h3",
|
|
"ready_for_generation": False,
|
|
"blocked_reasons": ["json_parse_invalid", "salvaged_output"],
|
|
}
|
|
)
|
|
observed_stages: list[str] = []
|
|
|
|
def audit_legacy(*_args, **kwargs):
|
|
backend_options = kwargs["backend_options"]
|
|
backend_options.call_budget.claim()
|
|
observed_stages.append(backend_options.stage)
|
|
result = self.backend_result(ledger, transport_confirmed=True)
|
|
result.stage = backend_options.stage
|
|
kwargs["backend_results"].append(result)
|
|
return (
|
|
copy.deepcopy(invalid_packet),
|
|
"not strict JSON",
|
|
"",
|
|
"",
|
|
False,
|
|
"salvaged malformed output",
|
|
copy.deepcopy(invalid_packet["metadata"]),
|
|
)
|
|
|
|
with patch.object(
|
|
self.nodes, "_run_generation_packet_legacy", side_effect=audit_legacy
|
|
) as legacy, patch.object(self.nodes, "_maybe_unload") as unload:
|
|
result = self.nodes._run_generation_packet(
|
|
runtime,
|
|
context,
|
|
target,
|
|
grounding_guard_config={
|
|
"mode": "audit",
|
|
"retry_on_uncertain": False,
|
|
},
|
|
)
|
|
|
|
metadata = result[-1]
|
|
report = metadata["grounding_guard"]
|
|
self.assertEqual(legacy.call_count, 1)
|
|
self.assertEqual(observed_stages, ["audit_director"])
|
|
self.assertFalse(result[4])
|
|
self.assertTrue(result[5])
|
|
self.assertFalse(metadata["ready_for_generation"])
|
|
self.assertEqual(report["attempt_count"], 1)
|
|
self.assertEqual(report["model_call_count"], 1)
|
|
self.assertEqual(report["retry_reasons"], [])
|
|
unload.assert_called_once_with(runtime)
|
|
|
|
def test_audit_ltx_parse_invalid_retries_once_and_uses_retry_result(self) -> None:
|
|
runtime = self.runtime()
|
|
context = self.image_context()
|
|
ledger = self.grounded_ledger(
|
|
self.nodes._grounding_asset_registry(context)
|
|
)
|
|
invalid_packet = self.packet("salvaged invalid LTX output")
|
|
invalid_packet["metadata"].update(
|
|
{
|
|
"ready_for_generation": False,
|
|
"blocked_reasons": ["json_parse_invalid", "salvaged_output"],
|
|
}
|
|
)
|
|
valid_packet = self.packet("strict JSON LTX retry result")
|
|
observed_stages: list[str] = []
|
|
|
|
def audit_legacy(*_args, **kwargs):
|
|
backend_options = kwargs["backend_options"]
|
|
backend_options.call_budget.claim()
|
|
observed_stages.append(backend_options.stage)
|
|
result = self.backend_result(ledger, transport_confirmed=True)
|
|
result.stage = backend_options.stage
|
|
kwargs["backend_results"].append(result)
|
|
if backend_options.stage == "audit_director_retry":
|
|
return self.legacy_result(copy.deepcopy(valid_packet))
|
|
return (
|
|
copy.deepcopy(invalid_packet),
|
|
"not strict JSON",
|
|
"",
|
|
"",
|
|
False,
|
|
"salvaged malformed output",
|
|
copy.deepcopy(invalid_packet["metadata"]),
|
|
)
|
|
|
|
with patch.object(
|
|
self.nodes, "_run_generation_packet_legacy", side_effect=audit_legacy
|
|
) as legacy, patch.object(self.nodes, "_maybe_unload") as unload:
|
|
result = self.nodes._run_generation_packet(
|
|
runtime,
|
|
context,
|
|
self.target(),
|
|
grounding_guard_config={
|
|
"mode": "audit",
|
|
"retry_on_uncertain": True,
|
|
},
|
|
)
|
|
|
|
packet, metadata = result[0], result[-1]
|
|
report = metadata["grounding_guard"]
|
|
self.assertEqual(legacy.call_count, 2)
|
|
self.assertEqual(observed_stages, ["audit_director", "audit_director_retry"])
|
|
self.assertEqual(packet["ltx_prompt"], "strict JSON LTX retry result")
|
|
self.assertTrue(result[4])
|
|
self.assertFalse(result[5])
|
|
self.assertTrue(metadata["ready_for_generation"])
|
|
self.assertEqual(report["attempt_count"], 2)
|
|
self.assertEqual(report["model_call_count"], 2)
|
|
self.assertEqual(
|
|
report["retry_reasons"],
|
|
["audit_retry:prompt_packet_was_not_strict_json"],
|
|
)
|
|
self.assertEqual(
|
|
[call["stage"] for call in report["trajectory_summary"]["calls"]],
|
|
["audit_director", "audit_director_retry"],
|
|
)
|
|
unload.assert_called_once_with(runtime)
|
|
|
|
def test_audit_ltx_schema_invalid_ledger_retries_once(self) -> None:
|
|
runtime = self.runtime()
|
|
context = self.image_context()
|
|
registry = self.nodes._grounding_asset_registry(context)
|
|
valid_ledger = self.grounded_ledger(registry)
|
|
schema_invalid_ledger = {
|
|
"schema": "dg-grounding-ledger/1",
|
|
"analysis_status": "grounded",
|
|
}
|
|
packet = self.packet("valid LTX packet on both audit attempts")
|
|
observed_stages: list[str] = []
|
|
|
|
def audit_legacy(*_args, **kwargs):
|
|
backend_options = kwargs["backend_options"]
|
|
backend_options.call_budget.claim()
|
|
stage = backend_options.stage
|
|
observed_stages.append(stage)
|
|
ledger = (
|
|
valid_ledger
|
|
if stage == "audit_director_retry"
|
|
else schema_invalid_ledger
|
|
)
|
|
result = self.backend_result(ledger, transport_confirmed=True)
|
|
result.stage = stage
|
|
kwargs["backend_results"].append(result)
|
|
return self.legacy_result(copy.deepcopy(packet))
|
|
|
|
with patch.object(
|
|
self.nodes, "_run_generation_packet_legacy", side_effect=audit_legacy
|
|
) as legacy, patch.object(self.nodes, "_maybe_unload") as unload:
|
|
result = self.nodes._run_generation_packet(
|
|
runtime,
|
|
context,
|
|
self.target(),
|
|
grounding_guard_config={
|
|
"mode": "audit",
|
|
"retry_on_uncertain": True,
|
|
},
|
|
)
|
|
|
|
metadata = result[-1]
|
|
report = metadata["grounding_guard"]
|
|
self.assertEqual(legacy.call_count, 2)
|
|
self.assertEqual(observed_stages, ["audit_director", "audit_director_retry"])
|
|
self.assertTrue(metadata["ready_for_generation"])
|
|
self.assertEqual(report["attempt_count"], 2)
|
|
self.assertEqual(report["model_call_count"], 2)
|
|
self.assertEqual(
|
|
report["retry_reasons"],
|
|
["audit_retry:grounding_ledger_schema_invalid"],
|
|
)
|
|
self.assertTrue(report["validation"]["schema_valid"])
|
|
self.assertEqual(report["validation"]["blocking_reasons"], [])
|
|
self.assertEqual(
|
|
[call["stage"] for call in report["trajectory_summary"]["calls"]],
|
|
["audit_director", "audit_director_retry"],
|
|
)
|
|
unload.assert_called_once_with(runtime)
|
|
|
|
def test_audit_ltx_parse_invalid_retry_exhaustion_remains_fail_closed(self) -> None:
|
|
runtime = self.runtime()
|
|
context = self.image_context()
|
|
ledger = self.grounded_ledger(
|
|
self.nodes._grounding_asset_registry(context)
|
|
)
|
|
observed_stages: list[str] = []
|
|
|
|
def audit_legacy(*_args, **kwargs):
|
|
backend_options = kwargs["backend_options"]
|
|
backend_options.call_budget.claim()
|
|
stage = backend_options.stage
|
|
observed_stages.append(stage)
|
|
backend_result = self.backend_result(ledger, transport_confirmed=True)
|
|
backend_result.stage = stage
|
|
kwargs["backend_results"].append(backend_result)
|
|
invalid_packet = self.packet(
|
|
f"salvaged invalid LTX output from {stage}"
|
|
)
|
|
invalid_packet["metadata"].update(
|
|
{
|
|
"ready_for_generation": False,
|
|
"blocked_reasons": ["json_parse_invalid", "salvaged_output"],
|
|
}
|
|
)
|
|
return (
|
|
invalid_packet,
|
|
"not strict JSON",
|
|
"",
|
|
"",
|
|
False,
|
|
"salvaged malformed output",
|
|
invalid_packet["metadata"],
|
|
)
|
|
|
|
with patch.object(
|
|
self.nodes, "_run_generation_packet_legacy", side_effect=audit_legacy
|
|
) as legacy, patch.object(self.nodes, "_maybe_unload") as unload:
|
|
result = self.nodes._run_generation_packet(
|
|
runtime,
|
|
context,
|
|
self.target(),
|
|
grounding_guard_config={
|
|
"mode": "audit",
|
|
"retry_on_uncertain": True,
|
|
},
|
|
)
|
|
|
|
packet, metadata = result[0], result[-1]
|
|
report = metadata["grounding_guard"]
|
|
self.assertEqual(legacy.call_count, 2)
|
|
self.assertEqual(observed_stages, ["audit_director", "audit_director_retry"])
|
|
self.assertFalse(result[4])
|
|
self.assertTrue(result[5])
|
|
self.assertFalse(metadata["ready_for_generation"])
|
|
self.assertIn("json_parse_invalid", metadata["blocked_reasons"])
|
|
self.assertEqual(report["attempt_count"], 2)
|
|
self.assertEqual(report["model_call_count"], 2)
|
|
self.assertEqual(
|
|
report["retry_reasons"],
|
|
["audit_retry:prompt_packet_was_not_strict_json"],
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "did not pass validation"):
|
|
self.nodes.DiffusionGemmaGenerationGate().gate(
|
|
packet["ltx_prompt"],
|
|
metadata["ready_for_generation"],
|
|
json.dumps(metadata),
|
|
)
|
|
unload.assert_called_once_with(runtime)
|
|
|
|
def test_audit_h3_parse_invalid_retry_exhaustion_remains_fail_closed(self) -> None:
|
|
runtime = self.runtime()
|
|
context = self.image_context()
|
|
ledger = self.grounded_ledger(
|
|
self.nodes._grounding_asset_registry(context)
|
|
)
|
|
target = self.nodes.TargetProfileConfig(
|
|
target_profile="minimax_h3",
|
|
minimax_h3_mode="ref2va",
|
|
)
|
|
observed_stages: list[str] = []
|
|
|
|
def audit_legacy(*_args, **kwargs):
|
|
backend_options = kwargs["backend_options"]
|
|
backend_options.call_budget.claim()
|
|
observed_stages.append(backend_options.stage)
|
|
result = self.backend_result(ledger, transport_confirmed=True)
|
|
result.stage = backend_options.stage
|
|
kwargs["backend_results"].append(result)
|
|
packet = self.packet("")
|
|
packet["ltx_prompt"] = ""
|
|
packet["minimax_h3_prompt"] = (
|
|
f"salvaged invalid H3 output from {backend_options.stage}"
|
|
)
|
|
packet["metadata"].update(
|
|
{
|
|
"target_profile": "minimax_h3",
|
|
"ready_for_generation": False,
|
|
"blocked_reasons": [
|
|
"json_parse_invalid",
|
|
"salvaged_output",
|
|
"minimax_h3_ref_missing_summary",
|
|
],
|
|
}
|
|
)
|
|
return (
|
|
packet,
|
|
"not strict JSON",
|
|
"",
|
|
"",
|
|
False,
|
|
"salvaged malformed output",
|
|
packet["metadata"],
|
|
)
|
|
|
|
with patch.object(
|
|
self.nodes, "_run_generation_packet_legacy", side_effect=audit_legacy
|
|
) as legacy, patch.object(self.nodes, "_maybe_unload") as unload:
|
|
result = self.nodes._run_generation_packet(
|
|
runtime,
|
|
context,
|
|
target,
|
|
grounding_guard_config={
|
|
"mode": "audit",
|
|
"retry_on_uncertain": True,
|
|
},
|
|
)
|
|
|
|
packet, metadata = result[0], result[-1]
|
|
report = metadata["grounding_guard"]
|
|
self.assertEqual(legacy.call_count, 2)
|
|
self.assertEqual(observed_stages, ["audit_director", "audit_director_retry"])
|
|
self.assertFalse(result[4])
|
|
self.assertTrue(result[5])
|
|
self.assertFalse(metadata["ready_for_generation"])
|
|
self.assertIn("json_parse_invalid", metadata["blocked_reasons"])
|
|
self.assertEqual(report["attempt_count"], 2)
|
|
self.assertEqual(report["model_call_count"], 2)
|
|
self.assertEqual(
|
|
report["retry_reasons"],
|
|
["audit_retry:prompt_packet_was_not_strict_json"],
|
|
)
|
|
self.assertEqual(
|
|
[call["stage"] for call in report["trajectory_summary"]["calls"]],
|
|
["audit_director", "audit_director_retry"],
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "did not pass validation"):
|
|
self.nodes.DiffusionGemmaGenerationGate().gate(
|
|
packet["minimax_h3_prompt"],
|
|
metadata["ready_for_generation"],
|
|
json.dumps(metadata),
|
|
)
|
|
unload.assert_called_once_with(runtime)
|
|
|
|
def test_audit_preserves_initial_ledger_and_aggregates_repair_telemetry(self) -> None:
|
|
runtime = self.runtime()
|
|
context = self.image_context()
|
|
context.visual_description = "unverified purple dragon description"
|
|
context.media_metadata["visual_description"] = context.visual_description
|
|
context.media_metadata["observations"] = ["unverified purple dragon observation"]
|
|
registry = self.nodes._grounding_asset_registry(context)
|
|
ledger = self.grounded_ledger(registry)
|
|
combined_packet = self.packet("initial prompt", ledger)
|
|
initial = self.backend_result(ledger)
|
|
initial.decoded_text = json.dumps(combined_packet)
|
|
initial.stage = "audit_director"
|
|
repair = self.backend_result(ledger, telemetry_refusal=True)
|
|
repair.decoded_text = "I cannot help with this request."
|
|
repair.stage = "target_repair_1"
|
|
repaired_packet = self.packet("accepted repaired prompt")
|
|
|
|
def audit_legacy(*_args, **kwargs):
|
|
kwargs["backend_results"].extend([initial, repair])
|
|
return self.legacy_result(repaired_packet)
|
|
|
|
with patch.object(
|
|
self.nodes, "_run_generation_packet_legacy", side_effect=audit_legacy
|
|
), patch.object(self.nodes, "_maybe_unload"):
|
|
result = self.nodes._run_generation_packet(
|
|
runtime,
|
|
context,
|
|
self.target(),
|
|
grounding_guard_config={"mode": "audit"},
|
|
)
|
|
|
|
packet, report = result[0], result[-1]["grounding_guard"]
|
|
self.assertEqual(packet["ltx_prompt"], "accepted repaired prompt")
|
|
self.assertTrue(packet["metadata"]["ready_for_generation"])
|
|
self.assertEqual(report["validated_ledger"]["analysis_status"], "grounded")
|
|
self.assertEqual(report["decision"], "warn")
|
|
self.assertEqual(report["analysis_status"], "refused")
|
|
self.assertEqual(report["model_call_count"], 2)
|
|
self.assertEqual(
|
|
[call["stage"] for call in report["trajectory_summary"]["calls"]],
|
|
["audit_director", "target_repair_1"],
|
|
)
|
|
|
|
def test_strict_refusal_blocks_without_compiler_and_unloads_once(self) -> None:
|
|
runtime = self.runtime()
|
|
context = self.image_context()
|
|
refused = self.uncertain_ledger("refused")
|
|
refused["uncertainties"] = []
|
|
refused["grounding_failure_reasons"] = ["I cannot analyze this image."]
|
|
with patch.object(
|
|
self.nodes,
|
|
"_run_backend_detailed",
|
|
return_value=self.backend_result(refused, telemetry_refusal=True),
|
|
) as evidence, patch.object(
|
|
self.nodes, "_run_generation_packet_legacy"
|
|
) as compiler, patch.object(self.nodes, "_maybe_unload") as unload:
|
|
result = self.nodes._run_generation_packet(
|
|
runtime,
|
|
context,
|
|
self.target(),
|
|
grounding_guard_config={
|
|
"mode": "strict",
|
|
"retry_on_uncertain": False,
|
|
},
|
|
)
|
|
|
|
packet, metadata = result[0], result[-1]
|
|
evidence.assert_called_once()
|
|
compiler.assert_not_called()
|
|
self.assertEqual(packet["ltx_prompt"], "")
|
|
self.assertFalse(metadata["ready_for_generation"])
|
|
self.assertEqual(
|
|
metadata["blocked_reasons"][:2],
|
|
["visual_grounding_unverified", "visual_grounding_refused"],
|
|
)
|
|
unload.assert_called_once_with(runtime)
|
|
|
|
def test_strict_grounded_pass_compiles_from_ledger_without_pixels(self) -> None:
|
|
runtime = self.runtime()
|
|
context = self.image_context()
|
|
registry = self.nodes._grounding_asset_registry(context)
|
|
ledger = self.grounded_ledger(registry)
|
|
captured: dict = {}
|
|
|
|
def compiler(*args, **kwargs):
|
|
compiler_context = args[1]
|
|
captured["context"] = compiler_context
|
|
captured["options"] = kwargs["backend_options"]
|
|
captured["max_repairs"] = kwargs["max_refinement_attempts_override"]
|
|
packet = self.packet("ledger-only compiler prompt")
|
|
packet["metadata"]["grounding_evidence_report_id"] = compiler_context.media_metadata[
|
|
"grounding_evidence_report_id"
|
|
]
|
|
packet["metadata"]["used_grounding_fact_ids"] = [
|
|
fact["fact_id"]
|
|
for fact in compiler_context.media_metadata["verified_grounding_ledger"][
|
|
"observed_facts"
|
|
]
|
|
]
|
|
packet["grounding_ledger"] = {"spoofed": True}
|
|
return self.legacy_result(packet)
|
|
|
|
with patch.object(
|
|
self.nodes,
|
|
"_run_backend_detailed",
|
|
return_value=self.backend_result(ledger),
|
|
), patch.object(
|
|
self.nodes, "_run_generation_packet_legacy", side_effect=compiler
|
|
) as compiler_mock, patch.object(self.nodes, "_maybe_unload") as unload:
|
|
result = self.nodes._run_generation_packet(
|
|
runtime,
|
|
context,
|
|
self.target(),
|
|
grounding_guard_config={"mode": "strict", "seed": 17},
|
|
)
|
|
|
|
compiler_mock.assert_called_once()
|
|
compiler_context = captured["context"]
|
|
self.assertIsNone(compiler_context.images)
|
|
self.assertEqual(
|
|
compiler_context.media_metadata["verified_grounding_ledger"], ledger
|
|
)
|
|
self.assertFalse(compiler_context.media_metadata["pixels_sent_to_backend"])
|
|
self.assertNotIn("visual_description", compiler_context.media_metadata)
|
|
self.assertNotIn("observations", compiler_context.media_metadata)
|
|
self.assertNotIn("purple dragon", json.dumps(compiler_context.media_metadata))
|
|
self.assertEqual(captured["options"].asset_registry["expected_image_count"], 0)
|
|
self.assertEqual(captured["max_repairs"], 2)
|
|
packet, metadata = result[0], result[-1]
|
|
self.assertEqual(packet["ltx_prompt"], "ledger-only compiler prompt")
|
|
self.assertNotIn("grounding_ledger", packet)
|
|
self.assertEqual(metadata["grounding_guard"]["decision"], "pass")
|
|
self.assertEqual(
|
|
metadata["grounding_evidence_report_id"],
|
|
metadata["grounding_guard"]["evidence_report_id"],
|
|
)
|
|
self.assertEqual(
|
|
metadata["used_grounding_fact_ids"],
|
|
[fact["fact_id"] for fact in ledger["observed_facts"]],
|
|
)
|
|
self.assertEqual(
|
|
metadata["available_grounding_fact_ids"],
|
|
[fact["fact_id"] for fact in ledger["observed_facts"]],
|
|
)
|
|
self.assertFalse(metadata["grounding_compiler_pixels_sent_to_backend"])
|
|
self.assertEqual(
|
|
metadata["grounding_guard"]["compiler_transport"],
|
|
{
|
|
"pixels_sent_to_backend": False,
|
|
"evidence_source": "validated_ledger_only",
|
|
},
|
|
)
|
|
unload.assert_called_once_with(runtime)
|
|
|
|
def test_strict_compiler_unknown_fact_provenance_blocks(self) -> None:
|
|
runtime = self.runtime()
|
|
context = self.image_context()
|
|
ledger = self.grounded_ledger(self.nodes._grounding_asset_registry(context))
|
|
|
|
def compiler(*args, **_kwargs):
|
|
compiler_context = args[1]
|
|
packet = self.packet("untrusted compiler prompt")
|
|
packet["metadata"]["grounding_evidence_report_id"] = compiler_context.media_metadata[
|
|
"grounding_evidence_report_id"
|
|
]
|
|
packet["metadata"]["used_grounding_fact_ids"] = ["invented-fact-id"]
|
|
return self.legacy_result(packet)
|
|
|
|
with patch.object(
|
|
self.nodes, "_run_backend_detailed", return_value=self.backend_result(ledger)
|
|
), patch.object(
|
|
self.nodes, "_run_generation_packet_legacy", side_effect=compiler
|
|
), patch.object(self.nodes, "_maybe_unload"):
|
|
result = self.nodes._run_generation_packet(
|
|
runtime,
|
|
context,
|
|
self.target(),
|
|
grounding_guard_config={"mode": "strict", "retry_on_uncertain": False},
|
|
)
|
|
|
|
self.assertEqual(result[0]["ltx_prompt"], "")
|
|
self.assertFalse(result[-1]["ready_for_generation"])
|
|
self.assertEqual(
|
|
result[-1]["blocked_reasons"][:2],
|
|
["visual_grounding_unverified", "visual_grounding_schema_invalid"],
|
|
)
|
|
|
|
def test_strict_compiler_readiness_failure_preserves_validator_reasons(self) -> None:
|
|
runtime = self.runtime()
|
|
context = self.image_context()
|
|
ledger = self.grounded_ledger(self.nodes._grounding_asset_registry(context))
|
|
|
|
def compiler(*args, **_kwargs):
|
|
compiler_context = args[1]
|
|
packet = self.packet("")
|
|
packet["metadata"]["ready_for_generation"] = False
|
|
packet["metadata"]["blocked_reasons"] = [
|
|
"minimax_h3_cut_timestamp_invalid"
|
|
]
|
|
packet["metadata"]["grounding_evidence_report_id"] = (
|
|
compiler_context.media_metadata["grounding_evidence_report_id"]
|
|
)
|
|
packet["metadata"]["used_grounding_fact_ids"] = [
|
|
fact["fact_id"]
|
|
for fact in compiler_context.media_metadata["verified_grounding_ledger"][
|
|
"observed_facts"
|
|
]
|
|
]
|
|
packet["metadata"]["minimax_h3_refinement"] = {
|
|
"candidate_reasons": ["minimax_h3_noncanonical_time_range"]
|
|
}
|
|
return self.legacy_result(packet)
|
|
|
|
with patch.object(
|
|
self.nodes, "_run_backend_detailed", return_value=self.backend_result(ledger)
|
|
), patch.object(
|
|
self.nodes, "_run_generation_packet_legacy", side_effect=compiler
|
|
), patch.object(self.nodes, "_maybe_unload"):
|
|
result = self.nodes._run_generation_packet(
|
|
runtime,
|
|
context,
|
|
self.target(),
|
|
grounding_guard_config={"mode": "strict", "retry_on_uncertain": False},
|
|
)
|
|
|
|
packet, metadata = result[0], result[-1]
|
|
self.assertEqual(packet["ltx_prompt"], "")
|
|
self.assertFalse(metadata["ready_for_generation"])
|
|
retry_detail = metadata["grounding_guard"]["retry_reasons"][0]
|
|
self.assertIn("minimax_h3_cut_timestamp_invalid", retry_detail)
|
|
self.assertIn("minimax_h3_noncanonical_time_range", retry_detail)
|
|
|
|
gate = self.nodes.DiffusionGemmaGenerationGate()
|
|
with self.assertRaises(ValueError) as raised:
|
|
gate.gate("", False, json.dumps(metadata))
|
|
self.assertIn("minimax_h3_cut_timestamp_invalid", str(raised.exception))
|
|
self.assertIn("minimax_h3_noncanonical_time_range", str(raised.exception))
|
|
|
|
def test_strict_compiler_retries_malformed_json_once_within_call_budget(self) -> None:
|
|
runtime = self.runtime()
|
|
context = self.image_context()
|
|
ledger = self.grounded_ledger(self.nodes._grounding_asset_registry(context))
|
|
calls: list[dict] = []
|
|
|
|
def compiler(*args, **kwargs):
|
|
compiler_context = args[1]
|
|
options = kwargs["backend_options"]
|
|
options.call_budget.claim()
|
|
calls.append(
|
|
{
|
|
"stage": options.stage,
|
|
"seed": options.seed,
|
|
"max_repairs": kwargs["max_refinement_attempts_override"],
|
|
}
|
|
)
|
|
packet = self.packet("ledger-only compiler prompt")
|
|
if len(calls) == 1:
|
|
return packet, "malformed compiler output", "", "", False, "", packet["metadata"]
|
|
packet["metadata"]["grounding_evidence_report_id"] = (
|
|
compiler_context.media_metadata["grounding_evidence_report_id"]
|
|
)
|
|
packet["metadata"]["used_grounding_fact_ids"] = [
|
|
fact["fact_id"]
|
|
for fact in compiler_context.media_metadata["verified_grounding_ledger"][
|
|
"observed_facts"
|
|
]
|
|
]
|
|
return self.legacy_result(packet)
|
|
|
|
with patch.object(
|
|
self.nodes, "_run_backend_detailed", return_value=self.backend_result(ledger)
|
|
), patch.object(
|
|
self.nodes, "_run_generation_packet_legacy", side_effect=compiler
|
|
), patch.object(self.nodes, "_maybe_unload"):
|
|
result = self.nodes._run_generation_packet(
|
|
runtime,
|
|
context,
|
|
self.target(),
|
|
grounding_guard_config={"mode": "strict", "seed": 9},
|
|
)
|
|
|
|
self.assertEqual(
|
|
calls,
|
|
[
|
|
{"stage": "compiler", "seed": 1009, "max_repairs": 2},
|
|
{"stage": "compiler_retry", "seed": 1009, "max_repairs": 1},
|
|
],
|
|
)
|
|
packet, metadata = result[0], result[-1]
|
|
self.assertEqual(packet["ltx_prompt"], "ledger-only compiler prompt")
|
|
self.assertTrue(metadata["ready_for_generation"])
|
|
self.assertEqual(metadata["grounding_guard"]["model_call_count"], 3)
|
|
self.assertIn(
|
|
"compiler_retry:compiler_packet_was_not_strict_json",
|
|
metadata["grounding_guard"]["retry_reasons"],
|
|
)
|
|
|
|
def test_compiler_provenance_must_cover_every_required_asset(self) -> None:
|
|
metadata = {
|
|
"grounding_evidence_report_id": "dggr-test",
|
|
"used_grounding_fact_ids": ["fact-image"],
|
|
}
|
|
fact_assets = {
|
|
"fact-image": {"image:1"},
|
|
"fact-video": {"video:1"},
|
|
}
|
|
with self.assertRaisesRegex(
|
|
self.nodes.GroundingLedgerError,
|
|
"compiler_fact_coverage_missing_assets:video:1",
|
|
):
|
|
self.nodes._validated_compiler_provenance(
|
|
metadata,
|
|
"dggr-test",
|
|
["fact-image", "fact-video"],
|
|
fact_assets,
|
|
["image:1", "video:1"],
|
|
)
|
|
|
|
metadata["used_grounding_fact_ids"] = ["fact-image", "fact-video"]
|
|
self.assertEqual(
|
|
self.nodes._validated_compiler_provenance(
|
|
metadata,
|
|
"dggr-test",
|
|
["fact-image", "fact-video"],
|
|
fact_assets,
|
|
["image:1", "video:1"],
|
|
),
|
|
["fact-image", "fact-video"],
|
|
)
|
|
|
|
def test_h3_target_repair_uses_only_the_same_ledger_context(self) -> None:
|
|
runtime = self.runtime()
|
|
ledger = self.grounded_ledger(
|
|
self.nodes._grounding_asset_registry(self.image_context())
|
|
)
|
|
compiler_context = self.nodes.GemmaContext(
|
|
user_prompt="Create a concise H3 shot.",
|
|
images=None,
|
|
source="image",
|
|
media_metadata={
|
|
"source": "image",
|
|
"verified_grounding_ledger": ledger,
|
|
"pixels_sent_to_backend": False,
|
|
"visual_grounding_mode": "verified_ledger",
|
|
},
|
|
)
|
|
target = self.nodes.TargetProfileConfig(
|
|
target_profile="minimax_h3",
|
|
minimax_h3_mode="t2va",
|
|
)
|
|
invalid_packet = self.packet("")
|
|
invalid_packet["minimax_h3_prompt"] = "invalid timeline"
|
|
invalid_packet["metadata"]["target_profile"] = "minimax_h3"
|
|
invalid_packet["metadata"]["grounding_evidence_report_id"] = "dggr-test"
|
|
invalid_packet["metadata"]["used_grounding_fact_ids"] = [
|
|
ledger["observed_facts"][0]["fact_id"]
|
|
]
|
|
repaired_packet = copy.deepcopy(invalid_packet)
|
|
repaired_packet["minimax_h3_prompt"] = "verified repaired timeline"
|
|
calls: list[tuple] = []
|
|
|
|
def detailed(_config, prompt, media_context, _max_tokens, options, **_kwargs):
|
|
calls.append((prompt, media_context, copy.copy(options)))
|
|
return self.nodes.BackendRunResult(decoded_text=f"response-{len(calls)}")
|
|
|
|
def parse_packet(raw, *_args, **_kwargs):
|
|
packet = invalid_packet if raw == "response-1" else repaired_packet
|
|
return copy.deepcopy(packet), True, ""
|
|
|
|
def h3_reasons(prompt, *_args, **_kwargs):
|
|
return [] if "verified repaired" in prompt else ["minimax_h3_empty_visual_timeline"]
|
|
|
|
budget = self.nodes.BackendCallBudget(max_calls=4)
|
|
options = self.nodes.BackendRunOptions(
|
|
sampling_profile="checkpoint_defaults",
|
|
seed=31,
|
|
enable_telemetry=True,
|
|
asset_registry=self.nodes.build_asset_registry({}, 0),
|
|
call_budget=budget,
|
|
stage="compiler",
|
|
)
|
|
backend_results: list = []
|
|
with patch.object(
|
|
self.nodes, "_run_backend_detailed", side_effect=detailed
|
|
), patch.object(
|
|
self.nodes, "repair_or_salvage_prompt_packet", side_effect=parse_packet
|
|
), patch.object(
|
|
self.nodes, "_minimax_h3_prompt_validation_reasons", side_effect=h3_reasons
|
|
):
|
|
result = self.nodes._run_generation_packet_legacy(
|
|
runtime,
|
|
compiler_context,
|
|
target,
|
|
runtime_required=True,
|
|
backend_options=options,
|
|
backend_results=backend_results,
|
|
manage_unload=False,
|
|
max_refinement_attempts_override=2,
|
|
refinement_evidence_context=json.dumps(ledger, sort_keys=True),
|
|
)
|
|
|
|
self.assertEqual(len(calls), 2)
|
|
self.assertEqual(budget.attempted_calls, 2)
|
|
self.assertEqual([item.stage for item in backend_results], ["compiler", "target_repair_1"])
|
|
self.assertTrue(
|
|
all(call[1] is None or call[1].images is None for call in calls)
|
|
)
|
|
self.assertNotIn("VERIFIED VISUAL EVIDENCE LEDGER", calls[0][0])
|
|
self.assertIn("VERIFIED VISUAL EVIDENCE LEDGER", calls[1][0])
|
|
self.assertIn(ledger["observed_facts"][0]["fact_id"], calls[1][0])
|
|
self.assertEqual(result[0]["minimax_h3_prompt"], "verified repaired timeline")
|
|
self.assertEqual(result[0]["metadata"]["grounding_evidence_report_id"], "dggr-test")
|
|
self.assertEqual(
|
|
result[0]["metadata"]["used_grounding_fact_ids"],
|
|
[ledger["observed_facts"][0]["fact_id"]],
|
|
)
|
|
|
|
def test_legacy_runtime_exception_unloads_once(self) -> None:
|
|
runtime = self.runtime()
|
|
with patch.object(
|
|
self.nodes,
|
|
"_run_backend_for_packet",
|
|
side_effect=RuntimeError("backend exploded"),
|
|
), patch.object(self.nodes, "_release_transformers_runtime") as release:
|
|
with self.assertRaisesRegex(RuntimeError, "runtime_required is true"):
|
|
self.nodes._run_generation_packet_legacy(
|
|
runtime,
|
|
self.image_context(),
|
|
self.target(),
|
|
runtime_required=True,
|
|
)
|
|
release.assert_called_once()
|
|
|
|
def test_strict_compiler_failure_does_not_relabel_valid_evidence_as_bad_schema(self) -> None:
|
|
runtime = self.runtime()
|
|
context = self.image_context()
|
|
registry = self.nodes._grounding_asset_registry(context)
|
|
ledger = self.grounded_ledger(registry)
|
|
|
|
with patch.object(
|
|
self.nodes,
|
|
"_run_backend_detailed",
|
|
return_value=self.backend_result(ledger),
|
|
), patch.object(
|
|
self.nodes,
|
|
"_run_generation_packet_legacy",
|
|
side_effect=RuntimeError("compiler exploded"),
|
|
), patch.object(self.nodes, "_maybe_unload") as unload:
|
|
result = self.nodes._run_generation_packet(
|
|
runtime,
|
|
context,
|
|
self.target(),
|
|
grounding_guard_config={"mode": "strict", "retry_on_uncertain": False},
|
|
)
|
|
|
|
packet, metadata = result[0], result[-1]
|
|
self.assertFalse(metadata["ready_for_generation"])
|
|
self.assertEqual(packet["ltx_prompt"], "")
|
|
self.assertEqual(
|
|
metadata["blocked_reasons"][:2],
|
|
["visual_grounding_unverified", "visual_grounding_uncertain"],
|
|
)
|
|
self.assertNotIn(
|
|
"visual_grounding_schema_invalid",
|
|
metadata["grounding_guard"]["blocked_reasons"],
|
|
)
|
|
self.assertIsNotNone(metadata["grounding_guard"]["validated_ledger"])
|
|
unload.assert_called_once_with(runtime)
|
|
|
|
def test_truncated_strict_evidence_retries_concisely_and_compiles(self) -> None:
|
|
runtime = self.runtime()
|
|
context = self.image_context()
|
|
evidence_calls: list[tuple] = []
|
|
retry_ledgers: list[dict] = []
|
|
|
|
def evidence(_config, prompt, media_context, max_tokens, options, **_kwargs):
|
|
evidence_calls.append((prompt, media_context, options, max_tokens))
|
|
if len(evidence_calls) == 1:
|
|
return self.truncated_evidence_result(options.asset_registry)
|
|
ledger = self.grounded_ledger(options.asset_registry)
|
|
retry_ledgers.append(ledger)
|
|
return self.backend_result(ledger)
|
|
|
|
def compiler_after_retry(*args, **_kwargs):
|
|
compiler_context = args[1]
|
|
verified_ledger = compiler_context.media_metadata["verified_grounding_ledger"]
|
|
packet = self.packet("compiled after concise evidence retry")
|
|
packet["metadata"]["grounding_evidence_report_id"] = (
|
|
compiler_context.media_metadata["grounding_evidence_report_id"]
|
|
)
|
|
packet["metadata"]["used_grounding_fact_ids"] = [
|
|
verified_ledger["observed_facts"][0]["fact_id"]
|
|
]
|
|
return self.legacy_result(packet)
|
|
|
|
with patch.object(
|
|
self.nodes, "_run_backend_detailed", side_effect=evidence
|
|
), patch.object(
|
|
self.nodes,
|
|
"_run_generation_packet_legacy",
|
|
side_effect=compiler_after_retry,
|
|
) as compiler, patch.object(self.nodes, "_maybe_unload") as unload:
|
|
result = self.nodes._run_generation_packet(
|
|
runtime,
|
|
context,
|
|
self.target(),
|
|
grounding_guard_config={"mode": "strict", "retry_on_uncertain": True},
|
|
max_new_tokens=2048,
|
|
)
|
|
|
|
packet, metadata = result[0], result[-1]
|
|
self.assertEqual(len(evidence_calls), 2)
|
|
self.assertEqual([call[3] for call in evidence_calls], [768, 1024])
|
|
retry_prompt = evidence_calls[1][0]
|
|
self.assertIn("Use no more than 4 observed facts total", retry_prompt)
|
|
self.assertIn("keep each claim at 24 words or fewer", retry_prompt)
|
|
self.assertIn("Completing and closing the strict JSON object takes priority", retry_prompt)
|
|
self.assertEqual(int(evidence_calls[1][1].images.shape[0]), 1)
|
|
self.assertEqual(len(retry_ledgers), 1)
|
|
self.assertLessEqual(len(retry_ledgers[0]["observed_facts"]), 4)
|
|
compiler.assert_called_once()
|
|
self.assertEqual(packet["ltx_prompt"], "compiled after concise evidence retry")
|
|
self.assertTrue(metadata["ready_for_generation"])
|
|
self.assertEqual(metadata["grounding_guard"]["attempt_count"], 2)
|
|
self.assertEqual(metadata["grounding_guard"]["decision"], "pass")
|
|
self.assertTrue(
|
|
any(
|
|
reason.startswith("visual_grounding_schema_invalid:")
|
|
for reason in metadata["grounding_guard"]["retry_reasons"]
|
|
)
|
|
)
|
|
unload.assert_called_once_with(runtime)
|
|
|
|
def test_repeated_truncated_strict_evidence_fails_closed_without_partial_facts(self) -> None:
|
|
runtime = self.runtime()
|
|
context = self.image_context()
|
|
evidence_calls: list[tuple] = []
|
|
|
|
def evidence(_config, prompt, media_context, max_tokens, options, **_kwargs):
|
|
evidence_calls.append((prompt, media_context, options, max_tokens))
|
|
return self.truncated_evidence_result(options.asset_registry)
|
|
|
|
with patch.object(
|
|
self.nodes, "_run_backend_detailed", side_effect=evidence
|
|
) as backend, patch.object(
|
|
self.nodes, "_run_generation_packet_legacy"
|
|
) as compiler, patch.object(self.nodes, "_maybe_unload") as unload:
|
|
result = self.nodes._run_generation_packet(
|
|
runtime,
|
|
context,
|
|
self.target(),
|
|
grounding_guard_config={"mode": "strict", "retry_on_uncertain": True},
|
|
max_new_tokens=2048,
|
|
)
|
|
|
|
packet, metadata = result[0], result[-1]
|
|
report = metadata["grounding_guard"]
|
|
self.assertEqual(backend.call_count, 2)
|
|
self.assertEqual([call[3] for call in evidence_calls], [768, 1024])
|
|
compiler.assert_not_called()
|
|
self.assertEqual(packet["ltx_prompt"], "")
|
|
self.assertFalse(metadata["ready_for_generation"])
|
|
self.assertEqual(
|
|
metadata["blocked_reasons"][:2],
|
|
["visual_grounding_unverified", "visual_grounding_schema_invalid"],
|
|
)
|
|
self.assertEqual(report["attempt_count"], 2)
|
|
self.assertEqual(report["model_call_count"], 2)
|
|
self.assertEqual(len(report["retry_reasons"]), 2)
|
|
self.assertTrue(
|
|
all(
|
|
reason.startswith("visual_grounding_schema_invalid:")
|
|
for reason in report["retry_reasons"]
|
|
)
|
|
)
|
|
self.assertEqual(report["validated_ledger"].get("observed_facts", []), [])
|
|
self.assertEqual(report["validation"]["covered_asset_ids"], [])
|
|
self.assertFalse(report["validation"]["schema_valid"])
|
|
unload.assert_called_once_with(runtime)
|
|
|
|
def test_uncertain_evidence_retries_with_reduced_original_frames(self) -> None:
|
|
runtime = self.runtime()
|
|
context = self.video_context(5)
|
|
evidence_calls: list[tuple] = []
|
|
|
|
def evidence(_config, prompt, media_context, max_tokens, options, **_kwargs):
|
|
evidence_calls.append((prompt, media_context, options, max_tokens))
|
|
if len(evidence_calls) == 1:
|
|
return self.backend_result(self.uncertain_ledger())
|
|
return self.backend_result(self.grounded_ledger(options.asset_registry))
|
|
|
|
def retry_compiler(*args, **_kwargs):
|
|
compiler_context = args[1]
|
|
packet = self.packet("retry compiler prompt")
|
|
packet["metadata"]["grounding_evidence_report_id"] = compiler_context.media_metadata[
|
|
"grounding_evidence_report_id"
|
|
]
|
|
packet["metadata"]["used_grounding_fact_ids"] = [
|
|
compiler_context.media_metadata["verified_grounding_ledger"]["observed_facts"][0][
|
|
"fact_id"
|
|
]
|
|
]
|
|
return self.legacy_result(packet)
|
|
|
|
with patch.object(
|
|
self.nodes, "_run_backend_detailed", side_effect=evidence
|
|
), patch.object(
|
|
self.nodes,
|
|
"_run_generation_packet_legacy",
|
|
side_effect=retry_compiler,
|
|
) as compiler, patch.object(self.nodes, "_maybe_unload") as unload:
|
|
result = self.nodes._run_generation_packet(
|
|
runtime,
|
|
context,
|
|
self.target(),
|
|
grounding_guard_config={
|
|
"mode": "strict",
|
|
"retry_on_uncertain": True,
|
|
"seed": 23,
|
|
"evidence_token_budget": "1280",
|
|
},
|
|
max_new_tokens=512,
|
|
)
|
|
|
|
self.assertEqual(len(evidence_calls), 2)
|
|
retry_prompt, retry_media, retry_options, retry_max_tokens = evidence_calls[1]
|
|
self.assertIn("one permitted focused retry", retry_prompt)
|
|
self.assertEqual(evidence_calls[0][3], 1280)
|
|
self.assertEqual(retry_max_tokens, 1280)
|
|
self.assertEqual(int(retry_media.images.shape[0]), 3)
|
|
self.assertEqual(retry_options.asset_registry["expected_image_count"], 3)
|
|
self.assertEqual(
|
|
retry_media.metadata["grounding_retry_frame_reduction"][
|
|
"selected_video_sample_ordinals"
|
|
],
|
|
[1, 3, 5],
|
|
)
|
|
compiler.assert_called_once()
|
|
self.assertEqual(result[-1]["grounding_guard"]["attempt_count"], 2)
|
|
self.assertEqual(result[-1]["grounding_guard"]["decision"], "pass")
|
|
self.assertEqual(
|
|
result[-1]["grounding_guard"]["config"]["evidence_token_budget"],
|
|
"1280",
|
|
)
|
|
self.assertEqual(
|
|
compiler.call_args.kwargs["max_refinement_attempts_override"], 1
|
|
)
|
|
unload.assert_called_once_with(runtime)
|
|
|
|
def test_retry_exhaustion_blocks_without_compilation(self) -> None:
|
|
runtime = self.runtime()
|
|
context = self.video_context(5)
|
|
with patch.object(
|
|
self.nodes,
|
|
"_run_backend_detailed",
|
|
side_effect=[
|
|
self.backend_result(self.uncertain_ledger()),
|
|
self.backend_result(self.uncertain_ledger()),
|
|
],
|
|
) as evidence, patch.object(
|
|
self.nodes, "_run_generation_packet_legacy"
|
|
) as compiler, patch.object(self.nodes, "_maybe_unload") as unload:
|
|
result = self.nodes._run_generation_packet(
|
|
runtime,
|
|
context,
|
|
self.target(),
|
|
grounding_guard_config={"mode": "strict", "retry_on_uncertain": True},
|
|
)
|
|
|
|
packet, metadata = result[0], result[-1]
|
|
self.assertEqual(evidence.call_count, 2)
|
|
compiler.assert_not_called()
|
|
self.assertEqual(packet["ltx_prompt"], "")
|
|
self.assertFalse(metadata["ready_for_generation"])
|
|
self.assertEqual(metadata["grounding_guard"]["attempt_count"], 2)
|
|
self.assertEqual(metadata["grounding_guard"]["model_call_count"], 2)
|
|
self.assertEqual(
|
|
metadata["blocked_reasons"][:2],
|
|
["visual_grounding_unverified", "visual_grounding_uncertain"],
|
|
)
|
|
unload.assert_called_once_with(runtime)
|
|
|
|
def test_backend_exceptions_exhaust_retry_as_uncertain_not_transport(self) -> None:
|
|
runtime = self.runtime()
|
|
with patch.object(
|
|
self.nodes,
|
|
"_run_backend_detailed",
|
|
side_effect=[RuntimeError("model forward failed"), RuntimeError("model forward failed")],
|
|
) as evidence, patch.object(
|
|
self.nodes, "_run_generation_packet_legacy"
|
|
) as compiler, patch.object(self.nodes, "_maybe_unload") as unload:
|
|
result = self.nodes._run_generation_packet(
|
|
runtime,
|
|
self.image_context(),
|
|
self.target(),
|
|
grounding_guard_config={"mode": "strict", "retry_on_uncertain": True},
|
|
)
|
|
|
|
packet, metadata = result[0], result[-1]
|
|
self.assertEqual(evidence.call_count, 2)
|
|
compiler.assert_not_called()
|
|
self.assertEqual(packet["ltx_prompt"], "")
|
|
self.assertEqual(
|
|
metadata["blocked_reasons"][:2],
|
|
["visual_grounding_unverified", "visual_grounding_uncertain"],
|
|
)
|
|
self.assertNotIn("visual_transport_error", metadata["blocked_reasons"])
|
|
unload.assert_called_once_with(runtime)
|
|
|
|
def test_strict_telemetry_exception_uses_the_telemetry_block_reason(self) -> None:
|
|
runtime = self.runtime()
|
|
with patch.object(
|
|
self.nodes,
|
|
"_run_backend_detailed",
|
|
side_effect=RuntimeError("visual_grounding_telemetry_error:summary failed"),
|
|
) as evidence, patch.object(
|
|
self.nodes, "_run_generation_packet_legacy"
|
|
) as compiler, patch.object(self.nodes, "_maybe_unload") as unload:
|
|
result = self.nodes._run_generation_packet(
|
|
runtime,
|
|
self.image_context(),
|
|
self.target(),
|
|
grounding_guard_config={"mode": "strict", "retry_on_uncertain": True},
|
|
)
|
|
|
|
packet, metadata = result[0], result[-1]
|
|
evidence.assert_called_once()
|
|
compiler.assert_not_called()
|
|
self.assertEqual(packet["ltx_prompt"], "")
|
|
self.assertEqual(
|
|
metadata["blocked_reasons"][:2],
|
|
["visual_grounding_unverified", "visual_grounding_telemetry_error"],
|
|
)
|
|
unload.assert_called_once_with(runtime)
|
|
|
|
def test_strict_target_repair_telemetry_error_preserves_failure_reason(self) -> None:
|
|
runtime = self.runtime()
|
|
context = self.image_context()
|
|
registry = self.nodes._grounding_asset_registry(context)
|
|
ledger = self.grounded_ledger(registry)
|
|
target = self.nodes.TargetProfileConfig(
|
|
target_profile="minimax_h3",
|
|
minimax_h3_mode="t2va",
|
|
)
|
|
compiler_packet = self.packet("")
|
|
compiler_packet["ltx_prompt"] = ""
|
|
compiler_packet["minimax_h3_prompt"] = (
|
|
"[Shot 1] invalid timeline. [Shot 2] missing a required timestamp."
|
|
)
|
|
compiler_packet["metadata"]["target_profile"] = "minimax_h3"
|
|
evidence_report_id = self.nodes._grounding_report_id(ledger, registry)
|
|
stages: list[str] = []
|
|
|
|
def backend_side_effect(_config, _prompt, _media, _tokens, options, **_kwargs):
|
|
stages.append(options.stage)
|
|
if options.stage == "evidence_attempt_1":
|
|
return self.backend_result(ledger)
|
|
if options.stage == "compiler":
|
|
result = self.backend_result(compiler_packet)
|
|
result.decoded_text = (
|
|
"integrated_multimodal_description: [Shot 1] invalid timeline. "
|
|
"[Shot 2] missing a required timestamp.\n\n"
|
|
"overall_soundscape: N/A\n\n"
|
|
"non_diegetic_music: N/A\n\n"
|
|
f"GROUNDING_EVIDENCE_REPORT_ID: {evidence_report_id}\n"
|
|
"USED_GROUNDING_FACT_IDS: fact-1"
|
|
)
|
|
return result
|
|
if options.stage == "target_repair_1":
|
|
raise RuntimeError("visual_grounding_telemetry_error:repair summary failed")
|
|
raise AssertionError(f"unexpected backend stage: {options.stage}")
|
|
|
|
with patch.object(
|
|
self.nodes,
|
|
"_run_backend_detailed",
|
|
side_effect=backend_side_effect,
|
|
), patch.object(self.nodes, "_maybe_unload") as unload:
|
|
result = self.nodes._run_generation_packet(
|
|
runtime,
|
|
context,
|
|
target,
|
|
grounding_guard_config={"mode": "strict", "retry_on_uncertain": False},
|
|
)
|
|
|
|
self.assertEqual(stages, ["evidence_attempt_1", "compiler", "target_repair_1"])
|
|
self.assertEqual(result[0]["minimax_h3_prompt"], "")
|
|
self.assertFalse(result[-1]["ready_for_generation"])
|
|
self.assertEqual(
|
|
result[-1]["blocked_reasons"][:2],
|
|
["visual_grounding_unverified", "visual_grounding_telemetry_error"],
|
|
)
|
|
self.assertEqual(result[-1]["grounding_guard"]["model_call_count"], 3)
|
|
unload.assert_called_once_with(runtime)
|
|
|
|
def test_call_budget_counts_failed_backend_attempts(self) -> None:
|
|
budget = self.nodes.BackendCallBudget(max_calls=4)
|
|
options = self.nodes.BackendRunOptions(
|
|
enable_telemetry=False,
|
|
call_budget=budget,
|
|
)
|
|
with patch.object(
|
|
self.nodes, "_run_backend_detailed", side_effect=RuntimeError("boom")
|
|
) as backend:
|
|
for _ in range(4):
|
|
with self.assertRaisesRegex(RuntimeError, "boom"):
|
|
self.nodes._run_backend_for_packet(
|
|
self.runtime(),
|
|
"prompt",
|
|
None,
|
|
128,
|
|
None,
|
|
options,
|
|
[],
|
|
)
|
|
with self.assertRaisesRegex(RuntimeError, r"4-call ceiling"):
|
|
self.nodes._run_backend_for_packet(
|
|
self.runtime(),
|
|
"prompt",
|
|
None,
|
|
128,
|
|
None,
|
|
options,
|
|
[],
|
|
)
|
|
|
|
self.assertEqual(budget.attempted_calls, 4)
|
|
self.assertEqual(backend.call_count, 4)
|
|
|
|
def test_compiler_retry_uses_the_next_adjacent_effective_seed(self) -> None:
|
|
budget = self.nodes.BackendCallBudget(max_calls=4)
|
|
budget.claim() # Evidence attempt 1 already consumed the first call.
|
|
effective: list[tuple[str, int]] = []
|
|
|
|
def backend(_config, _prompt, _media, _tokens, options, **_kwargs):
|
|
effective.append((options.stage, options.seed))
|
|
return self.nodes.BackendRunResult(decoded_text="{}")
|
|
|
|
results: list = []
|
|
with patch.object(self.nodes, "_run_backend_detailed", side_effect=backend):
|
|
for stage in ("compiler", "compiler_retry"):
|
|
options = self.nodes.BackendRunOptions(
|
|
seed=1009,
|
|
stage=stage,
|
|
enable_telemetry=False,
|
|
call_budget=budget,
|
|
)
|
|
self.nodes._run_backend_for_packet(
|
|
self.runtime(),
|
|
"prompt",
|
|
None,
|
|
128,
|
|
None,
|
|
options,
|
|
results,
|
|
)
|
|
|
|
self.assertEqual(effective, [("compiler", 1010), ("compiler_retry", 1011)])
|
|
self.assertEqual(budget.attempted_calls, 3)
|
|
|
|
def test_guarded_transformers_call_restores_cpu_rng_state(self) -> None:
|
|
nodes = self.nodes
|
|
|
|
class FakeProcessor:
|
|
tokenizer = None
|
|
image_token_id = 111
|
|
video_token_id = 222
|
|
|
|
def apply_chat_template(self, *_args, **_kwargs):
|
|
return {
|
|
"input_ids": torch.tensor([[7, 8]], dtype=torch.long),
|
|
"attention_mask": torch.ones((1, 2), dtype=torch.long),
|
|
}
|
|
|
|
def batch_decode(self, _sequences, **_kwargs):
|
|
return ["deterministic output"]
|
|
|
|
class FakeModel:
|
|
device = torch.device("cpu")
|
|
config = SimpleNamespace(
|
|
image_token_id=111,
|
|
video_token_id=None,
|
|
canvas_length=256,
|
|
)
|
|
|
|
def __init__(self):
|
|
self.random_draw = None
|
|
|
|
def generate(self, **kwargs):
|
|
self.random_draw = int(torch.randint(0, 10000, (1,)).item())
|
|
generated = torch.tensor([[self.random_draw % 100]], dtype=torch.long)
|
|
return SimpleNamespace(
|
|
sequences=torch.cat([kwargs["input_ids"], generated], dim=-1),
|
|
tokens_per_forward=torch.tensor([1]),
|
|
)
|
|
|
|
model = FakeModel()
|
|
options = nodes.BackendRunOptions(
|
|
sampling_profile="checkpoint_defaults",
|
|
thinking_mode="off",
|
|
seed=987654321,
|
|
enable_telemetry=False,
|
|
asset_registry=nodes.build_asset_registry({}, 0),
|
|
)
|
|
torch.manual_seed(12345)
|
|
state_before = torch.random.get_rng_state().clone()
|
|
with patch.object(
|
|
nodes, "_load_transformers_model", return_value=(FakeProcessor(), model)
|
|
), patch.object(
|
|
nodes,
|
|
"_run_blocking_with_progress",
|
|
side_effect=lambda _label, function, **_kwargs: function(),
|
|
), patch.object(torch.cuda, "is_available", return_value=False), patch.object(
|
|
torch.cuda, "device_count", return_value=0
|
|
):
|
|
result = nodes._run_transformers_detailed(
|
|
self.runtime(),
|
|
"prompt",
|
|
None,
|
|
128,
|
|
options,
|
|
)
|
|
|
|
self.assertEqual(result.decoded_text, "deterministic output")
|
|
self.assertIsNotNone(model.random_draw)
|
|
self.assertTrue(torch.equal(state_before, torch.random.get_rng_state()))
|
|
|
|
def test_passive_telemetry_does_not_change_tokens_for_the_same_seed(self) -> None:
|
|
nodes = self.nodes
|
|
|
|
class FakeProcessor:
|
|
tokenizer = None
|
|
image_token_id = 111
|
|
video_token_id = 222
|
|
|
|
def apply_chat_template(self, *_args, **_kwargs):
|
|
return {
|
|
"input_ids": torch.tensor([[7, 8]], dtype=torch.long),
|
|
"attention_mask": torch.ones((1, 2), dtype=torch.long),
|
|
}
|
|
|
|
def batch_decode(self, sequences, **_kwargs):
|
|
return [" ".join(str(int(token)) for token in sequences[0].tolist())]
|
|
|
|
class FakeModel:
|
|
device = torch.device("cpu")
|
|
config = SimpleNamespace(
|
|
image_token_id=111,
|
|
video_token_id=None,
|
|
canvas_length=256,
|
|
)
|
|
|
|
def __init__(self):
|
|
self.generated_token = None
|
|
|
|
def generate(self, **kwargs):
|
|
scores = torch.rand((1, 256, 17), dtype=torch.float32)
|
|
processors = kwargs.get("logits_processor")
|
|
if processors:
|
|
observed = processors[0](kwargs["input_ids"], scores, cur_step=1)
|
|
if observed is not scores:
|
|
raise AssertionError("telemetry replaced the scores tensor")
|
|
token = scores[:, 0, :].argmax(dim=-1, keepdim=True)
|
|
self.generated_token = int(token[0, 0])
|
|
streamer = kwargs.get("streamer")
|
|
if streamer is not None:
|
|
streamer.put(kwargs["input_ids"])
|
|
streamer.put_draft(token)
|
|
streamer.put(token)
|
|
streamer.end()
|
|
return SimpleNamespace(
|
|
sequences=torch.cat([kwargs["input_ids"], token], dim=-1),
|
|
tokens_per_forward=torch.tensor([1]),
|
|
)
|
|
|
|
def run(enable_telemetry: bool):
|
|
model = FakeModel()
|
|
options = nodes.BackendRunOptions(
|
|
sampling_profile="checkpoint_defaults",
|
|
thinking_mode="off",
|
|
seed=424242,
|
|
enable_telemetry=enable_telemetry,
|
|
fail_closed_telemetry=enable_telemetry,
|
|
asset_registry=nodes.build_asset_registry({}, 0),
|
|
)
|
|
with patch.object(
|
|
nodes, "_load_transformers_model", return_value=(FakeProcessor(), model)
|
|
), patch.object(
|
|
nodes,
|
|
"_run_blocking_with_progress",
|
|
side_effect=lambda _label, function, **_kwargs: function(),
|
|
), patch.object(torch.cuda, "is_available", return_value=False), patch.object(
|
|
torch.cuda, "device_count", return_value=0
|
|
):
|
|
result = nodes._run_transformers_detailed(
|
|
self.runtime(), "prompt", None, 128, options
|
|
)
|
|
return model.generated_token, result
|
|
|
|
torch.manual_seed(9988)
|
|
state_before = torch.random.get_rng_state().clone()
|
|
token_without, result_without = run(False)
|
|
token_with, result_with = run(True)
|
|
self.assertEqual(token_without, token_with)
|
|
self.assertEqual(result_without.decoded_text, result_with.decoded_text)
|
|
self.assertFalse(result_without.telemetry)
|
|
self.assertEqual(result_with.telemetry["forward_count"], 1)
|
|
self.assertTrue(torch.equal(state_before, torch.random.get_rng_state()))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|