nodes/olivv-cs-ComfyUI-FunPack
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Stage 2 (2/3): delete dead prompt-repair methods + their tests

Browse Source 
Removes the now-unreferenced repair-only machinery left behind after
Stage 2 (1/3) neutralized the application:
  - _v2_prompt_repair_text_allowed
  - _v2_emphasized_prompt
  - _v2_repair_prompt_for_missing_axes and the repair-candidate helpers
  - perfect_repairs / rejected_repairs storage + perfect-repair eviction
    note in the intent-family learn path
Shared helpers still used by intent-alignment (e.g. _v2_repair_intent_roots,
_v2_repair_candidate_matches_intent, _v2_prompt_contains_text) are kept, as
is the repair_feedback / active_repair_axes "what's missing" signal the
advisor reads.

Deletes the 11 category-alignment tests that exercised the removed methods
directly. No new test failures (the 10/31 remaining are pre-existing env
failures from missing server/comfy modules).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
digital-garbage
2026-06-15 18:25:44 +03:00
co-authored by Claude Opus 4.8
parent 65816e353a
commit e39a64de57
2 changed files with 1 additions and 920 deletions
Download Patch File Download Diff File Expand all files Collapse all files
conditioning.py
+1 -430
View File
@@ -7307,13 +7307,6 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner):
item["source"] = "scene_builder_user"
return True
def _v2_prompt_repair_text_allowed(self, text):
if not self._v2_clean_phrase_text(text):
return False
if self._v2_item_has_blocked_auto_category(text):
return False
return self._v2_primary_category_for_text(text) in self.AUTO_INJECT_ALLOWED_CATEGORIES
def _v2_auto_inject_entry_allowed(self, entry, prompt=""):
text = str(entry.get("text", "") if isinstance(entry, dict) else entry).strip()
if not text:
@@ -8479,25 +8472,6 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner):
anchor_count = 0
delta_updates = 0
if axis_feedback and set(axis_feedback.get("wrong_axes", [])):
rejected_repairs = slot.setdefault("rejected_repairs", {})
for repair in repairs:
text = self._v2_clean_phrase_text(repair.get("text", ""))
if not text or self._v2_phrase_represented_by(text, intent_phrases):
continue
entry = rejected_repairs.setdefault(text, {"text": text, "count": 0, "last_seen_iter": 0})
entry["count"] = int(entry.get("count", 0)) + 1
entry["last_seen_iter"] = int(iter_num)
evicted_repairs = []
if rating_profile.get("key") in {"awful", "wrong_appearance"} and isinstance(slot.get("perfect_repairs"), dict):
evict_texts = {
self._v2_clean_phrase_text(item.get("text", ""))
for item in (last_run.get("intent_alignment_adjustments") or [])
if isinstance(item, dict) and item.get("source") == "perfect_repair"
}
for text in evict_texts:
if text and slot["perfect_repairs"].pop(text, None) is not None:
evicted_repairs.append(text)
if rating_profile.get("key") == "like" and isinstance(last_run.get("conditioning"), dict):
anchor_payload = {
"intent_prompt": source_intent,
@@ -8528,22 +8502,6 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner):
if self._v2_store_conditioning_delta_average(positive_delta, delta):
positive_delta["last_seen_iter"] = int(iter_num)
delta_updates += 1
perfect_repairs = slot.setdefault("perfect_repairs", {})
for repair in repairs:
text = self._v2_clean_phrase_text(repair.get("text", ""))
if not text:
continue
entry = perfect_repairs.setdefault(text, {
"text": text,
"axes": {},
"count": 0,
"last_seen_iter": 0,
})
entry["count"] = int(entry.get("count", 0)) + 1
entry["last_seen_iter"] = int(iter_num)
axes = entry.setdefault("axes", {})
for axis in self._v2_order_axes(set(repair.get("axes", []))):
axes[axis] = int(axes.get(axis, 0)) + 1
if len(variants) > 40:
slot["variant_evidence"] = dict(sorted(
@@ -8557,14 +8515,11 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner):
)[:40])
self._v2_sync_intent_family_aliases(global_state, family_key, slot)
eviction_note = (
f" Perfect-repair evicted: {', '.join(evicted_repairs)}." if evicted_repairs else ""
)
return (
"Intent family learned: "
f"family={family_key[:8]} sim={similarity:.2f}, repeated preference(s)={preference_updates}, "
f"anchors={anchor_count or len(slot.get('perfect_anchors', {})) + len(slot.get('loved_variants', {}))}, "
f"variant reward {variant['avg_reward']:+.2f}, delta update(s)={delta_updates}.{eviction_note}"
f"variant reward {variant['avg_reward']:+.2f}, delta update(s)={delta_updates}."
), family_key
def _v2_apply_intent_family_delta(self, conditioning, family_slot, strength):
@@ -10144,33 +10099,6 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner):
f" Total delta clamped to 7.5% of conditioning norm"
)
def _v2_emphasized_prompt(self, prompt, phrases, global_state, rating_profile):
missing_axes = set(global_state.get("last_missing_axes", [])) | set(rating_profile.get("missing_axes", []))
if not missing_axes:
return prompt, "Prompt emphasis: none."
candidates = []
for phrase in phrases:
phrase_axes = self._v2_axes_for_scores(
phrase.get("effective_category_scores", phrase.get("category_scores", {})) or
{phrase.get("primary", "details"): 1.0}
)
if phrase_axes & missing_axes:
candidates.append(phrase.get("text", ""))
candidates = [item for item in candidates if item]
if not candidates:
return prompt, "Prompt emphasis: no matching current phrases."
emphasized = []
seen = set()
for text in candidates:
key = text.lower()
if key in seen:
continue
seen.add(key)
emphasized.append(text)
if len(emphasized) >= 3:
break
return f"{prompt}, {', '.join(emphasized)}", f"Prompt emphasis: repeated {', '.join(emphasized)}."
def _v2_prompt_contains_text(self, prompt, text):
prompt_key = re.sub(r"\s+", " ", str(prompt or "").strip().lower())
text_key = re.sub(r"\s+", " ", str(text or "").strip().lower())
@@ -10225,363 +10153,6 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner):
scores = self._v2_heuristic_scores(text)
return len(words) <= 3 and not self._v2_axes_for_scores(scores)
def _v2_repair_reference_context(self, text, reference_phrases):
candidate_roots = self._v2_repair_intent_roots(text)
if not candidate_roots:
return {}
best_phrase = None
best_overlap = 0
for phrase in reference_phrases or []:
if not isinstance(phrase, dict):
continue
overlap = len(candidate_roots & self._v2_repair_intent_roots(phrase.get("text", "")))
if overlap > best_overlap:
best_phrase = phrase
best_overlap = overlap
if not isinstance(best_phrase, dict):
return {}
return best_phrase.get("context", {}) if isinstance(best_phrase.get("context", {}), dict) else {}
def _v2_repair_context_matches_memory(self, text, entry, reference_phrases):
if not isinstance(entry, dict):
return True
context = self._v2_repair_reference_context(text, reference_phrases)
context_words = context.get("context_words", []) if isinstance(context, dict) else []
senses = entry.get("context_senses", {})
if isinstance(senses, dict) and senses:
sense, similarity = self._v2_best_context_sense(entry, context)
if isinstance(sense, dict) and similarity >= 0.30:
return True
return len(self._v2_phrase_words(text)) > 1 and not context_words
learned_context = entry.get("context", {})
learned_words = learned_context.get("context_words", []) if isinstance(learned_context, dict) else []
if learned_words and context_words:
return self._v2_context_similarity(learned_context, context) >= 0.30
if learned_words and len(self._v2_phrase_words(text)) <= 1:
return False
return True
def _v2_repair_prompt_for_missing_axes(
self,
prompt,
current_phrases,
global_state,
previous_run,
axis_feedback,
intent_phrases=None,
intent_family_slot=None,
allow_axis_fallback=True,
apply=True,
):
missing_axes = set(axis_feedback.get("missing_axes", [])) if isinstance(axis_feedback, dict) else set()
missing_axes = set(self._v2_order_axes(missing_axes))
if not missing_axes:
return prompt, "Prompt repair: none.", []
candidates = {}
intent_roots = self._v2_repair_intent_roots(prompt)
reference_phrases = [
phrase
for phrase in list(current_phrases or []) + list(intent_phrases or [])
if isinstance(phrase, dict)
]
for item in intent_phrases or []:
intent_roots |= self._v2_repair_intent_roots(item.get("text", ""))
def candidate_axes(item):
if not isinstance(item, dict):
return set()
axes = self._v2_phrase_axes(item)
wanted_axes = item.get("wanted_axes", {})
if isinstance(wanted_axes, dict):
axes |= {axis for axis, count in wanted_axes.items() if int(count or 0) > 0}
axis_counts = item.get("axes", {})
if isinstance(axis_counts, dict):
axes |= {axis for axis, count in axis_counts.items() if int(count or 0) > 0}
return axes
def add_candidate(text, axes, score, source, cluster=None, memory_entry=None):
clean = self._v2_clean_phrase_text(text)
if not clean:
return
if not self._v2_prompt_repair_text_allowed(clean):
return
words = self._v2_phrase_words(clean)
if len(words) > 16:
return
axes = set(axes or set()) & missing_axes
if not axes or self._v2_prompt_contains_text(prompt, clean):
return
if source != "intent" and not self._v2_repair_candidate_matches_intent(clean, intent_roots):
return
if source != "intent" and not self._v2_repair_context_matches_memory(clean, memory_entry, reference_phrases):
return
safe_cluster = [
self._v2_clean_phrase_text(item)
for item in (cluster or [])
if (
self._v2_prompt_repair_text_allowed(item) and
self._v2_repair_candidate_matches_intent(item, intent_roots) and
self._v2_repair_context_matches_memory(item, memory_entry, reference_phrases)
)
]
safe_cluster = [item for item in safe_cluster if item]
existing = candidates.get(clean)
payload = {
"text": clean,
"axes": axes,
"score": float(score),
"source": source,
"cluster": safe_cluster,
}
if existing is None or payload["score"] > existing["score"]:
candidates[clean] = payload
for item in intent_phrases or []:
axes = candidate_axes(item)
if axes & missing_axes:
add_candidate(item.get("text", ""), axes, 3.0, "intent")
if isinstance(intent_family_slot, dict):
for entry in intent_family_slot.get("intent_preference_phrases", {}).values():
if not isinstance(entry, dict):
continue
if int(entry.get("seen_count", 0)) < 2:
continue
axes = candidate_axes(entry)
if not (axes & missing_axes):
continue
score = (
1.35 +
min(0.75, int(entry.get("seen_count", 0)) * 0.14) +
min(0.60, float(entry.get("score", 0.0)) * 0.16)
)
add_candidate(
entry.get("text", ""),
axes,
score,
"intent_preference",
memory_entry=entry,
)
if isinstance(previous_run, dict):
for item in self._v2_concept_units_for_run(previous_run):
if item.get("kind") == "token":
continue
axes = candidate_axes(item)
if axes & missing_axes:
add_candidate(item.get("text", ""), axes, 1.75, "previous", memory_entry=item)
preferred_memory = global_state.get("preferred_context_memory", {}) if isinstance(global_state, dict) else {}
for entry in preferred_memory.values() if isinstance(preferred_memory, dict) else []:
if not isinstance(entry, dict):
continue
axes = candidate_axes(entry)
if not (axes & missing_axes):
continue
cluster = [
self._v2_clean_phrase_text(item)
for item in entry.get("phrases", [])
]
cluster = [
item for item in cluster
if item and self._v2_prompt_repair_text_allowed(item) and not self._v2_prompt_contains_text(prompt, item)
]
score = (
2.0 +
float(entry.get("score", 0.0)) * 0.34 +
min(1.8, int(entry.get("liked_count", 0) or 0) * 0.42) +
min(0.8, int(entry.get("satisfied_count", 0) or 0) * 0.16)
)
if cluster:
for offset, item_text in enumerate(cluster[:8]):
add_candidate(item_text, axes, score - offset * 0.03, "preferred_context", cluster=cluster[:8], memory_entry=entry)
else:
add_candidate(entry.get("anchor", ""), axes, score, "preferred_context", memory_entry=entry)
memory = global_state.get("phrase_memory", {}) if isinstance(global_state, dict) else {}
for text, entry in memory.items():
if not isinstance(entry, dict):
continue
axes = candidate_axes(entry)
wanted_axes = entry.get("wanted_axes", {}) if isinstance(entry.get("wanted_axes", {}), dict) else {}
axis_hits = sum(int(wanted_axes.get(axis, 0) or 0) for axis in missing_axes)
evidence = int(entry.get("category_evidence_count", 0) or 0)
score = (
1.0 +
float(entry.get("score", 0.0)) * 0.18 +
min(1.2, axis_hits * 0.34) +
min(0.5, evidence * 0.05) +
min(0.4, int(entry.get("liked_count", 0) or 0) * 0.08) -
min(1.0, int(entry.get("bad_count", 0) or 0) * 0.20) -
min(0.9, int(entry.get("wrong_count", 0) or 0) * 0.16)
)
if axes & missing_axes and score > 0.60:
add_candidate(entry.get("text", text), axes, score, "memory", memory_entry=entry)
current_axes = set()
for phrase in current_phrases or []:
current_axes |= candidate_axes(phrase)
missing_not_represented = missing_axes - current_axes
source_priority = {
"intent": 4,
"intent_preference": 3,
"previous": 3,
"preferred_context": 2,
"memory": 1,
}
ranked = sorted(
candidates.values(),
key=lambda item: (
bool(item["axes"] & missing_not_represented),
source_priority.get(item["source"], 0),
item["score"],
len(item["text"]),
),
reverse=True,
)
selected = []
selected_texts = set()
seen_axes = set()
for item in ranked:
cluster = item.get("cluster", []) if item.get("source") == "preferred_context" else []
for cluster_text in cluster:
if (
cluster_text in selected_texts or
self._v2_prompt_contains_text(prompt, cluster_text) or
not self._v2_prompt_repair_text_allowed(cluster_text)
):
continue
cluster_item = dict(item)
cluster_item["text"] = cluster_text
selected.append(cluster_item)
selected_texts.add(cluster_text)
seen_axes |= item["axes"]
if len(selected) >= 10:
break
if len(selected) >= 10:
break
if item["text"] in selected_texts:
continue
selected.append(item)
selected_texts.add(item["text"])
seen_axes |= item["axes"]
if len(selected) >= 10 or (len(selected) >= 4 and missing_axes <= seen_axes):
break
if not selected:
fallback_axes = set(V2_FEEDBACK_AXES) - missing_axes
if allow_axis_fallback and fallback_axes:
fallback_feedback = dict(axis_feedback or {})
fallback_feedback["missing_axes"] = self._v2_order_axes(fallback_axes)
fallback_prompt, fallback_status, fallback_candidates = self._v2_repair_prompt_for_missing_axes(
prompt,
current_phrases,
global_state,
previous_run,
fallback_feedback,
intent_phrases=intent_phrases,
intent_family_slot=intent_family_slot,
allow_axis_fallback=False,
apply=apply,
)
if fallback_candidates:
return (
fallback_prompt,
"Prompt repair: no candidates for requested axes "
f"({', '.join(self._v2_order_axes(missing_axes))}); "
f"showing other-axis repair candidates. {fallback_status}",
fallback_candidates,
)
return (
prompt,
"Prompt repair: no stored candidates for requested axes "
f"({', '.join(self._v2_order_axes(missing_axes))}); "
"no other-axis repair candidates either.",
[],
)
return prompt, f"Prompt repair: no stored candidates for missing {', '.join(self._v2_order_axes(missing_axes))}.", []
additions = [item["text"] for item in selected]
source_counts = {}
for item in selected:
source_counts[item["source"]] = int(source_counts.get(item["source"], 0)) + 1
source_summary = ", ".join(f"{source}:{count}" for source, count in sorted(source_counts.items()))
axes_label = ', '.join(self._v2_order_axes(missing_axes))
if not apply:
return (
prompt,
f"Prompt repair: {len(additions)} suggestion(s) for missing {axes_label} ({source_summary}): {', '.join(additions)}.",
selected,
)
repaired = f"{prompt}, {', '.join(additions)}" if str(prompt or "").strip() else ", ".join(additions)
return (
repaired,
f"Prompt repair: added {len(additions)} phrase(s) for missing {axes_label} ({source_summary}): {', '.join(additions)}.",
selected,
)
def _v2_apply_perfect_repair_phrases(
self,
prompt,
intent_family_slot,
intent_prompt="",
intent_phrases=None,
clip=None,
encode_cache=None,
):
if not isinstance(intent_family_slot, dict):
return prompt, "Perfect repairs: none.", []
repairs = intent_family_slot.get("perfect_repairs", {})
if not isinstance(repairs, dict) or not repairs:
return prompt, "Perfect repairs: none.", []
rejected = intent_family_slot.get("rejected_repairs", {})
rejected = rejected if isinstance(rejected, dict) else {}
additions = []
for entry in sorted(
repairs.values(),
key=lambda item: (
int(item.get("count", 0)) if isinstance(item, dict) else 0,
int(item.get("last_seen_iter", 0)) if isinstance(item, dict) else 0,
),
reverse=True,
):
if not isinstance(entry, dict):
continue
text = self._v2_clean_phrase_text(entry.get("text", ""))
if (
not text or
self._v2_prompt_contains_text(prompt, text) or
not self._v2_prompt_repair_text_allowed(text) or
not self._v2_memory_entry_matches_current_scene(
text,
intent_prompt or prompt,
intent_phrases or [],
clip=clip,
encode_cache=encode_cache,
)
):
continue
rejection = rejected.get(text)
if (
isinstance(rejection, dict) and
int(rejection.get("count", 0) or 0) >= int(entry.get("count", 0) or 0) and
not self._v2_phrase_represented_by(text, intent_phrases or [])
):
continue
additions.append(text)
if len(additions) >= 6:
break
if not additions:
return prompt, "Perfect repairs: already represented.", []
repaired = f"{prompt}, {', '.join(additions)}" if str(prompt or "").strip() else ", ".join(additions)
return (
repaired,
f"Perfect repairs: preserved {len(additions)} Perfect-proven phrase(s): {', '.join(additions)}.",
[{"text": text, "source": "perfect_repair", "action": "added"} for text in additions],
)
def _v2_serializable_repair_candidates(self, candidates):
serializable = []
for item in candidates or []:
tests/test_v2_category_alignment.py
-490
View File
@@ -259,62 +259,6 @@ def test_wrong_appearance_rating_is_available():
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 = {
@@ -428,26 +372,6 @@ def test_lucky_composition_gates_unrelated_old_action_by_current_intent():
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": {}}
@@ -616,96 +540,6 @@ def test_intent_alignment_omits_bad_token_in_new_enhancer_phrase():
]
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": []}
@@ -731,78 +565,6 @@ def test_active_repair_axes_persist_until_perfect():
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 = {
@@ -1899,258 +1661,6 @@ def test_normal_previous_run_can_train_before_current_refusal_is_discarded(tmp_p
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": {}}