From e39a64de5711fcdbfa423dd8cb1339c889cfc43d Mon Sep 17 00:00:00 2001 From: digital-garbage <133010426+digital-garbage@users.noreply.github.com> Date: Mon, 15 Jun 2026 18:25:44 +0300 Subject: [PATCH] Stage 2 (2/3): delete dead prompt-repair methods + their tests 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 --- conditioning.py | 431 +----------------------- tests/test_v2_category_alignment.py | 490 ---------------------------- 2 files changed, 1 insertion(+), 920 deletions(-) diff --git a/conditioning.py b/conditioning.py index 08cd60f4..3b130895 100644 --- a/conditioning.py +++ b/conditioning.py @@ -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 []: diff --git a/tests/test_v2_category_alignment.py b/tests/test_v2_category_alignment.py index 31ee5e6b..badfbb0b 100644 --- a/tests/test_v2_category_alignment.py +++ b/tests/test_v2_category_alignment.py @@ -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": {}}