Files
olivv-cs-ComfyUI-FunPack/tests/test_v2_category_alignment.py
T
digital-garbageandClaude Sonnet 4.6 f5f2fdc0a4 Add transition-based prompt splitting and auto window sizing
Refiner V2 and Studio: new split_by_transitions option detects
transition words in the final prompt (then, suddenly, cut to, etc.),
extracts the character description (text before the first comma),
and encodes one conditioning per scene window with the character
description prepended to each segment. Returns a multi-entry
conditioning list that routes per-window via split_conds_to_windows.

Context Transition Windows: new total_frames optional input.
When > 0 and split_conds_to_windows is enabled, context_length
is auto-computed at sampling time as total_frames / num_cond_entries
so windows size themselves to match Refiner's segment count.

Also fixes pre-existing bugs: early error return in refine_v2 was
missing video_latent (7th return value), and test unpacking was stale.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-20 15:02:30 +03:00

2322 lines
83 KiB
Python

import json
import sys
import types
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
sys.modules.setdefault("folder_paths", types.SimpleNamespace(models_dir=""))
import torch
from conditioning import (
FunPackVideoRefinerV2,
V2_RATING_LABELS,
normalize_refiner_v2_rating,
tensor_to_serializable,
)
class FakeClip:
def tokenize(self, text):
return text
def encode_from_tokens_scheduled(self, tokens):
return [(torch.ones(1, 4, 3), {"pooled_output": torch.ones(1, 3)})]
class CountingClip(FakeClip):
def __init__(self):
self.calls = 0
def encode_from_tokens_scheduled(self, tokens):
self.calls += 1
return super().encode_from_tokens_scheduled(tokens)
class GeneratingClip(FakeClip):
def __init__(self, generated_text):
self.generated_text = generated_text
self.tokenize_calls = []
self.generate_kwargs = {}
def tokenize(self, text, **kwargs):
self.tokenize_calls.append((text, kwargs))
return text
def generate(self, tokens, **kwargs):
self.generate_kwargs = kwargs
return [1, 2, 3]
def decode(self, token_ids, skip_special_tokens=True):
return self.generated_text
def primary_category(phrase):
refiner = FunPackVideoRefinerV2()
scores = refiner._v2_heuristic_scores(phrase)
primary, confidence = refiner._v2_scores_primary(scores)
return primary, confidence, scores
def classified_phrase(refiner, text):
return refiner._v2_classify_phrases(
None,
[{"text": text, "tokens": refiner._v2_phrase_words(text)}],
)[0]
def train_phrase(refiner, text, rating_label, global_state=None, previous_missing_axes=None, iterations=1):
global_state = global_state or {"phrase_memory": {}}
phrase = classified_phrase(refiner, text)
last_run = {
"prompt": text,
"phrases": [phrase],
}
profile = normalize_refiner_v2_rating(rating_label)
feedback = refiner._v2_axis_feedback(profile, previous_missing_axes)
for iteration in range(iterations):
refiner._v2_update_phrase_memory(global_state, last_run, profile, iteration + 1, feedback)
return global_state["phrase_memory"][text.lower()], global_state, feedback
def prompt_items(refiner, words):
return [
{"text": word, "tokens": refiner._v2_phrase_words(word)}
for word in words
]
def prompt_phrases(refiner, prompt, global_state=None):
return refiner._v2_classify_phrases(
None,
refiner._ordered_prompt_phrases(prompt),
global_state or {"phrase_memory": {}},
)
def train_prompt_context(refiner, words, rating_label, global_state=None, iterations=1):
global_state = global_state or {"phrase_memory": {}}
phrases = refiner._v2_classify_phrases(None, prompt_items(refiner, words), global_state)
last_run = {
"prompt": " ".join(words),
"phrases": phrases,
}
profile = normalize_refiner_v2_rating(rating_label)
feedback = refiner._v2_axis_feedback(profile, None)
for iteration in range(iterations):
refiner._v2_update_phrase_memory(global_state, last_run, profile, iteration + 1, feedback)
return global_state, feedback
def test_environment_descriptions_do_not_become_actions():
primary, confidence, scores = primary_category("detailed background")
assert primary == "environment"
assert confidence >= 0.60
assert scores["action"] == 0.0
def test_weathered_environment_descriptions_do_not_become_actions():
primary, confidence, scores = primary_category("weathered stone room")
assert primary == "environment"
assert confidence >= 0.60
assert scores["action"] == 0.0
def test_appearance_descriptions_stay_appearance():
primary, confidence, scores = primary_category("flowing hair and blue eyes")
assert primary == "appearance"
assert confidence >= 0.60
assert scores["action"] == 0.0
def test_clothing_descriptions_stay_appearance():
primary, confidence, scores = primary_category("wearing red dress")
assert primary == "appearance"
assert confidence >= 0.60
assert scores["action"] == 0.0
def test_motion_descriptions_stay_action():
primary, confidence, scores = primary_category("running through the street")
assert primary == "action"
assert confidence >= 0.60
assert scores["action"] >= scores["environment"]
def test_axis_feedback_treats_unmentioned_axes_as_satisfied():
refiner = FunPackVideoRefinerV2()
profile = normalize_refiner_v2_rating("Missing quality")
feedback = refiner._v2_axis_feedback(profile, ["details", "action"])
assert feedback["missing_axes"] == ["quality"]
assert feedback["satisfied_axes"] == ["details", "action"]
assert feedback["resolved_axes"] == ["details", "action"]
assert feedback["regressed_axes"] == ["quality"]
def test_axis_feedback_does_not_infer_regression_without_previous_rating():
refiner = FunPackVideoRefinerV2()
profile = normalize_refiner_v2_rating("Missing quality")
feedback = refiner._v2_axis_feedback(profile, None)
assert feedback["missing_axes"] == ["quality"]
assert feedback["satisfied_axes"] == ["details", "action"]
assert feedback["resolved_axes"] == []
assert feedback["regressed_axes"] == []
def test_conditioning_memory_records_missing_and_satisfied_axes():
refiner = FunPackVideoRefinerV2()
global_state = {"axis_conditioning_memory": {}}
quality_good = tensor_to_serializable(torch.ones(1, 3, 2))
action_good = tensor_to_serializable(torch.zeros(1, 3, 2))
first_profile = normalize_refiner_v2_rating("Missing details + action")
first_feedback = refiner._v2_axis_feedback(first_profile, None)
refiner._v2_update_conditioning_memory(
global_state,
{"conditioning": quality_good},
first_profile,
first_feedback,
)
second_profile = normalize_refiner_v2_rating("Missing quality")
second_feedback = refiner._v2_axis_feedback(second_profile, ["details", "action"])
refiner._v2_update_conditioning_memory(
global_state,
{"conditioning": action_good},
second_profile,
second_feedback,
)
memory = global_state["axis_conditioning_memory"]
assert memory["quality"]["positive"]["count"] == 1
assert memory["quality"]["negative"]["count"] == 1
assert memory["action"]["positive"]["count"] >= 1
assert memory["details"]["positive"]["count"] >= 1
assert memory["action"]["negative"]["count"] == 1
assert memory["details"]["negative"]["count"] == 1
def test_awful_lora_feedback_reduces_before_missing_axis_boosts():
refiner = FunPackVideoRefinerV2()
prompt_history = {}
global_state = {"lora_weight_memory": {}}
profile = normalize_refiner_v2_rating("Awful")
feedback = refiner._v2_axis_feedback(profile, [])
refiner._v2_update_lora_suggestions(
{
"loras": [
{
"id": "motion",
"name": "motion_lora.safetensors",
"type": "action",
"base_model_weight": 1.0,
}
]
},
prompt_history,
global_state,
[{"text": "running", "primary": "action"}],
profile,
feedback,
)
suggestion = prompt_history["lora_weight_suggestions"]["motion"]
assert suggestion["model_weight"] < 1.0
def test_category_weights_are_recorded_for_every_learning_rating():
refiner = FunPackVideoRefinerV2()
learning_labels = [
label for label in V2_RATING_LABELS
if label not in {"-Just forget it-", "Wrong appearance"}
]
for label in learning_labels:
entry, _, _ = train_phrase(refiner, f"running test {label}", label)
assert entry["category_evidence_count"] == 1
assert set(entry["category_weights"]) == set(refiner.CATEGORY_DESCRIPTIONS)
assert set(entry["clip_heuristic_scores"]) == set(refiner.CATEGORY_DESCRIPTIONS)
assert set(entry["effective_category_scores"]) == set(refiner.CATEGORY_DESCRIPTIONS)
def test_wrong_appearance_rating_is_available():
profile = normalize_refiner_v2_rating("Wrong appearance")
assert "Wrong appearance" in V2_RATING_LABELS
assert profile["key"] == "wrong_appearance"
assert profile["wrong_categories"] == ["appearance", "subject", "environment"]
def test_prompt_repair_blocks_appearance_subject_and_environment():
refiner = FunPackVideoRefinerV2()
global_state = {
"phrase_memory": {
"white tights": {
"text": "white tights",
"primary": "appearance",
"effective_category_scores": refiner._v2_heuristic_scores("white tights"),
"wanted_axes": {"details": 3},
"score": 6.0,
"liked_count": 4,
},
"detailed background": {
"text": "detailed background",
"primary": "environment",
"effective_category_scores": refiner._v2_heuristic_scores("detailed background"),
"wanted_axes": {"details": 3},
"score": 6.0,
"liked_count": 4,
},
"female character": {
"text": "female character",
"primary": "subject",
"effective_category_scores": refiner._v2_heuristic_scores("female character"),
"wanted_axes": {"details": 3},
"score": 6.0,
"liked_count": 4,
},
"tiny smoke curls": {
"text": "tiny smoke curls",
"primary": "details",
"effective_category_scores": refiner._v2_heuristic_scores("tiny smoke curls"),
"wanted_axes": {"details": 3},
"score": 4.0,
"liked_count": 2,
},
}
}
profile = normalize_refiner_v2_rating("Missing details")
feedback = refiner._v2_axis_feedback(profile, None)
repaired, _, candidates = refiner._v2_repair_prompt_for_missing_axes(
"person smoking",
refiner._v2_classify_phrases(None, prompt_items(refiner, ["person smoking"]), global_state),
global_state,
None,
feedback,
)
assert "tiny smoke curls" in repaired
assert "white tights" not in repaired
assert "detailed background" not in repaired
assert "female character" not in repaired
assert all(candidate["text"] != "white tights" for candidate in candidates)
def test_scene_builder_user_category_overrides_refiner_classification():
refiner = FunPackVideoRefinerV2()
scene_db = {
"universal_memory": {
"running pose": {
"text": "running pose",
"category": "appearance",
"category_source": "user",
"category_locked": True,
}
}
}
phrase = refiner._v2_classify_phrases(
None,
[{"text": "running pose", "tokens": refiner._v2_phrase_words("running pose")}],
{"phrase_memory": {}},
scene_db=scene_db,
)[0]
assert phrase["primary"] == "appearance"
assert phrase["source"] == "scene_builder_user"
assert phrase["scene_category_locked"] is True
def test_scene_builder_locked_category_is_not_trained_by_refiner():
refiner = FunPackVideoRefinerV2()
scene_db = {
"universal_memory": {
"running pose": {
"text": "running pose",
"category": "appearance",
"category_source": "user",
"category_locked": True,
}
}
}
global_state = {"phrase_memory": {}}
phrase = refiner._v2_classify_phrases(
None,
[{"text": "running pose", "tokens": refiner._v2_phrase_words("running pose")}],
global_state,
scene_db=scene_db,
)[0]
refiner._v2_update_phrase_memory(
global_state,
{"prompt": "running pose", "phrases": [phrase]},
normalize_refiner_v2_rating("Perfect"),
1,
refiner._v2_axis_feedback(normalize_refiner_v2_rating("Perfect"), None),
)
entry = global_state["phrase_memory"]["running pose"]
assert entry["primary"] == "appearance"
assert entry["category_locked"] is True
assert entry["category_evidence_count"] == 0
assert entry["liked_count"] == 0
def test_scene_builder_wildcard_cleanup_runs_after_refiner_processing():
refiner = FunPackVideoRefinerV2()
scene_db = {
"universal_memory": {
"red dress": {"text": "red dress", "category": "appearance", "wildcard": True},
"blue dress": {"text": "blue dress", "category": "appearance", "wildcard": True},
}
}
prompt, status = refiner._v2_resolve_scene_builder_wildcards(
"person walking, red dress, blue dress",
scene_db,
)
assert prompt == "person walking, red dress"
assert "removed 1 duplicate" in status
def test_lucky_composition_gates_unrelated_old_action_by_current_intent():
refiner = FunPackVideoRefinerV2()
global_state = {
"phrase_memory": {
"dancing in rain": {
"text": "dancing in rain",
"primary": "action",
"effective_category_scores": refiner._v2_heuristic_scores("dancing in rain"),
"score": 6.0,
"liked_count": 6,
"wanted_axes": {"action": 4},
},
"walking on beach": {
"text": "walking on beach",
"primary": "action",
"effective_category_scores": refiner._v2_heuristic_scores("walking on beach"),
"score": 2.0,
"liked_count": 1,
"wanted_axes": {"action": 1},
},
}
}
lucky, status = refiner._v2_compose_lucky_prompt(
"walking on beach",
prompt_phrases(refiner, "walking on beach", global_state),
global_state,
intent_prompt="walking on beach",
)
assert "walking on beach" in lucky
assert "dancing in rain" not in lucky
assert "Lucky: on" in status
def test_perfect_repair_does_not_inject_after_scene_change():
refiner = FunPackVideoRefinerV2()
slot = {
"perfect_repairs": {
"dancing in rain": {"text": "dancing in rain", "axes": {"action": 1}, "count": 3}
}
}
repaired, status, adjustments = refiner._v2_apply_perfect_repair_phrases(
"walking on beach",
slot,
intent_prompt="walking on beach",
intent_phrases=prompt_phrases(refiner, "walking on beach"),
)
assert repaired == "walking on beach"
assert adjustments == []
assert "already represented" in status
def test_intent_alignment_learns_missing_original_intent_from_enhancer_variant():
refiner = FunPackVideoRefinerV2()
global_state = {"phrase_memory": {}, "intent_alignment_memory": {}}
intent_prompt = "woman walking through neon rain"
enhanced_prompt = "woman smiling, cinematic studio portrait"
profile = normalize_refiner_v2_rating("Missing action")
feedback = refiner._v2_axis_feedback(profile, None)
status = refiner._v2_update_intent_alignment_memory(
global_state,
{
"prompt": enhanced_prompt,
"phrases": prompt_phrases(refiner, enhanced_prompt, global_state),
"intent_prompt": intent_prompt,
"intent_phrases": prompt_phrases(refiner, intent_prompt, global_state),
},
profile,
1,
feedback,
)
slot = next(iter(global_state["intent_alignment_memory"].values()))
missing = slot["missing_intent_phrases"]["woman walking through neon rain"]
variant = next(iter(slot["variants"].values()))
assert "Intent alignment learned" in status
assert missing["score"] > 0.5
assert missing["missing_count"] == 1
assert variant["missing_intent_count"] == 1
def test_intent_alignment_restores_learned_missing_original_phrase():
refiner = FunPackVideoRefinerV2()
global_state = {"phrase_memory": {}, "intent_alignment_memory": {}}
intent_prompt = "woman walking through neon rain"
profile = normalize_refiner_v2_rating("Missing action")
feedback = refiner._v2_axis_feedback(profile, None)
refiner._v2_update_intent_alignment_memory(
global_state,
{
"prompt": "woman smiling, cinematic studio portrait",
"phrases": prompt_phrases(refiner, "woman smiling, cinematic studio portrait", global_state),
"intent_prompt": intent_prompt,
"intent_phrases": prompt_phrases(refiner, intent_prompt, global_state),
},
profile,
1,
feedback,
)
aligned, status, adjustments = refiner._v2_apply_intent_alignment_memory(
"woman smiling, moody closeup",
prompt_phrases(refiner, "woman smiling, moody closeup", global_state),
intent_prompt,
prompt_phrases(refiner, intent_prompt, global_state),
global_state,
)
assert "woman walking through neon rain" in aligned
assert "restored 1 original phrase" in status
assert adjustments == [
{"text": "woman walking through neon rain", "source": "intent_missing", "action": "added"}
]
def test_intent_alignment_removes_rejected_enhancer_only_extra():
refiner = FunPackVideoRefinerV2()
global_state = {"phrase_memory": {}, "intent_alignment_memory": {}}
intent_prompt = "woman walking through neon rain"
enhanced_prompt = "woman walking through neon rain, white tights"
profile = normalize_refiner_v2_rating("Wrong appearance")
feedback = refiner._v2_axis_feedback(profile, None)
refiner._v2_update_intent_alignment_memory(
global_state,
{
"prompt": enhanced_prompt,
"phrases": prompt_phrases(refiner, enhanced_prompt, global_state),
"intent_prompt": intent_prompt,
"intent_phrases": prompt_phrases(refiner, intent_prompt, global_state),
},
profile,
1,
feedback,
)
aligned, status, adjustments = refiner._v2_apply_intent_alignment_memory(
enhanced_prompt,
prompt_phrases(refiner, enhanced_prompt, global_state),
intent_prompt,
prompt_phrases(refiner, intent_prompt, global_state),
global_state,
)
assert "white tights" not in aligned
assert "woman walking through neon rain" in aligned
assert "removed 1 rejected enhancer-only phrase" in status
assert adjustments == [
{"text": "white tights", "source": "enhancer_extra", "action": "removed"}
]
def test_intent_alignment_stores_pairs_and_bad_tokens_to_omit():
refiner = FunPackVideoRefinerV2()
global_state = {"phrase_memory": {}, "intent_alignment_memory": {}}
intent_prompt = "woman walking through neon rain"
enhanced_prompt = "woman walking through neon rain, white tights"
profile = normalize_refiner_v2_rating("Wrong appearance")
feedback = refiner._v2_axis_feedback(profile, None)
refiner._v2_update_intent_alignment_memory(
global_state,
{
"prompt": enhanced_prompt,
"phrases": prompt_phrases(refiner, enhanced_prompt, global_state),
"intent_prompt": intent_prompt,
"intent_phrases": prompt_phrases(refiner, intent_prompt, global_state),
},
profile,
1,
feedback,
)
slot = next(iter(global_state["intent_alignment_memory"].values()))
pair = next(iter(slot["intent_enhance_pairs"].values()))
assert pair["intent_prompt"] == intent_prompt
assert pair["positive_prompt"] == enhanced_prompt
assert "woman" in slot["provided_tokens"]
assert slot["provided_tokens"]["woman"]["omit"] is False
assert slot["provided_tokens"]["white"]["omit"] is True
assert slot["provided_tokens"]["tights"]["omit"] is True
assert slot["provided_tokens"]["white tights"]["kind"] == "pair"
assert slot["provided_tokens"]["white tights"]["omit"] is True
assert set(slot["bad_tokens"]) >= {"white", "tights", "white tights"}
def test_intent_alignment_omits_bad_token_in_new_enhancer_phrase():
refiner = FunPackVideoRefinerV2()
global_state = {"phrase_memory": {}, "intent_alignment_memory": {}}
intent_prompt = "woman walking through neon rain"
profile = normalize_refiner_v2_rating("Wrong appearance")
feedback = refiner._v2_axis_feedback(profile, None)
refiner._v2_update_intent_alignment_memory(
global_state,
{
"prompt": "woman walking through neon rain, white tights",
"phrases": prompt_phrases(refiner, "woman walking through neon rain, white tights", global_state),
"intent_prompt": intent_prompt,
"intent_phrases": prompt_phrases(refiner, intent_prompt, global_state),
},
profile,
1,
feedback,
)
aligned, _, adjustments = refiner._v2_apply_intent_alignment_memory(
"woman walking through neon rain, red tights",
prompt_phrases(refiner, "woman walking through neon rain, red tights", global_state),
intent_prompt,
prompt_phrases(refiner, intent_prompt, global_state),
global_state,
)
assert "red tights" not in aligned
assert "woman walking through neon rain" in aligned
assert adjustments == [
{"text": "red tights", "source": "enhancer_extra", "action": "removed"}
]
def test_repeated_intent_family_phrase_becomes_conservative_repair_candidate():
refiner = FunPackVideoRefinerV2()
global_state = {
"phrase_memory": {},
"intent_family_memory": {},
"perfect_anchors": {},
"variant_evidence": {},
"intent_preference_phrases": {},
"conditioning_deltas": {},
}
intent_prompt = "yellow car riding down the road"
enhanced_prompt = "yellow car in a desert"
profile = normalize_refiner_v2_rating("Missing action")
feedback = refiner._v2_axis_feedback(profile, None)
for iteration in range(2):
refiner._v2_update_intent_family_memory(
global_state,
{
"prompt": enhanced_prompt,
"phrases": prompt_phrases(refiner, enhanced_prompt, global_state),
"intent_prompt": intent_prompt,
"intent_phrases": prompt_phrases(refiner, intent_prompt, global_state),
},
profile,
iteration + 1,
feedback,
)
current_intent = "yellow car down the road"
_, family_slot, _ = refiner._v2_intent_family_slot(global_state, current_intent, create=False)
repaired, status, candidates = refiner._v2_repair_prompt_for_missing_axes(
enhanced_prompt,
prompt_phrases(refiner, enhanced_prompt, global_state),
global_state,
None,
feedback,
intent_phrases=prompt_phrases(refiner, current_intent, global_state),
intent_family_slot=family_slot,
)
assert "yellow car riding down the road" in repaired
assert "intent_preference" in status
assert any(candidate["source"] == "intent_preference" for candidate in candidates)
def test_repeated_intent_preference_does_not_repair_unrelated_family():
refiner = FunPackVideoRefinerV2()
global_state = {
"phrase_memory": {},
"intent_family_memory": {},
"perfect_anchors": {},
"variant_evidence": {},
"intent_preference_phrases": {},
"conditioning_deltas": {},
}
yellow_intent = "yellow car riding down the road"
dragon_intent = "dragon sleeping in a cave"
profile = normalize_refiner_v2_rating("Missing action")
feedback = refiner._v2_axis_feedback(profile, None)
for iteration in range(2):
refiner._v2_update_intent_family_memory(
global_state,
{
"prompt": "yellow car in a desert",
"phrases": prompt_phrases(refiner, "yellow car in a desert", global_state),
"intent_prompt": yellow_intent,
"intent_phrases": prompt_phrases(refiner, yellow_intent, global_state),
},
profile,
iteration + 1,
feedback,
)
_, dragon_slot, _ = refiner._v2_intent_family_slot(global_state, dragon_intent, create=True)
repaired, _, candidates = refiner._v2_repair_prompt_for_missing_axes(
"dragon in a cave",
prompt_phrases(refiner, "dragon in a cave", global_state),
global_state,
None,
feedback,
intent_phrases=prompt_phrases(refiner, dragon_intent, global_state),
intent_family_slot=dragon_slot,
)
assert "yellow car riding down the road" not in repaired
assert all(candidate["text"] != "yellow car riding down the road" for candidate in candidates)
def test_active_repair_axes_persist_until_perfect():
refiner = FunPackVideoRefinerV2()
global_state = {"active_repair_axes": []}
missing_action = normalize_refiner_v2_rating("Missing action")
action_feedback = refiner._v2_axis_feedback(missing_action, None)
repair_feedback, status = refiner._v2_active_repair_feedback(global_state, action_feedback, missing_action)
assert repair_feedback["missing_axes"] == ["action"]
assert global_state["active_repair_axes"] == ["action"]
assert "active until Perfect" in status
missing_quality = normalize_refiner_v2_rating("Missing quality")
quality_feedback = refiner._v2_axis_feedback(missing_quality, ["action"])
repair_feedback, _ = refiner._v2_active_repair_feedback(global_state, quality_feedback, missing_quality)
assert repair_feedback["missing_axes"] == ["action", "quality"]
assert global_state["active_repair_axes"] == ["action", "quality"]
perfect = normalize_refiner_v2_rating("Perfect")
perfect_feedback = refiner._v2_axis_feedback(perfect, ["action", "quality"])
repair_feedback, status = refiner._v2_active_repair_feedback(global_state, perfect_feedback, perfect)
assert repair_feedback["missing_axes"] == []
assert global_state["active_repair_axes"] == []
assert "cleared by Perfect" in status
def test_perfect_rating_preserves_successful_repair_phrases():
refiner = FunPackVideoRefinerV2()
global_state = {
"phrase_memory": {},
"intent_family_memory": {},
"perfect_anchors": {},
"variant_evidence": {},
"intent_preference_phrases": {},
"conditioning_deltas": {},
}
intent_prompt = "person smoking"
profile = normalize_refiner_v2_rating("Perfect")
feedback = refiner._v2_axis_feedback(profile, None)
source = tensor_to_serializable(torch.zeros(1, 3, 2))
refined = tensor_to_serializable(torch.ones(1, 3, 2))
refiner._v2_update_intent_family_memory(
global_state,
{
"prompt": "person smoking",
"encoded_prompt": "person smoking, tiny smoke curls",
"phrases": prompt_phrases(refiner, "person smoking", global_state),
"intent_prompt": intent_prompt,
"intent_phrases": prompt_phrases(refiner, intent_prompt, global_state),
"source_conditioning": source,
"conditioning": refined,
"repair_candidates": [{"text": "tiny smoke curls", "axes": ["details"], "score": 2.0, "source": "memory"}],
},
profile,
1,
feedback,
)
_, slot, _ = refiner._v2_intent_family_slot(global_state, intent_prompt, create=False)
repaired, status, adjustments = refiner._v2_apply_perfect_repair_phrases("person smoking", slot)
assert "tiny smoke curls" in repaired
assert "Perfect-proven" in status
assert adjustments == [{"text": "tiny smoke curls", "source": "perfect_repair", "action": "added"}]
def test_prompt_repair_falls_back_to_any_axis_candidates_when_requested_axis_is_empty():
refiner = FunPackVideoRefinerV2()
global_state = {
"phrase_memory": {
"portrait quality": {
"text": "portrait quality",
"primary": "quality",
"effective_category_scores": refiner._v2_heuristic_scores("portrait quality"),
"wanted_axes": {"quality": 3},
"score": 4.0,
"liked_count": 2,
"category_evidence_count": 2,
}
}
}
profile = normalize_refiner_v2_rating("Missing action")
feedback = refiner._v2_axis_feedback(profile, None)
repaired, status, candidates = refiner._v2_repair_prompt_for_missing_axes(
"static portrait",
prompt_phrases(refiner, "static portrait", global_state),
global_state,
None,
feedback,
)
assert "portrait quality" in repaired
assert "showing other-axis repair candidates" in status
assert candidates[0]["text"] == "portrait quality"
assert candidates[0]["source"] == "memory"
def test_intent_family_perfect_anchor_keeps_loved_variant():
refiner = FunPackVideoRefinerV2()
global_state = {
"phrase_memory": {},
"intent_family_memory": {},
"perfect_anchors": {},
"variant_evidence": {},
"intent_preference_phrases": {},
"conditioning_deltas": {},
}
intent_prompt = "yellow car riding down the road"
profile = normalize_refiner_v2_rating("Perfect")
feedback = refiner._v2_axis_feedback(profile, None)
source = tensor_to_serializable(torch.zeros(1, 3, 2))
first = tensor_to_serializable(torch.ones(1, 3, 2))
second = tensor_to_serializable(torch.ones(1, 3, 2) * 2.0)
refiner._v2_update_intent_family_memory(
global_state,
{
"prompt": "yellow car riding down the road",
"encoded_prompt": "yellow car riding down the road",
"phrases": prompt_phrases(refiner, "yellow car riding down the road", global_state),
"intent_prompt": intent_prompt,
"intent_phrases": prompt_phrases(refiner, intent_prompt, global_state),
"source_conditioning": source,
"conditioning": first,
},
profile,
1,
feedback,
)
refiner._v2_update_intent_family_memory(
global_state,
{
"prompt": "yellow car riding down the road, camera focused on wheels",
"encoded_prompt": "yellow car riding down the road, camera focused on wheels",
"phrases": prompt_phrases(refiner, "yellow car riding down the road, camera focused on wheels", global_state),
"intent_prompt": intent_prompt,
"intent_phrases": prompt_phrases(refiner, intent_prompt, global_state),
"source_conditioning": source,
"conditioning": second,
},
profile,
2,
feedback,
)
_, slot, _ = refiner._v2_intent_family_slot(global_state, intent_prompt, create=False)
assert slot["perfect_anchors"]["base"]["positive_prompt"] == "yellow car riding down the road"
assert len(slot["loved_variants"]) == 1
assert global_state["perfect_anchors"][slot["family_key"]]["base"]["positive_prompt"] == "yellow car riding down the road"
assert slot["conditioning_deltas"]["positive"]["count"] == 2
def test_pre_perfect_missing_intent_learning_is_conservative():
refiner = FunPackVideoRefinerV2()
global_state = {"phrase_memory": {}, "intent_alignment_memory": {}, "intent_family_memory": {}}
intent_prompt = "woman walking through neon rain"
profile = normalize_refiner_v2_rating("Missing action")
feedback = refiner._v2_axis_feedback(profile, None)
refiner._v2_update_intent_family_memory(
global_state,
{
"prompt": "cinematic studio portrait",
"phrases": prompt_phrases(refiner, "cinematic studio portrait", global_state),
"intent_prompt": intent_prompt,
"intent_phrases": prompt_phrases(refiner, intent_prompt, global_state),
},
profile,
1,
feedback,
)
refiner._v2_update_intent_alignment_memory(
global_state,
{
"prompt": "cinematic studio portrait",
"phrases": prompt_phrases(refiner, "cinematic studio portrait", global_state),
"intent_prompt": intent_prompt,
"intent_phrases": prompt_phrases(refiner, intent_prompt, global_state),
},
profile,
1,
feedback,
)
slot = next(iter(global_state["intent_alignment_memory"].values()))
missing = slot["missing_intent_phrases"]["woman walking through neon rain"]
assert 0.5 <= missing["score"] < 0.8
def test_negative_repair_never_adds_current_intent_phrase():
refiner = FunPackVideoRefinerV2()
global_state = {"phrase_memory": {}, "negative_prompt_memory": {}, "intent_family_memory": {}}
intent_prompt = "Nicole is shooting"
enhanced_prompt = "Nicole from Zenless Zone Zero holding a gun and shooting"
profile = normalize_refiner_v2_rating("Wrong action")
feedback = refiner._v2_axis_feedback(profile, None)
intent_phrases = prompt_phrases(refiner, intent_prompt, global_state)
phrases = prompt_phrases(refiner, enhanced_prompt, global_state)
refiner._v2_update_intent_family_memory(
global_state,
{
"prompt": enhanced_prompt,
"phrases": phrases,
"intent_prompt": intent_prompt,
"intent_phrases": intent_phrases,
},
profile,
1,
feedback,
)
status = refiner._v2_update_negative_prompt_memory(
global_state,
{
"prompt": enhanced_prompt,
"phrases": phrases,
"intent_prompt": intent_prompt,
"intent_phrases": intent_phrases,
},
profile,
feedback,
)
_, family_slot, _ = refiner._v2_intent_family_slot(global_state, intent_prompt, create=False)
repaired, repair_status = refiner._v2_repair_negative_prompt(
"",
global_state,
feedback,
current_prompt=enhanced_prompt,
intent_prompt=intent_prompt,
intent_phrases=intent_phrases,
intent_family_slot=family_slot,
)
assert repaired == ""
assert "Skipped 1 intent-locked" in status
assert "intent/current" in repair_status or "no stored poor-rated tags" in repair_status
def test_negative_repair_blocks_requested_action_overlap_without_exact_phrase():
refiner = FunPackVideoRefinerV2()
global_state = {
"negative_prompt_memory": {
"tags": {
"holding a gun and shooting": {
"text": "holding a gun and shooting",
"count": 4,
"axes": {"action": 2},
"last_seen_iter": 1,
}
}
}
}
profile = normalize_refiner_v2_rating("Missing action")
feedback = refiner._v2_axis_feedback(profile, None)
repaired, status = refiner._v2_repair_negative_prompt(
"",
global_state,
feedback,
current_prompt="Nicole is shooting",
)
assert repaired == ""
assert "intent/current 1" in status
def test_negative_memory_skips_partial_requested_action_with_explicit_intent():
refiner = FunPackVideoRefinerV2()
global_state = {"phrase_memory": {}, "negative_prompt_memory": {}, "intent_family_memory": {}}
profile = normalize_refiner_v2_rating("Wrong action")
feedback = refiner._v2_axis_feedback(profile, None)
intent_prompt = "Nicole is shooting"
status = refiner._v2_update_negative_prompt_memory(
global_state,
{
"prompt": "Nicole from Zenless Zone Zero holding a gun and shooting",
"phrases": prompt_phrases(refiner, "holding a gun and shooting", global_state),
"intent_prompt": intent_prompt,
"intent_phrases": prompt_phrases(refiner, intent_prompt, global_state),
},
profile,
feedback,
)
assert global_state["negative_prompt_memory"]["tags"] == {}
assert "Skipped 1 intent-locked" in status
def test_one_word_intent_matches_expanded_enhancer_phrase():
refiner = FunPackVideoRefinerV2()
global_state = {"phrase_memory": {}, "intent_alignment_memory": {}, "intent_family_memory": {}}
intent_prompt = "shooting"
enhanced_prompt = "Nicole from Zenless Zone Zero holding a gun and shooting"
profile = normalize_refiner_v2_rating("Wrong action")
feedback = refiner._v2_axis_feedback(profile, None)
refiner._v2_update_intent_alignment_memory(
global_state,
{
"prompt": enhanced_prompt,
"phrases": prompt_phrases(refiner, enhanced_prompt, global_state),
"intent_prompt": intent_prompt,
"intent_phrases": prompt_phrases(refiner, intent_prompt, global_state),
},
profile,
1,
feedback,
)
slot = next(iter(global_state["intent_alignment_memory"].values()))
assert slot["extra_positive_phrases"] == {}
assert slot["bad_tokens"] == {}
assert refiner._v2_phrase_texts_match(enhanced_prompt, intent_prompt)
def test_one_word_intent_blocks_expanded_action_negative_repair():
refiner = FunPackVideoRefinerV2()
global_state = {
"negative_prompt_memory": {
"tags": {
"holding a gun and shooting": {
"text": "holding a gun and shooting",
"count": 4,
"axes": {"action": 2},
"last_seen_iter": 1,
}
}
}
}
profile = normalize_refiner_v2_rating("Missing action")
feedback = refiner._v2_axis_feedback(profile, None)
repaired, status = refiner._v2_repair_negative_prompt(
"",
global_state,
feedback,
current_prompt="Nicole holding a gun and shooting",
intent_prompt="shooting",
intent_phrases=prompt_phrases(refiner, "shooting", global_state),
)
assert repaired == ""
assert "intent/current 1" in status
def test_short_semantic_intent_locks_enhancer_expansion_without_shared_word():
refiner = FunPackVideoRefinerV2()
global_state = {"phrase_memory": {}, "intent_alignment_memory": {}, "intent_family_memory": {}}
intent_prompt = "defenstration"
enhanced_prompt = "a human is thrown out of window"
intent_phrases = prompt_phrases(refiner, intent_prompt, global_state)
phrases = prompt_phrases(refiner, enhanced_prompt, global_state)
refiner._v2_mark_semantic_intent_locks(
FakeClip(),
phrases,
intent_prompt,
intent_phrases,
encode_cache={},
)
profile = normalize_refiner_v2_rating("Wrong action")
feedback = refiner._v2_axis_feedback(profile, None)
refiner._v2_update_intent_alignment_memory(
global_state,
{
"prompt": enhanced_prompt,
"phrases": phrases,
"intent_prompt": intent_prompt,
"intent_phrases": intent_phrases,
},
profile,
1,
feedback,
)
slot = next(iter(global_state["intent_alignment_memory"].values()))
assert phrases[0]["semantic_intent_locked"] is True
assert slot["missing_intent_phrases"] == {}
assert slot["extra_positive_phrases"] == {}
assert slot["bad_tokens"] == {}
def test_short_semantic_intent_blocks_expansion_from_negative_memory():
refiner = FunPackVideoRefinerV2()
global_state = {"phrase_memory": {}, "negative_prompt_memory": {}, "intent_family_memory": {}}
intent_prompt = "defenstration"
enhanced_prompt = "a human is thrown out of window"
intent_phrases = prompt_phrases(refiner, intent_prompt, global_state)
phrases = prompt_phrases(refiner, enhanced_prompt, global_state)
refiner._v2_mark_semantic_intent_locks(
FakeClip(),
phrases,
intent_prompt,
intent_phrases,
encode_cache={},
)
profile = normalize_refiner_v2_rating("Wrong action")
feedback = refiner._v2_axis_feedback(profile, None)
status = refiner._v2_update_negative_prompt_memory(
global_state,
{
"prompt": enhanced_prompt,
"phrases": phrases,
"intent_prompt": intent_prompt,
"intent_phrases": intent_phrases,
},
profile,
feedback,
)
assert global_state["negative_prompt_memory"]["tags"] == {}
assert "Skipped 1 intent-locked" in status
def test_rejected_repair_candidate_can_be_penalized_as_negative_memory():
refiner = FunPackVideoRefinerV2()
global_state = {"phrase_memory": {}, "negative_prompt_memory": {}, "intent_family_memory": {}}
profile = normalize_refiner_v2_rating("Wrong details")
feedback = refiner._v2_axis_feedback(profile, None)
refiner._v2_update_negative_prompt_memory(
global_state,
{
"prompt": "person smoking",
"phrases": prompt_phrases(refiner, "person smoking", global_state),
"intent_prompt": "person smoking",
"intent_phrases": prompt_phrases(refiner, "person smoking", global_state),
"repair_candidates": [{"text": "tiny smoke curls", "axes": ["details"], "score": 2.0, "source": "memory"}],
},
profile,
feedback,
)
assert "tiny smoke curls" in global_state["negative_prompt_memory"]["tags"]
assert global_state["negative_prompt_memory"]["tags"]["tiny smoke curls"]["source"] == "repair_candidate"
def test_intent_family_delta_ignores_incompatible_shapes_and_caps_strength():
refiner = FunPackVideoRefinerV2()
conditioning = torch.ones(1, 3, 2)
slot = {
"conditioning_deltas": {
"positive": {
"count": 1,
"delta": tensor_to_serializable(torch.ones(1, 4, 2)),
}
}
}
unchanged, status = refiner._v2_apply_intent_family_delta(conditioning, slot, 0.05)
assert torch.equal(unchanged, conditioning)
assert status == "intent-family idle"
slot["conditioning_deltas"]["positive"]["delta"] = tensor_to_serializable(torch.ones(1, 3, 2) * 100.0)
changed, status = refiner._v2_apply_intent_family_delta(conditioning, slot, 0.05)
assert status.startswith("intent-family positive delta")
assert torch.max(torch.abs(changed - conditioning)).item() < 1.0
def test_lucky_skips_appearance_memory_unless_prompt_mentions_it():
refiner = FunPackVideoRefinerV2()
global_state = {"phrase_memory": {}, "preferred_context_memory": {}}
phrases = refiner._v2_classify_phrases(
None,
prompt_items(refiner, ["white tights", "walking"]),
global_state,
)
profile = normalize_refiner_v2_rating("Perfect")
feedback = refiner._v2_axis_feedback(profile, None)
refiner._v2_update_phrase_memory(
global_state,
{"prompt": "white tights walking", "phrases": phrases},
profile,
1,
feedback,
)
lucky_prompt, _ = refiner._v2_compose_lucky_prompt("", [], global_state)
explicit_prompt, _ = refiner._v2_compose_lucky_prompt(
"white tights",
refiner._v2_classify_phrases(None, prompt_items(refiner, ["white tights"]), global_state),
global_state,
)
assert "walking" in lucky_prompt
assert "white tights" not in lucky_prompt
assert "tights" not in lucky_prompt
assert "white tights" in explicit_prompt
def test_wrong_appearance_suppresses_only_auto_inserted_appearance_memory():
refiner = FunPackVideoRefinerV2()
global_state = {"phrase_memory": {}, "preferred_context_memory": {}}
phrases = refiner._v2_classify_phrases(None, prompt_items(refiner, ["walking"]), global_state)
profile = normalize_refiner_v2_rating("Wrong appearance")
feedback = refiner._v2_axis_feedback(profile, None)
refiner._v2_update_phrase_memory(
global_state,
{
"prompt": "walking",
"encoded_prompt": "walking, white tights",
"phrases": phrases,
"repair_candidates": [{"text": "white tights", "axes": ["details"], "score": 3.0, "source": "memory"}],
},
profile,
1,
feedback,
)
appearance = global_state["phrase_memory"]["white tights"]
motion = global_state["phrase_memory"]["walking"]
assert appearance["auto_inject_suppressed"] is True
assert appearance["wrong_appearance_count"] >= 1
assert motion.get("wrong_appearance_count", 0) == 0
assert motion["category_evidence_count"] == 0
def test_repeated_wrong_appearance_keeps_entry_out_of_lucky():
refiner = FunPackVideoRefinerV2()
global_state = {"phrase_memory": {}, "preferred_context_memory": {}}
phrases = refiner._v2_classify_phrases(None, prompt_items(refiner, ["walking"]), global_state)
profile = normalize_refiner_v2_rating("Wrong appearance")
feedback = refiner._v2_axis_feedback(profile, None)
for iteration in range(2):
refiner._v2_update_phrase_memory(
global_state,
{
"prompt": "walking",
"encoded_prompt": "walking, white tights",
"phrases": phrases,
"repair_candidates": [{"text": "white tights", "axes": ["details"], "score": 3.0, "source": "memory"}],
},
profile,
iteration + 1,
feedback,
)
global_state["phrase_memory"]["white tights"]["score"] = 6.0
global_state["phrase_memory"]["white tights"]["liked_count"] = 8
lucky_prompt, _ = refiner._v2_compose_lucky_prompt("", [], global_state)
assert global_state["phrase_memory"]["white tights"]["auto_inject_blocked_count"] >= 2
assert "white tights" not in lucky_prompt
def test_forget_rating_skips_category_weight_learning():
refiner = FunPackVideoRefinerV2()
global_state = {"phrase_memory": {}}
phrase = classified_phrase(refiner, "running")
profile = normalize_refiner_v2_rating("-Just forget it-")
status = refiner._v2_update_phrase_memory(
global_state,
{"prompt": "running", "phrases": [phrase]},
profile,
1,
refiner._v2_axis_feedback(profile, None),
)
assert status == "Lucky memory: no learning update."
assert global_state["phrase_memory"] == {}
def test_missing_details_trains_micro_movement_as_detail_without_losing_action_context():
refiner = FunPackVideoRefinerV2()
entry, global_state, _ = train_phrase(
refiner,
"patting head",
"Missing details",
iterations=5,
)
assert entry["category_weights"]["details"] > entry["category_weights"]["action"]
assert entry["category_weights"]["action"] > 0.0
learned = refiner._v2_classify_phrases(
None,
[{"text": "patting head", "tokens": ["patting", "head"]}],
global_state,
)[0]
assert learned["primary"] == "details"
assert "details" in refiner._v2_axes_for_scores(learned["effective_category_scores"])
def test_missing_quality_trains_quality_while_preserving_satisfied_motion():
refiner = FunPackVideoRefinerV2()
entry, _, feedback = train_phrase(
refiner,
"running",
"Missing quality",
previous_missing_axes=["details", "action"],
)
assert feedback["missing_axes"] == ["quality"]
assert feedback["resolved_axes"] == ["details", "action"]
assert entry["category_weights"]["quality"] > 0.0
assert entry["category_weights"]["action"] > 0.0
assert entry["resolved_axes"]["action"] == 1
def test_paired_missing_ratings_train_both_requested_axes():
refiner = FunPackVideoRefinerV2()
details_action, _, _ = train_phrase(refiner, "hugging", "Missing details + action")
details_quality, _, _ = train_phrase(refiner, "hands in frame", "Missing details + quality")
action_quality, _, _ = train_phrase(refiner, "smoking", "Missing action + quality")
assert details_action["category_weights"]["details"] > 0.0
assert details_action["category_weights"]["action"] > 0.0
assert details_quality["category_weights"]["details"] > 0.0
assert details_quality["category_weights"]["quality"] > 0.0
assert action_quality["category_weights"]["action"] > 0.0
assert action_quality["category_weights"]["quality"] > 0.0
def test_repeated_user_feedback_can_override_initial_machine_category():
refiner = FunPackVideoRefinerV2()
machine_primary, _, _ = primary_category("detailed background")
assert machine_primary == "environment"
entry, global_state, _ = train_phrase(
refiner,
"detailed background",
"Missing action",
iterations=9,
)
learned = refiner._v2_classify_phrases(
None,
[{"text": "detailed background", "tokens": ["detailed", "background"]}],
global_state,
)[0]
assert entry["category_weights"]["action"] > entry["category_weights"]["environment"]
assert learned["machine_primary"] == "environment"
assert learned["primary"] == "action"
assert learned["source"] == "rating_weighted"
def test_old_phrase_memory_entries_receive_category_weight_defaults():
refiner = FunPackVideoRefinerV2()
memory = {
"running": {
"text": "running",
"primary": "action",
"category_scores": {"action": 0.7},
}
}
entry = refiner._v2_ensure_phrase_memory_entry(memory, "running")
assert set(entry["category_weights"]) == set(refiner.CATEGORY_DESCRIPTIONS)
assert set(entry["clip_heuristic_scores"]) == set(refiner.CATEGORY_DESCRIPTIONS)
assert set(entry["effective_category_scores"]) == set(refiner.CATEGORY_DESCRIPTIONS)
def test_context_senses_keep_polysemous_token_categories_separate():
refiner = FunPackVideoRefinerV2()
global_state = {"phrase_memory": {}}
train_prompt_context(refiner, ["smoke", "rising", "lips"], "Missing action", global_state, iterations=8)
train_prompt_context(refiner, ["smoke", "makeup", "eyes"], "Missing details", global_state, iterations=8)
action_smoke = refiner._v2_classify_phrases(
None,
prompt_items(refiner, ["smoke", "rising", "lips"]),
global_state,
)[0]
detail_smoke = refiner._v2_classify_phrases(
None,
prompt_items(refiner, ["smoke", "makeup", "eyes"]),
global_state,
)[0]
assert action_smoke["primary"] == "action"
assert detail_smoke["primary"] == "details"
assert action_smoke["context_source"] == "exact"
assert detail_smoke["context_source"] == "exact"
assert len(global_state["phrase_memory"]["smoke"]["context_senses"]) >= 2
def test_context_senses_are_pruned_by_evidence_and_recency():
refiner = FunPackVideoRefinerV2()
entry = {"context_senses": {}}
for index in range(30):
entry["context_senses"][f"mid|context-{index}"] = {
"category_evidence_count": index,
"occurrence_count": index,
"last_seen_iter": index,
}
refiner._v2_prune_context_senses(entry, limit=24)
assert len(entry["context_senses"]) == 24
assert "mid|context-29" in entry["context_senses"]
assert "mid|context-0" not in entry["context_senses"]
def test_training_diagnostics_explain_learning_state_and_guidance():
refiner = FunPackVideoRefinerV2()
phrases = refiner._v2_classify_phrases(None, prompt_items(refiner, ["smoke", "rising"]))
profile = normalize_refiner_v2_rating("Missing action")
feedback = refiner._v2_axis_feedback(profile, None)
first_run_guidance = refiner._v2_training_guidance(
False,
profile,
feedback,
phrases,
"Category memory trained: 0 concept unit(s).",
"LoRA suggestions: no FunPack LoRA stack connected.",
)
diagnostics = refiner._v2_category_diagnostics(phrases)
assert "first V2 run only seeds" in first_run_guidance
assert "Category diagnostics:" in diagnostics
assert "smoke:" in diagnostics
assert "ctx=" in diagnostics
def test_refiner_training_info_uses_readable_sections(tmp_path):
refiner = FunPackVideoRefinerV2()
state_path = tmp_path / "state.json"
refiner._v2_state_path = lambda refinement_key: str(state_path)
_, _, training_info, _, _, _, _ = refiner.refine_v2(
"woman walking through neon rain",
FakeClip(),
"Perfect",
"readable-test",
)
assert "\n\nMemory\n" in training_info
assert "\n\nPrompt Analysis\n" in training_info
assert "\n\nAdaptation\n" in training_info
assert "\n\nLoRA\n" in training_info
def test_refiner_v2_exposes_clip_and_conditioning_as_optional_inputs():
inputs = FunPackVideoRefinerV2.INPUT_TYPES()
assert "clip" not in inputs["required"]
assert "positive_conditioning" not in inputs["required"]
assert inputs["required"]["mode"][0] == ["Refine", "Prompt only", "Learning"]
assert inputs["required"]["advisor_mode"][0] == ["Off", "Only diagnostics", "Only prompt", "Full"]
assert "encoded_prompts" in FunPackVideoRefinerV2.RETURN_NAMES
assert "model" in FunPackVideoRefinerV2.RETURN_NAMES
assert FunPackVideoRefinerV2.RETURN_TYPES[FunPackVideoRefinerV2.RETURN_NAMES.index("encoded_prompts")] == "STRING"
assert "clip" in inputs["optional"]
assert "advisor_clip" in inputs["optional"]
assert "positive_conditioning" in inputs["optional"]
assert inputs["optional"]["advisor_thinking"][1]["default"] is True
def test_refiner_v2_advisor_uses_explicit_system_prompt_previous_prompt_thinking_and_image():
refiner = FunPackVideoRefinerV2()
image = torch.zeros(1, 8, 8, 3)
clip = GeneratingClip(
"DIAGNOSTIC: add clearer smoke motion.\n"
"REPAIRED_PROMPT: person smoking, smoke trails drifting upward"
)
prompt, status, diagnostic, applied, _ = refiner._v2_prompt_advisor(
clip,
"Diagnostics",
"person smoking",
"person smoking",
{"missing_axes": ["details"], "wrong_axes": []},
[],
previous_run={"prompt": "old prompt", "encoded_prompt": "old encoded prompt"},
image=image,
thinking=True,
seed=123,
)
advisor_prompt, kwargs = clip.tokenize_calls[0]
assert prompt == "person smoking"
assert applied is False
assert "diagnostics only" in status
assert diagnostic == "add clearer smoke motion."
assert "Prompt: person smoking" in advisor_prompt
assert "Previous prompt: old encoded prompt" in advisor_prompt
assert kwargs["image"] is image
assert kwargs["thinking"] is True
assert clip.generate_kwargs["seed"] == 123
def test_refiner_v2_advisor_repair_applies_validated_generated_prompt(tmp_path):
refiner = FunPackVideoRefinerV2()
state_path = tmp_path / "state.json"
state_path.write_text(json.dumps({
"version": 2,
"refinement_key": "advisor-repair-test",
"state_namespace": "clip",
"global": {
"total_iterations": 1,
"avg_reward_ema": 0.0,
"good_streak": 0,
"bad_streak": 1,
"last_rating_label": "Missing action",
"last_missing_axes": ["action"],
"phrase_memory": {},
"axis_conditioning_memory": {},
"lora_weight_memory": {},
"preferred_context_memory": {},
"intent_alignment_memory": {},
"intent_family_memory": {},
"perfect_anchors": {},
"variant_evidence": {},
"intent_preference_phrases": {},
"conditioning_deltas": {},
"active_repair_axes": [],
"negative_prompt_memory": {},
"vision_memory": {},
"loss_history": [],
},
"prompt_histories": {},
"last_run": {
"prompt": "person smoking",
"encoded_prompt": "person smoking",
"source_conditioning": tensor_to_serializable(torch.zeros(1, 4, 3)),
"conditioning": tensor_to_serializable(torch.zeros(1, 4, 3)),
"phrases": prompt_phrases(refiner, "person smoking"),
"rating_label": "Unrated",
"iteration": 1,
},
}), encoding="utf-8")
refiner._v2_state_path = lambda refinement_key: str(state_path)
clip = GeneratingClip(
"DIAGNOSTIC: add visible smoke motion.\n"
"REPAIRED_PROMPT: person smoking, smoke trails drifting upward"
)
_, status, training_info, _, encoded_prompts, _, _ = refiner.refine_v2(
"person smoking",
clip,
"Missing details",
"advisor-repair-test",
user_intent_prompt="person smoking",
advisor_mode="Repair prompt",
advisor_thinking=True,
)
state = json.loads(state_path.read_text(encoding="utf-8"))
assert "Advisor: applied repair" in status
assert "Encoded as: advisor repaired prompt" in training_info
assert "Positive prompt: person smoking, smoke trails drifting upward" in encoded_prompts
assert state["last_run"]["encoded_prompt"] == "person smoking, smoke trails drifting upward"
assert state["last_run"]["advisor"]["applied"] is True
def test_refiner_v2_advisor_uses_separate_advisor_clip_when_connected(tmp_path):
refiner = FunPackVideoRefinerV2()
state_path = tmp_path / "state.json"
state_path.write_text(json.dumps({
"version": 2,
"refinement_key": "advisor-clip-test",
"state_namespace": "clip",
"global": {
"total_iterations": 1,
"avg_reward_ema": 0.0,
"good_streak": 0,
"bad_streak": 1,
"last_rating_label": "Missing action",
"last_missing_axes": ["action"],
"phrase_memory": {},
"axis_conditioning_memory": {},
"lora_weight_memory": {},
"preferred_context_memory": {},
"intent_alignment_memory": {},
"intent_family_memory": {},
"perfect_anchors": {},
"variant_evidence": {},
"intent_preference_phrases": {},
"conditioning_deltas": {},
"active_repair_axes": [],
"negative_prompt_memory": {},
"vision_memory": {},
"loss_history": [],
},
"prompt_histories": {},
"last_run": {
"prompt": "person smoking",
"encoded_prompt": "person smoking",
"source_conditioning": tensor_to_serializable(torch.zeros(1, 4, 3)),
"conditioning": tensor_to_serializable(torch.zeros(1, 4, 3)),
"phrases": prompt_phrases(refiner, "person smoking"),
"rating_label": "Unrated",
"iteration": 1,
},
}), encoding="utf-8")
refiner._v2_state_path = lambda refinement_key: str(state_path)
main_clip = CountingClip()
advisor_clip = GeneratingClip(
"DIAGNOSTIC: add visible smoke motion.\n"
"REPAIRED_PROMPT: person smoking, smoke trails drifting upward"
)
refiner.refine_v2(
"person smoking",
main_clip,
"Missing details",
"advisor-clip-test",
user_intent_prompt="person smoking",
advisor_mode="Repair prompt",
advisor_clip=advisor_clip,
)
state = json.loads(state_path.read_text(encoding="utf-8"))
assert advisor_clip.tokenize_calls
assert state["last_run"]["encoded_prompt"] == "person smoking, smoke trails drifting upward"
assert state["last_run"]["advisor"]["applied"] is True
def test_refiner_v2_advisor_skips_when_no_generation_clip_is_available(tmp_path):
refiner = FunPackVideoRefinerV2()
state_path = tmp_path / "state.json"
state_path.write_text(json.dumps({
"version": 2,
"refinement_key": "advisor-no-generation-test",
"state_namespace": "clip",
"global": {
"total_iterations": 1,
"avg_reward_ema": 0.0,
"good_streak": 0,
"bad_streak": 1,
"last_rating_label": "Missing action",
"last_missing_axes": ["action"],
"phrase_memory": {},
"axis_conditioning_memory": {},
"lora_weight_memory": {},
"preferred_context_memory": {},
"intent_alignment_memory": {},
"intent_family_memory": {},
"perfect_anchors": {},
"variant_evidence": {},
"intent_preference_phrases": {},
"conditioning_deltas": {},
"active_repair_axes": [],
"negative_prompt_memory": {},
"vision_memory": {},
"loss_history": [],
},
"prompt_histories": {},
"last_run": {
"prompt": "person smoking",
"encoded_prompt": "person smoking",
"source_conditioning": tensor_to_serializable(torch.zeros(1, 4, 3)),
"conditioning": tensor_to_serializable(torch.zeros(1, 4, 3)),
"phrases": prompt_phrases(refiner, "person smoking"),
"rating_label": "Unrated",
"iteration": 1,
},
}), encoding="utf-8")
refiner._v2_state_path = lambda refinement_key: str(state_path)
_, status, training_info, _, _, _, _ = refiner.refine_v2(
"person smoking",
FakeClip(),
"Missing details",
"advisor-no-generation-test",
advisor_mode="Repair prompt",
)
state = json.loads(state_path.read_text(encoding="utf-8"))
assert "Advisor: unavailable; connected CLIP does not expose text generation" in status
assert "Advisor: unavailable; connected CLIP does not expose text generation" in training_info
assert state["last_run"]["encoded_prompt"] == "person smoking"
assert state["last_run"]["advisor"]["applied"] is False
def test_refiner_v2_learning_mode_passes_prompt_and_conditioning_through(tmp_path):
refiner = FunPackVideoRefinerV2()
state_path = tmp_path / "state.json"
state_path.write_text(json.dumps({
"version": 2,
"refinement_key": "learning-mode-test",
"state_namespace": "clip",
"global": {
"total_iterations": 1,
"avg_reward_ema": 0.0,
"good_streak": 0,
"bad_streak": 1,
"last_rating_label": "Missing action",
"last_missing_axes": ["action"],
"phrase_memory": {
"tiny smoke curls": {
"text": "tiny smoke curls",
"primary": "details",
"effective_category_scores": refiner._v2_heuristic_scores("tiny smoke curls"),
"wanted_axes": {"details": 4},
"score": 8.0,
"liked_count": 6,
}
},
"axis_conditioning_memory": {},
"lora_weight_memory": {},
"preferred_context_memory": {},
"intent_alignment_memory": {},
"intent_family_memory": {},
"perfect_anchors": {},
"variant_evidence": {},
"intent_preference_phrases": {},
"conditioning_deltas": {},
"active_repair_axes": ["details"],
"negative_prompt_memory": {
"tags": {
"bad repaired detail": {
"text": "bad repaired detail",
"count": 3,
"axes": {"details": 2},
"last_seen_iter": 1,
}
}
},
"vision_memory": {},
"loss_history": [],
"liked_conditioning": tensor_to_serializable(torch.flip(torch.arange(12, dtype=torch.float32).reshape(1, 4, 3), dims=[-1])),
},
"prompt_histories": {},
"last_run": {
"prompt": "person smoking",
"encoded_prompt": "person smoking, tiny smoke curls",
"source_conditioning": tensor_to_serializable(torch.zeros(1, 4, 3)),
"conditioning": tensor_to_serializable(torch.ones(1, 4, 3)),
"phrases": [],
"rating_label": "Unrated",
"iteration": 1,
},
}), encoding="utf-8")
refiner._v2_state_path = lambda refinement_key: str(state_path)
positive_conditioning = [(torch.arange(12, dtype=torch.float32).reshape(1, 4, 3), {"pooled_output": torch.ones(1, 3)})]
modified, status, training_info, _, encoded_prompts, _, _ = refiner.refine_v2(
"person smoking",
None,
"Missing details",
"learning-mode-test",
positive_conditioning=positive_conditioning,
mode="Learning",
)
state = json.loads(state_path.read_text(encoding="utf-8"))
assert "Mode: Learning" in training_info
assert torch.equal(modified[0][0], positive_conditioning[0][0])
assert state["last_run"]["encoded_prompt"] == "person smoking"
assert "tiny smoke curls" not in state["last_run"]["encoded_prompt"]
def test_refiner_v2_accepts_conditioning_without_clip_and_loads_gemma3_tokenizer(tmp_path, monkeypatch):
refiner = FunPackVideoRefinerV2()
state_path = tmp_path / "state.json"
refiner._v2_state_path = lambda refinement_key: str(state_path)
tokenizer_modes = []
class FakeTokenizer:
name_or_path = "DreamFast/gemma-3-12b-it-heretic-v2"
def fake_get_tokenizer(mode="ltx2"):
tokenizer_modes.append(mode)
return FakeTokenizer()
monkeypatch.setattr(refiner, "_get_tokenizer", fake_get_tokenizer)
positive_conditioning = [(torch.full((1, 4, 3), 2.0), {"pooled_output": torch.ones(1, 3)})]
modified, status, training_info, _, _, _, _ = refiner.refine_v2(
"woman walking through neon rain",
None,
"Perfect",
"conditioning-input-test",
positive_conditioning=positive_conditioning,
)
assert tokenizer_modes == ["ltx2"]
assert modified[0][0].shape == positive_conditioning[0][0].shape
def test_refiner_v2_prefers_clip_when_both_clip_and_conditioning_are_connected(tmp_path, monkeypatch):
refiner = FunPackVideoRefinerV2()
state_path = tmp_path / "state.json"
refiner._v2_state_path = lambda refinement_key: str(state_path)
monkeypatch.setattr(refiner, "_get_tokenizer", lambda mode="ltx2": (_ for _ in ()).throw(AssertionError("unexpected tokenizer load")))
positive_conditioning = [(torch.full((1, 4, 3), 9.0), {"pooled_output": torch.ones(1, 3)})]
modified, status, training_info, _, _, _, _ = refiner.refine_v2(
"woman walking through neon rain",
FakeClip(),
"Perfect",
"clip-priority-test",
positive_conditioning=positive_conditioning,
)
assert not torch.allclose(modified[0][0], torch.full((1, 4, 3), 9.0))
assert torch.allclose(modified[0][0], torch.ones(1, 4, 3))
def test_refiner_v2_errors_without_clip_or_conditioning(tmp_path):
refiner = FunPackVideoRefinerV2()
state_path = tmp_path / "state.json"
refiner._v2_state_path = lambda refinement_key: str(state_path)
modified, status, training_info, _, encoded_prompts, _, _ = refiner.refine_v2(
"woman walking through neon rain",
None,
"Perfect",
"missing-conditioning-test",
)
assert modified == []
assert "Positive prompt: woman walking through neon rain" in encoded_prompts
assert "ERROR: V2 could not prepare conditioning" in status
def test_prompt_enhancer_refusal_is_not_stored_as_last_run(tmp_path):
refiner = FunPackVideoRefinerV2()
state_path = tmp_path / "state.json"
refiner._v2_state_path = lambda refinement_key: str(state_path)
refiner.refine_v2(
"I'm sorry, I cannot help you with this request.",
FakeClip(),
"Perfect",
"refusal-test",
)
state = json.loads(state_path.read_text(encoding="utf-8"))
assert state["last_run"] is None
assert state["prompt_histories"] == {}
assert state["global"]["phrase_memory"] == {}
assert state["global"]["total_iterations"] == 0
def test_saved_refusal_last_run_does_not_train_memory(tmp_path):
state_path = tmp_path / "state.json"
state_path.write_text(json.dumps({
"version": 2,
"refinement_key": "refusal-test",
"state_namespace": "clip",
"global": {
"total_iterations": 1,
"avg_reward_ema": 0.0,
"good_streak": 0,
"bad_streak": 0,
"last_rating_label": "Initial discovery",
"last_missing_axes": [],
"phrase_memory": {},
"axis_conditioning_memory": {},
"lora_weight_memory": {},
"loss_history": [],
},
"prompt_histories": {},
"last_run": {
"prompt": "I'm sorry, I cannot help you with this request.",
"encoded_prompt": "I'm sorry, I cannot help you with this request.",
"conditioning": {},
"phrases": [{"text": "i'm sorry", "tokens": ["sorry"], "primary": "details"}],
"rating_label": "Unrated",
"iteration": 1,
},
}), encoding="utf-8")
refiner = FunPackVideoRefinerV2()
refiner._v2_state_path = lambda refinement_key: str(state_path)
refiner.refine_v2("wide cinematic shot", FakeClip(), "Perfect", "refusal-test")
state = json.loads(state_path.read_text(encoding="utf-8"))
assert state["global"]["phrase_memory"] == {}
assert state["last_run"]["prompt"] == "wide cinematic shot"
def test_normal_previous_run_can_train_before_current_refusal_is_discarded(tmp_path):
refiner = FunPackVideoRefinerV2()
state_path = tmp_path / "state.json"
refiner._v2_state_path = lambda refinement_key: str(state_path)
refiner.refine_v2("woman walking through neon rain", FakeClip(), "Perfect", "refusal-test")
refiner.refine_v2(
"Sorry, but I can't assist with that request.",
FakeClip(),
"Perfect",
"refusal-test",
)
state = json.loads(state_path.read_text(encoding="utf-8"))
assert "woman walking through neon rain" in state["global"]["phrase_memory"]
assert state["last_run"] is None
assert "sorry, but i can't assist with that request." not in state["prompt_histories"]
def test_liked_action_detail_context_clusters_repair_missing_axes():
refiner = FunPackVideoRefinerV2()
global_state = {"phrase_memory": {}, "preferred_context_memory": {}}
rich_prompt = "woman walking reaching hand tiny particles rain reflections"
liked_phrases = refiner._v2_classify_phrases(
None,
prompt_items(refiner, [
"woman",
"walking",
"reaching hand",
"tiny particles",
"rain reflections",
]),
global_state,
)
profile = normalize_refiner_v2_rating("Perfect")
feedback = refiner._v2_axis_feedback(profile, None)
refiner._v2_update_phrase_memory(
global_state,
{"prompt": rich_prompt, "phrases": liked_phrases},
profile,
1,
feedback,
)
assert "Preferred context stored" in refiner._v2_update_preferred_context_memory(
global_state,
{"phrases": liked_phrases},
profile,
2,
feedback,
)
missing_profile = normalize_refiner_v2_rating("Missing details + action")
missing_feedback = refiner._v2_axis_feedback(missing_profile, None)
current_phrases = refiner._v2_classify_phrases(
None,
prompt_items(refiner, ["woman", "rain"]),
global_state,
)
intent_phrases = refiner._v2_classify_phrases(
None,
refiner._ordered_prompt_phrases(rich_prompt),
global_state,
)
repaired, status, candidates = refiner._v2_repair_prompt_for_missing_axes(
"woman, rain",
current_phrases,
global_state,
None,
missing_feedback,
intent_phrases=intent_phrases,
)
assert "walking" in repaired
assert "tiny particles" in repaired
assert "intent" in status or "preferred_context" in status
assert len(candidates) >= 2
def test_prompt_repair_does_not_pull_unrequested_memory_for_any_missing_axis():
refiner = FunPackVideoRefinerV2()
global_state = {
"phrase_memory": {
"walking": {
"text": "walking",
"primary": "action",
"effective_category_scores": refiner._v2_heuristic_scores("walking"),
"wanted_axes": {"action": 5},
"score": 5.0,
"liked_count": 5,
},
"tiny particles": {
"text": "tiny particles",
"primary": "details",
"effective_category_scores": refiner._v2_heuristic_scores("tiny particles"),
"wanted_axes": {"details": 5},
"score": 5.0,
"liked_count": 5,
},
"cinematic lighting": {
"text": "cinematic lighting",
"primary": "style",
"effective_category_scores": refiner._v2_heuristic_scores("cinematic lighting"),
"wanted_axes": {"quality": 5},
"score": 5.0,
"liked_count": 5,
},
"glass reflections": {
"text": "glass reflections",
"primary": "details",
"effective_category_scores": refiner._v2_heuristic_scores("glass reflections"),
"wanted_axes": {"details": 2},
"score": 2.0,
"liked_count": 1,
},
},
}
profile = normalize_refiner_v2_rating("Awful")
feedback = refiner._v2_axis_feedback(profile, None)
current_phrases = refiner._v2_classify_phrases(
None,
prompt_items(refiner, ["woman", "glass"]),
global_state,
)
repaired, _, candidates = refiner._v2_repair_prompt_for_missing_axes(
"woman, glass",
current_phrases,
global_state,
None,
feedback,
)
candidate_texts = {candidate["text"] for candidate in candidates}
assert "glass reflections" in repaired
assert "walking" not in repaired
assert "tiny particles" not in repaired
assert "cinematic lighting" not in repaired
assert "walking" not in candidate_texts
assert "tiny particles" not in candidate_texts
assert "cinematic lighting" not in candidate_texts
def test_prompt_repair_treats_same_word_with_different_neighbours_as_different():
refiner = FunPackVideoRefinerV2()
global_state = {
"phrase_memory": {
"turning head": {
"text": "turning head",
"primary": "action",
"effective_category_scores": refiner._v2_heuristic_scores("turning head"),
"wanted_axes": {"action": 4},
"score": 4.0,
"liked_count": 4,
"context_senses": {
"mid|eyes,portrait": {
"context": {
"anchor_words": ["head", "turning"],
"context_words": ["portrait", "eyes"],
"position_bucket": "mid",
"window": 3,
},
"category_weights": refiner._v2_category_template(0.0),
"effective_category_scores": refiner._v2_heuristic_scores("turning head"),
"category_evidence_count": 4,
"occurrence_count": 4,
},
},
},
},
}
profile = normalize_refiner_v2_rating("Missing action")
feedback = refiner._v2_axis_feedback(profile, None)
current_phrases = refiner._v2_classify_phrases(
None,
prompt_items(refiner, ["car", "turning wheel", "street"]),
global_state,
)
repaired, _, candidates = refiner._v2_repair_prompt_for_missing_axes(
"car, turning wheel, street",
current_phrases,
global_state,
None,
feedback,
)
assert "turning head" not in repaired
assert all(candidate["text"] != "turning head" for candidate in candidates)
def test_vague_user_intent_lets_enhanced_positive_prompt_anchor_repair():
refiner = FunPackVideoRefinerV2()
global_state = {
"phrase_memory": {
"walking slowly": {
"text": "walking slowly",
"primary": "action",
"effective_category_scores": refiner._v2_heuristic_scores("walking slowly"),
"wanted_axes": {"action": 3},
"score": 3.0,
"liked_count": 3,
},
},
}
assert refiner._v2_user_intent_prompt_is_vague("Figure it out")
assert not refiner._v2_user_intent_prompt_is_vague("person walking")
profile = normalize_refiner_v2_rating("Missing action")
feedback = refiner._v2_axis_feedback(profile, None)
current_phrases = refiner._v2_classify_phrases(
None,
prompt_items(refiner, ["woman", "walking through glass hallway"]),
global_state,
)
repaired, _, candidates = refiner._v2_repair_prompt_for_missing_axes(
"woman, walking through glass hallway",
current_phrases,
global_state,
None,
feedback,
intent_phrases=[],
)
assert "walking slowly" in repaired
assert any(candidate["text"] == "walking slowly" for candidate in candidates)
def test_complex_user_intent_ranks_before_matching_preferred_memory():
refiner = FunPackVideoRefinerV2()
global_state = {
"phrase_memory": {},
"preferred_context_memory": {
"reach-memory": {
"anchor": "reaching hand slowly",
"axes": {"action": 4},
"context": {},
"phrases": ["reaching hand slowly"],
"score": 10.0,
"liked_count": 8,
},
},
}
profile = normalize_refiner_v2_rating("Missing action")
feedback = refiner._v2_axis_feedback(profile, None)
current_phrases = refiner._v2_classify_phrases(
None,
prompt_items(refiner, ["woman"]),
global_state,
)
intent_phrases = refiner._v2_classify_phrases(
None,
refiner._ordered_prompt_phrases("woman reaching hand toward glass"),
global_state,
)
repaired, _, candidates = refiner._v2_repair_prompt_for_missing_axes(
"woman",
current_phrases,
global_state,
None,
feedback,
intent_phrases=intent_phrases,
)
assert "reaching hand toward glass" in repaired
assert candidates[0]["source"] == "intent"
def test_phrase_clusters_train_more_strongly_than_ngrams_and_tokens():
refiner = FunPackVideoRefinerV2()
global_state = {"phrase_memory": {}, "preferred_context_memory": {}}
phrases = refiner._v2_classify_phrases(
None,
[{"text": "reaching hand slowly", "tokens": ["reaching", "hand", "slowly"]}],
global_state,
)
profile = normalize_refiner_v2_rating("Perfect")
feedback = refiner._v2_axis_feedback(profile, None)
refiner._v2_update_phrase_memory(
global_state,
{"prompt": "reaching hand slowly", "phrases": phrases},
profile,
1,
feedback,
)
phrase_score = global_state["phrase_memory"]["reaching hand slowly"]["score"]
ngram_score = global_state["phrase_memory"]["reaching hand"]["score"]
token_score = global_state["phrase_memory"]["reaching"]["score"]
assert phrase_score > ngram_score > token_score
def test_repair_candidates_saved_as_json_lists(tmp_path):
refiner = FunPackVideoRefinerV2()
state_path = tmp_path / "state.json"
refiner._v2_state_path = lambda refinement_key: str(state_path)
rich_prompt = "woman, walking, reaching hand, tiny particles, rain reflections"
refiner.refine_v2(rich_prompt, FakeClip(), "Perfect", "json-repair-test")
refiner.refine_v2(rich_prompt, FakeClip(), "Perfect", "json-repair-test")
refiner.refine_v2(
"woman, rain",
FakeClip(),
"Missing details + action",
"json-repair-test",
user_intent_prompt=rich_prompt,
)
state = json.loads(state_path.read_text(encoding="utf-8"))
candidates = state["last_run"]["repair_candidates"]
assert candidates
assert isinstance(candidates[0]["axes"], list)
assert "walking" in state["last_run"]["encoded_prompt"]
def test_wrong_action_rating_preserves_quality_but_marks_action_wrong():
refiner = FunPackVideoRefinerV2()
global_state = {"phrase_memory": {}, "preferred_context_memory": {}}
phrases = refiner._v2_classify_phrases(
None,
prompt_items(refiner, ["walking", "cinematic lighting"]),
global_state,
)
profile = normalize_refiner_v2_rating("Wrong action")
feedback = refiner._v2_axis_feedback(profile, None)
refiner._v2_update_phrase_memory(
global_state,
{"prompt": "walking cinematic lighting", "phrases": phrases},
profile,
1,
feedback,
)
assert feedback["missing_axes"] == ["action"]
assert feedback["wrong_axes"] == ["action"]
assert feedback["satisfied_axes"] == ["details", "quality"]
assert global_state["phrase_memory"]["walking"]["wrong_count"] == 1
assert global_state["phrase_memory"]["cinematic lighting"]["satisfied_count"] == 1
def test_ordered_prompt_phrases_preserve_stopwords_for_repair_text():
refiner = FunPackVideoRefinerV2()
phrases = refiner._ordered_prompt_phrases("running through the street, hands in the frame")
assert phrases[0]["text"] == "running through the street"
assert phrases[0]["tokens"] == ["running", "through", "street"]
assert phrases[1]["text"] == "hands in the frame"
def test_v2_image_metadata_detects_aspect_bucket_and_changed_fingerprint():
refiner = FunPackVideoRefinerV2()
first = torch.zeros(1, 32, 64, 3)
second = torch.ones(1, 32, 64, 3)
metadata, status = refiner._v2_image_metadata(first)
changed, _ = refiner._v2_image_metadata(second, metadata)
assert metadata["width"] == 64
assert metadata["height"] == 32
assert metadata["aspect_bucket"] == "ultrawide"
assert "64x32" in status
assert changed["changed_from_previous"] is True
def test_refiner_v2_caches_repeated_clip_category_encodes():
refiner = FunPackVideoRefinerV2()
clip = CountingClip()
phrases = [
{"text": "soft glow", "tokens": ["soft", "glow"]},
{"text": "soft glow", "tokens": ["soft", "glow"]},
]
refiner._v2_classify_phrases(clip, phrases, {"phrase_memory": {}}, encode_cache={})
assert clip.calls <= len(refiner.CATEGORY_DESCRIPTIONS) + 1
def test_split_by_transitions_returns_multi_entry_conditioning(tmp_path):
refiner = FunPackVideoRefinerV2()
state_path = tmp_path / "state.json"
refiner._v2_state_path = lambda refinement_key: str(state_path)
cond, status, _, _, _, _, _ = refiner.refine_v2(
"a woman in a red dress, then she runs, suddenly stops",
FakeClip(),
"Perfect",
"transition-test",
split_by_transitions=True,
)
assert len(cond) > 1, "Expected multiple conditioning entries from transition split"
assert "Transition split" in status
def test_split_by_transitions_disabled_returns_single_entry(tmp_path):
refiner = FunPackVideoRefinerV2()
state_path = tmp_path / "state.json"
refiner._v2_state_path = lambda refinement_key: str(state_path)
cond, status, _, _, _, _, _ = refiner.refine_v2(
"a woman in a red dress, then she runs, suddenly stops",
FakeClip(),
"Perfect",
"transition-test-off",
split_by_transitions=False,
)
assert len(cond) == 1, "Expected single conditioning entry when split_by_transitions is False"
assert "Transition split" not in status
def test_split_prompt_by_transitions_detects_known_phrases():
refiner = FunPackVideoRefinerV2()
segments = refiner._v2_split_prompt_by_transitions(
"a dog running in a field, then jumping over a fence, suddenly stopping"
)
assert len(segments) == 3
segments_none = refiner._v2_split_prompt_by_transitions("a dog running in a field")
assert len(segments_none) == 1
def test_split_char_from_scene_extracts_character_at_first_comma():
refiner = FunPackVideoRefinerV2()
char, scene = refiner._v2_split_char_from_scene("a woman in a red dress, walking in a park")
assert char == "a woman in a red dress"
assert scene == "walking in a park"
char_only, scene_empty = refiner._v2_split_char_from_scene("a woman in a red dress")
assert char_only == "a woman in a red dress"
assert scene_empty == ""