From 9a3ea8c35c4c336e7b430ea50dc35fa4b40a4e10 Mon Sep 17 00:00:00 2001 From: digital-garbage <133010426+digital-garbage@users.noreply.github.com> Date: Sun, 20 Sep 2026 00:58:02 +0300 Subject: [PATCH] fix: timed phrases reach the sampler when a wired conditioning owns the prompt The editor strips the brackets before the encoding node sees them and hands the windows to Studio as char spans on the text it gave that node. Co-Authored-By: Claude Opus 5 --- conditioning.py | 38 +++++++++++++++++++++++++--- movie_editor/backend/bridge.py | 26 ++++++++++++++++--- movie_editor/backend/builder.py | 3 ++- movie_editor/server.py | 5 +++- tests/test_h3_token_weights.py | 44 +++++++++++++++++++++++++++++++++ 5 files changed, 107 insertions(+), 9 deletions(-) diff --git a/conditioning.py b/conditioning.py index b192ff49..9fd3a43d 100644 --- a/conditioning.py +++ b/conditioning.py @@ -11961,14 +11961,29 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner): out.append([cond, meta] if isinstance(entry, list) else (cond, meta)) return out - def _v2_apply_h3_timed_phrases(self, conditioning_list, clip): + def _v2_apply_h3_timed_phrases(self, conditioning_list, clip, link_texts=None): """Locate each scene's timed phrases as token spans for the sampler's attention bias. `_v2_encode_prompt` stripped the `@t0-t1` syntax and recorded the char spans against the clean text it encoded; this maps them onto that text's tokens (offset mapping, no re-encode) and stores them next to the rating-learned spans in `funpack_h3_token_weights`. Always on: the user typed the window. + + A WIRED conditioning was encoded by another node from the editor's expanded text + (`link_texts`): the editor stripped the syntax before that node saw it and hands the + windows over as `link_texts["timed"]`, char spans on `link_texts["prompt"]`. The + text that was actually encoded is verified against the tensor's own tag run + (`choose_encoded_text`), same as phrase emphasis, so the spans land on real tokens. """ + link_timed = (link_texts or {}).get("timed") if isinstance(link_texts, dict) else None + if link_timed and isinstance(conditioning_list, list): + conditioning_list = [ + ([e[0], dict(e[1], funpack_h3_timed=[tuple(t) for t in link_timed], + funpack_h3_timed_link=True)] + if isinstance(e, (list, tuple)) and len(e) >= 2 and isinstance(e[1], dict) + and e[1].get("funpack_conditioning_owner") == "wired" + and not e[1].get("funpack_h3_timed") else e) + for e in conditioning_list] try: try: from . import minimax_h3 as _h3 @@ -11990,8 +12005,25 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner): tokenizer = tokenizer or _tw.h3_tokenizer(clip) if tokenizer is None: return conditioning_list - text = str(meta.get("funpack_h3_timed_text") or "") cond_len = int(cond.shape[1]) if hasattr(cond, "shape") and cond.dim() >= 2 else 0 + if meta.pop("funpack_h3_timed_link", False): + text, _n, verified = _tw.choose_encoded_text( + tokenizer, [link_texts.get("full_prompt"), link_texts.get("prompt")], + meta.get("minimax_token_tags"), cond_len) + if text is None: + _log.feature("FunPackStudio", "H3 timed phrases", False, + "the positive CONDITIONING is wired and neither editor " + "text tokenizes to the tensor's prompt run, so the " + "windows cannot be placed.") + out.append(entry) + continue + if verified is not None: + meta.setdefault("funpack_h3_token_weights", {}) + tw0 = dict(meta["funpack_h3_token_weights"]) + tw0.setdefault("base", int(verified)) + meta["funpack_h3_token_weights"] = tw0 + else: + text = str(meta.get("funpack_h3_timed_text") or "") enc = tokenizer(text, add_special_tokens=False, return_offsets_mapping=True) offsets = list(enc["offset_mapping"]) spans = [] @@ -12269,7 +12301,7 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner): auto_strength=auto_strength, variables=variables, link_texts=link_texts) - out = self._v2_apply_h3_timed_phrases(out, clip) + out = self._v2_apply_h3_timed_phrases(out, clip, link_texts=link_texts) if mode in ("relative", "both"): out = _step("relative steering", out, lambda c: self._v2_apply_scene_refinement_keys( c, scene_refinement_keys, refinement_key, diff --git a/movie_editor/backend/bridge.py b/movie_editor/backend/bridge.py index 207fbb12..09a77bb4 100644 --- a/movie_editor/backend/bridge.py +++ b/movie_editor/backend/bridge.py @@ -100,11 +100,29 @@ def expand_prompt_for_node(text: str, variables=None) -> str: Peeks at the shortcut revolver rather than committing it: the run's own expansion inside Studio is what advances the no-repeat cycle, and doing it twice would skip a slot. """ + return expand_prompt_with_windows(text, variables)[0] + + +def expand_prompt_with_windows(text: str, variables=None): + """`expand_prompt_for_node`, also returning the timed-phrase windows it stripped. + + `[walks left@2.0-3.5]` is an instruction for the sampler, not prompt text: a node that + encodes the string itself must never see the brackets, and Studio - which never sees the + encoded string - needs the windows handed to it separately, as char spans on the clean + text. -> (clean_expanded_text, [(start, end, weight, t0, t1), ...]) + """ expanded = expand_prompt_fragment(text) - if not expanded or not variables: - return expanded - resolve_variables = _funpack_attr("templates", "resolve_variables") - return resolve_variables(expanded, variables)[0] + if expanded and variables: + resolve_variables = _funpack_attr("templates", "resolve_variables") + expanded = resolve_variables(expanded, variables)[0] + if not expanded: + return expanded, [] + try: + parse_timed = _funpack_attr("h3_token_weights", "parse_timed") + clean, timed = parse_timed(expanded) + return clean, [list(t) for t in timed] + except Exception: # noqa: BLE001 - FunPack unreachable: the text goes out as typed + return expanded, [] def parse_timeline_raw(prompt: str) -> dict: diff --git a/movie_editor/backend/builder.py b/movie_editor/backend/builder.py index f13563f0..25c5097f 100644 --- a/movie_editor/backend/builder.py +++ b/movie_editor/backend/builder.py @@ -521,7 +521,8 @@ def build(object_info: dict, models_config: dict, params: dict, media: dict | No # anything that has to line up with those tokens is guessing. _exp = params.get("expanded") if isinstance(_exp, dict) and _exp: - _rf["link_texts"] = {k: str(v) for k, v in _exp.items() if isinstance(v, str)} + _rf["link_texts"] = {k: (str(v) if isinstance(v, str) else v) + for k, v in _exp.items() if isinstance(v, (str, list))} # Project `$name` variables — Studio resolves them dead-last (after split), per scene. _vars = params.get("variables") if isinstance(_vars, (list, dict)) and _vars: diff --git a/movie_editor/server.py b/movie_editor/server.py index 5d311a53..0e9e23e3 100644 --- a/movie_editor/server.py +++ b/movie_editor/server.py @@ -744,7 +744,7 @@ def _expanded_link_texts(target: Project, prompt: str) -> dict[str, str]: def ex(text): return bridge.expand_prompt_for_node(text, variables) - body = ex(prompt) + body, timed = bridge.expand_prompt_with_windows(prompt, variables) postfix = ex(effective_postfix(target)) return { "prompt": body, @@ -752,6 +752,9 @@ def _expanded_link_texts(target: Project, prompt: str) -> dict[str, str]: "postfix": postfix, "negative_prompt": ex(effective_negative_prompt(target)), "full_prompt": " ".join(t for t in (body, postfix) if t), + # Timed phrases stripped from `prompt`, as char spans on it. `full_prompt` starts with + # `prompt`, so the same spans hold there. + "timed": timed, } diff --git a/tests/test_h3_token_weights.py b/tests/test_h3_token_weights.py index 162a3f51..7e953906 100644 --- a/tests/test_h3_token_weights.py +++ b/tests/test_h3_token_weights.py @@ -873,3 +873,47 @@ def test_a_phrase_may_contain_parentheses_colons_and_at_signs(): clean, spans = tw.parse_timed("[camera: (slow) push-in, mail@x.com @1-2] [fast:0.5@2-3]") assert clean == "camera: (slow) push-in, mail@x.com fast" assert spans == [(0, 34, 1.0, 1.0, 2.0), (35, 39, 0.5, 2.0, 3.0)] + + +# --- timed phrases reach the sampler on both encode paths ---------------------- + +def _quiet(monkeypatch): + import conditioning + import minimax_h3 + monkeypatch.setattr(minimax_h3, "is_h3_clip", lambda c: True) + monkeypatch.setattr(conditioning, "_log", + types.SimpleNamespace(failed=lambda *a, **k: None, + note_on_change=lambda *a, **k: None, + feature=lambda *a, **k: None), raising=False) + + +def test_studio_encoded_timed_phrases_become_token_windows(refiner, monkeypatch): + _quiet(monkeypatch) + clean, timed = tw.parse_timed("cat [runs@1-2]") + meta = {"funpack_h3_timed": timed, "funpack_h3_timed_text": clean, + "minimax_token_tags": [1] * 8} + out = refiner._v2_apply_h3_timed_phrases([[torch.zeros(1, 8, 4), meta]], _H3Clip()) + tag = out[0][1]["funpack_h3_token_weights"] + assert tag["prompt_tokens"] == 8 and tag["base"] == 0 + assert tag["timed"] == [(4, 8, 1.0, 1.0, 2.0)] + + +def test_wired_conditioning_takes_its_windows_from_the_editor(refiner, monkeypatch): + """The editor stripped the syntax before the encoding node saw it and hands the windows + over as char spans on the text it gave that node.""" + _quiet(monkeypatch) + link = {"prompt": "cat runs", "full_prompt": "cat runs", "timed": [[4, 8, 1.0, 1.0, 2.0]]} + # reference block (tag 0) in front, then the 8-token prompt + meta = {"funpack_conditioning_owner": "wired", "minimax_token_tags": [0, 0, 0] + [1] * 8} + out = refiner._v2_apply_h3_timed_phrases([[torch.zeros(1, 11, 4), meta]], _H3Clip(), + link_texts=link) + tag = out[0][1]["funpack_h3_token_weights"] + assert tag["base"] == 3 and tag["timed"] == [(4, 8, 1.0, 1.0, 2.0)] + + +def test_a_wired_conditioning_without_editor_windows_is_left_alone(refiner, monkeypatch): + _quiet(monkeypatch) + meta = {"funpack_conditioning_owner": "wired", "minimax_token_tags": [1] * 8} + out = refiner._v2_apply_h3_timed_phrases([[torch.zeros(1, 8, 4), meta]], _H3Clip(), + link_texts={"prompt": "cat runs"}) + assert "funpack_h3_token_weights" not in out[0][1]