Extract Krea format dispatch route

docs/prompt-architecture-improvement-plan.md

@@ -342,6 +342,10 @@ Keep here:
 
 Already isolated:
 
+- `krea_format_route.py` owns top-level Krea dispatch, including option
+  normalization, metadata-vs-text input selection, single-vs-pair branching,
+  extra positive/negative merging, final prose hygiene, and output shape;
+  `krea_formatter.py` keeps the public wrapper.
 - `krea_configured_cast_formatter.py` owns normal metadata configured-cast
   Krea prose assembly behind `KreaConfiguredCastRequest`,
   `KreaConfiguredCastDependencies`, and `KreaConfiguredCastPrompt`;

docs/prompt-pool-routing-map.md

@@ -63,7 +63,7 @@ call the same core generation functions.
 | `SxCP Prompt Builder` | `build_prompt` -> `builder_prompt_route.py` | Direct single prompt generation. Can use built-in categories or JSON categories. |
 | `SxCP Prompt Builder From Configs` | `build_prompt_from_configs` -> `builder_config_route.py` -> `build_prompt` -> `builder_prompt_route.py` | Same generator, but inputs come from category/cast/profile/filter helper nodes. |
 | `SxCP Insta/OF Prompt Pair` | `build_insta_of_pair` | Builds a softcore row and hardcore row with shared cast/continuity options. |
-| `SxCP Krea2 Formatter` | `format_krea2_prompt` | Converts metadata rows or pair metadata into Krea2-friendly prose. |
+| `SxCP Krea2 Formatter` | `format_krea2_prompt` -> `krea_format_route.py` | Converts metadata rows or pair metadata into Krea2-friendly prose. |
 | `SxCP SDXL Formatter` | `format_sdxl_prompt` | Converts metadata rows or pair metadata into SDXL/tag style prompts. |
 | `SxCP Caption Naturalizer` | `naturalize_caption` | Converts rows into more natural sentence captions. |
 
@@ -108,6 +108,7 @@ Core helper ownership:
 | `pair_camera.py` | Insta/OF soft/hard camera route resolution behind `InstaPairCameraRoute`, same-as-softcore camera mode, camera-detail override, camera-aware composition mutation, POV camera suppression, synchronized row/root camera metadata, and legacy dict compatibility. |
 | `pair_clothing.py` | Insta/OF clothing sentence formatting and hardcore clothing continuity behind `HardcorePairClothingRoute`, body-exposure scene cleanup, action-aware body-access flags, conflicting outfit-piece cleanup, configured/default visible-person clothing, final root clothing-state assembly, and legacy dict compatibility. |
 | `pair_output.py` | Insta/OF final pair prompts, trigger preservation, negative prompts, captions, and root pair metadata assembly. |
+| `krea_format_route.py` | Top-level Krea dispatch, option normalization, metadata-vs-text input selection, single-vs-pair branching, extra positive/negative merging, final prose hygiene, and output shape. |
 | `hardcore_role_graphs.py` | Source role graph construction for hardcore configured-cast rows, including POV-aware interaction geometry, called through `row_role_graph.py` for row generation. |
 | `hardcore_role_fallback.py` | Solo, same-sex, mixed group fallback, and support-partner role graph wording for configured casts. |
 | `hardcore_role_interaction.py` | Foreplay, manual stimulation, body worship, clothing transition, dominant guidance, camera performance, aftercare, and group coordination role graph wording. |

krea_format_route.py

@@ -0,0 +1,134 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Callable
+
+
+@dataclass(frozen=True)
+class KreaFormatRequest:
+    source_text: str
+    metadata_json: str = ""
+    negative_prompt: str = ""
+    input_hint: str = "auto"
+    target: str = "auto"
+    detail_level: str = "balanced"
+    style_mode: str = "preserve"
+    preserve_trigger: bool = False
+    extra_positive: str = ""
+    extra_negative: str = ""
+
+
+@dataclass(frozen=True)
+class KreaFormatRoute:
+    output: dict[str, str]
+    branch: str
+    method: str
+    target: str
+    detail_level: str
+    style_mode: str
+
+
+@dataclass(frozen=True)
+class KreaFormatDependencies:
+    trigger_candidates: tuple[str, ...]
+    clean: Callable[[Any], str]
+    row_from_inputs: Callable[[str, str, str], tuple[dict[str, Any] | None, str]]
+    normal_row_to_krea: Callable[[dict[str, Any], str, str], tuple[str, str]]
+    insta_pair_to_krea: Callable[[dict[str, Any], str, str], tuple[str, str, str, str]]
+    fallback_text_to_krea: Callable[[str, bool, str, str], tuple[str, str, str]]
+    append_formatter_hints: Callable[..., str]
+    combine_negative: Callable[..., str]
+    sanitize_prose_text: Callable[..., str]
+    sanitize_negative_text: Callable[[str], str]
+
+
+def format_krea2_prompt_result(request: KreaFormatRequest, deps: KreaFormatDependencies) -> KreaFormatRoute:
+    detail_level = request.detail_level if request.detail_level in ("concise", "balanced", "dense") else "balanced"
+    style_mode = request.style_mode if request.style_mode in ("preserve", "photographic", "minimal") else "preserve"
+    target = request.target if request.target in ("auto", "single", "softcore", "hardcore") else "auto"
+    row, method = deps.row_from_inputs(request.source_text, request.metadata_json, request.input_hint)
+
+    if row and row.get("mode") == "Insta/OF":
+        soft_prompt, soft_negative, hard_prompt, hard_negative = deps.insta_pair_to_krea(
+            row,
+            detail_level,
+            style_mode,
+        )
+        soft_row = row.get("softcore_row") if isinstance(row.get("softcore_row"), dict) else {}
+        hard_row = row.get("hardcore_row") if isinstance(row.get("hardcore_row"), dict) else {}
+        soft_prompt = deps.append_formatter_hints(soft_prompt, row, soft_row)
+        hard_prompt = deps.append_formatter_hints(hard_prompt, row, hard_row)
+        if request.extra_positive.strip():
+            soft_prompt = f"{soft_prompt.rstrip()} {request.extra_positive.strip()}"
+            hard_prompt = f"{hard_prompt.rstrip()} {request.extra_positive.strip()}"
+        soft_prompt = deps.sanitize_prose_text(soft_prompt, triggers=deps.trigger_candidates)
+        hard_prompt = deps.sanitize_prose_text(hard_prompt, triggers=deps.trigger_candidates)
+        selected = hard_prompt if target == "hardcore" else soft_prompt if target == "softcore" else soft_prompt
+        selected_negative = hard_negative if target == "hardcore" else soft_negative
+        negative = deps.sanitize_negative_text(
+            deps.combine_negative(selected_negative, request.negative_prompt, request.extra_negative)
+        )
+        output = {
+            "krea_prompt": selected,
+            "negative_prompt": negative,
+            "krea_softcore_prompt": soft_prompt,
+            "krea_hardcore_prompt": hard_prompt,
+            "softcore_negative_prompt": deps.sanitize_negative_text(
+                deps.combine_negative(soft_negative, request.extra_negative)
+            ),
+            "hardcore_negative_prompt": deps.sanitize_negative_text(
+                deps.combine_negative(hard_negative, request.extra_negative)
+            ),
+            "method": f"{method}:krea2(insta_of_pair)",
+        }
+        return KreaFormatRoute(
+            output=output,
+            branch="insta_of_pair",
+            method=output["method"],
+            target=target,
+            detail_level=detail_level,
+            style_mode=style_mode,
+        )
+
+    if row:
+        prompt, kind = deps.normal_row_to_krea(row, detail_level, style_mode)
+        prompt = deps.append_formatter_hints(prompt, row)
+        extracted_negative = deps.clean(row.get("negative_prompt"))
+        method = f"{method}:krea2({kind})"
+        branch = kind
+    else:
+        prompt, extracted_negative, method = deps.fallback_text_to_krea(
+            request.source_text,
+            request.preserve_trigger,
+            detail_level,
+            style_mode,
+        )
+        branch = "fallback"
+
+    if request.extra_positive.strip():
+        prompt = f"{prompt.rstrip()} {request.extra_positive.strip()}"
+    prompt = deps.sanitize_prose_text(prompt, triggers=deps.trigger_candidates)
+    negative = deps.sanitize_negative_text(
+        deps.combine_negative(extracted_negative, request.negative_prompt, request.extra_negative)
+    )
+    output = {
+        "krea_prompt": prompt,
+        "negative_prompt": negative,
+        "krea_softcore_prompt": "",
+        "krea_hardcore_prompt": "",
+        "softcore_negative_prompt": "",
+        "hardcore_negative_prompt": "",
+        "method": method,
+    }
+    return KreaFormatRoute(
+        output=output,
+        branch=branch,
+        method=method,
+        target=target,
+        detail_level=detail_level,
+        style_mode=style_mode,
+    )
+
+
+def format_krea2_prompt(request: KreaFormatRequest, deps: KreaFormatDependencies) -> dict[str, str]:
+    return format_krea2_prompt_result(request, deps).output

krea_formatter.py

@@ -5,6 +5,7 @@ from typing import Any
 
 try:
     from . import formatter_input as input_policy
+    from . import krea_format_route
     from . import route_metadata as route_metadata_policy
     from .krea_action_context import (
         is_close_foreplay_text as _is_close_foreplay_text,
@@ -40,6 +41,7 @@ try:
     from .prompt_hygiene import sanitize_negative_text, sanitize_prose_text
 except ImportError:  # Allows local smoke tests with `python -c`.
     import formatter_input as input_policy
+    import krea_format_route
     import route_metadata as route_metadata_policy
     from krea_action_context import (
         is_close_foreplay_text as _is_close_foreplay_text,
@@ -604,6 +606,21 @@ def _fallback_text_to_krea(
     return _paragraph([positive]), negative, "text(fallback)"
 
 
+def _krea_format_dependencies() -> krea_format_route.KreaFormatDependencies:
+    return krea_format_route.KreaFormatDependencies(
+        trigger_candidates=TRIGGER_CANDIDATES,
+        clean=_clean,
+        row_from_inputs=_row_from_inputs,
+        normal_row_to_krea=_normal_row_to_krea,
+        insta_pair_to_krea=_insta_pair_to_krea,
+        fallback_text_to_krea=_fallback_text_to_krea,
+        append_formatter_hints=_append_formatter_hints,
+        combine_negative=_combine_negative,
+        sanitize_prose_text=sanitize_prose_text,
+        sanitize_negative_text=sanitize_negative_text,
+    )
+
+
 def format_krea2_prompt(
     source_text: str,
     metadata_json: str = "",
@@ -616,54 +633,18 @@ def format_krea2_prompt(
     extra_positive: str = "",
     extra_negative: str = "",
 ) -> dict[str, str]:
-    detail_level = detail_level if detail_level in ("concise", "balanced", "dense") else "balanced"
+    return krea_format_route.format_krea2_prompt(
-    style_mode = style_mode if style_mode in ("preserve", "photographic", "minimal") else "preserve"
+        krea_format_route.KreaFormatRequest(
-    target = target if target in ("auto", "single", "softcore", "hardcore") else "auto"
+            source_text=source_text,
-    row, method = _row_from_inputs(source_text, metadata_json, input_hint)
+            metadata_json=metadata_json,
-    extracted_negative = ""
+            negative_prompt=negative_prompt,
-
+            input_hint=input_hint,
-    if row and row.get("mode") == "Insta/OF":
+            target=target,
-        soft_prompt, soft_negative, hard_prompt, hard_negative = _insta_pair_to_krea(row, detail_level, style_mode)
+            detail_level=detail_level,
-        soft_row = row.get("softcore_row") if isinstance(row.get("softcore_row"), dict) else {}
+            style_mode=style_mode,
-        hard_row = row.get("hardcore_row") if isinstance(row.get("hardcore_row"), dict) else {}
+            preserve_trigger=preserve_trigger,
-        soft_prompt = _append_formatter_hints(soft_prompt, row, soft_row)
+            extra_positive=extra_positive,
-        hard_prompt = _append_formatter_hints(hard_prompt, row, hard_row)
+            extra_negative=extra_negative,
-        if extra_positive.strip():
+        ),
-            soft_prompt = f"{soft_prompt.rstrip()} {extra_positive.strip()}"
+        _krea_format_dependencies(),
-            hard_prompt = f"{hard_prompt.rstrip()} {extra_positive.strip()}"
+    )
-        soft_prompt = sanitize_prose_text(soft_prompt, triggers=TRIGGER_CANDIDATES)
-        hard_prompt = sanitize_prose_text(hard_prompt, triggers=TRIGGER_CANDIDATES)
-        selected = hard_prompt if target == "hardcore" else soft_prompt if target == "softcore" else soft_prompt
-        selected_negative = hard_negative if target == "hardcore" else soft_negative
-        negative = sanitize_negative_text(_combine_negative(selected_negative, negative_prompt, extra_negative))
-        return {
-            "krea_prompt": selected,
-            "negative_prompt": negative,
-            "krea_softcore_prompt": soft_prompt,
-            "krea_hardcore_prompt": hard_prompt,
-            "softcore_negative_prompt": sanitize_negative_text(_combine_negative(soft_negative, extra_negative)),
-            "hardcore_negative_prompt": sanitize_negative_text(_combine_negative(hard_negative, extra_negative)),
-            "method": f"{method}:krea2(insta_of_pair)",
-        }
-
-    if row:
-        prompt, kind = _normal_row_to_krea(row, detail_level, style_mode)
-        prompt = _append_formatter_hints(prompt, row)
-        extracted_negative = _clean(row.get("negative_prompt"))
-        method = f"{method}:krea2({kind})"
-    else:
-        prompt, extracted_negative, method = _fallback_text_to_krea(source_text, preserve_trigger, detail_level, style_mode)
-
-    if extra_positive.strip():
-        prompt = f"{prompt.rstrip()} {extra_positive.strip()}"
-    prompt = sanitize_prose_text(prompt, triggers=TRIGGER_CANDIDATES)
-    negative = sanitize_negative_text(_combine_negative(extracted_negative, negative_prompt, extra_negative))
-    return {
-        "krea_prompt": prompt,
-        "negative_prompt": negative,
-        "krea_softcore_prompt": "",
-        "krea_hardcore_prompt": "",
-        "softcore_negative_prompt": "",
-        "hardcore_negative_prompt": "",
-        "method": method,
-    }

tools/prompt_smoke.py

@@ -48,6 +48,7 @@ import index_switch_policy  # noqa: E402
 import node_tooltips  # noqa: E402
 import krea_cast  # noqa: E402
 import krea_configured_cast_formatter  # noqa: E402
+import krea_format_route  # noqa: E402
 import krea_formatter  # noqa: E402
 import krea_normal_formatter  # noqa: E402
 import krea_pair_formatter  # noqa: E402
@@ -2322,6 +2323,112 @@ def smoke_formatter_input_policy() -> None:
     _expect("blur" in fallback_sdxl.get("negative_prompt", ""), "SDXL fallback lost Avoid negative text")
 
 
+def smoke_krea_format_route_policy() -> None:
+    row = _prompt_row(
+        name="krea_format_route_single",
+        category="woman",
+        subcategory="random",
+        seed=3601,
+        men_count=0,
+        camera_config=_orbit_camera(horizontal_angle=45, vertical_angle=0, zoom=5.5),
+    )
+    single_request = krea_format_route.KreaFormatRequest(
+        source_text="",
+        metadata_json=_json(row),
+        target="single",
+        detail_level="dense",
+        style_mode="photographic",
+        extra_positive="krea route marker",
+        extra_negative="krea route negative",
+    )
+    typed_single = krea_format_route.format_krea2_prompt_result(
+        single_request,
+        krea_formatter._krea_format_dependencies(),
+    )
+    public_single = krea_formatter.format_krea2_prompt(
+        "",
+        metadata_json=single_request.metadata_json,
+        target=single_request.target,
+        detail_level=single_request.detail_level,
+        style_mode=single_request.style_mode,
+        extra_positive=single_request.extra_positive,
+        extra_negative=single_request.extra_negative,
+    )
+    _expect(typed_single.output == public_single, "Typed Krea format route should match public single formatter output")
+    _expect(typed_single.branch == "metadata(single)", "Typed Krea format route changed single branch")
+    _expect(typed_single.target == "single", "Typed Krea format route lost target normalization")
+    _expect("krea route marker" in typed_single.output.get("krea_prompt", ""), "Typed Krea route lost extra positive")
+    _expect("krea route negative" in typed_single.output.get("negative_prompt", ""), "Typed Krea route lost extra negative")
+
+    pair = pb.build_insta_of_pair(
+        row_number=1,
+        start_index=1,
+        seed=3602,
+        ethnicity="any",
+        figure="random",
+        no_plus_women=False,
+        no_black=False,
+        trigger=Trigger,
+        prepend_trigger_to_prompt=True,
+        options_json=_insta_options(),
+        character_cast=_character_cast(),
+        hardcore_position_config=_action_filter("penetration_only"),
+    )
+    pair_request = krea_format_route.KreaFormatRequest(
+        source_text="",
+        metadata_json=_json(pair),
+        target="hardcore",
+        detail_level="balanced",
+        style_mode="preserve",
+        extra_positive="pair route marker",
+        extra_negative="pair route negative",
+    )
+    typed_pair = krea_format_route.format_krea2_prompt_result(
+        pair_request,
+        krea_formatter._krea_format_dependencies(),
+    )
+    public_pair = krea_formatter.format_krea2_prompt(
+        "",
+        metadata_json=pair_request.metadata_json,
+        target=pair_request.target,
+        detail_level=pair_request.detail_level,
+        style_mode=pair_request.style_mode,
+        extra_positive=pair_request.extra_positive,
+        extra_negative=pair_request.extra_negative,
+    )
+    _expect(typed_pair.output == public_pair, "Typed Krea format route should match public pair formatter output")
+    _expect(typed_pair.branch == "insta_of_pair", "Typed Krea format route changed pair branch")
+    _expect_text("krea_format_route_policy.hard_prompt", typed_pair.output.get("krea_hardcore_prompt"), 40)
+    _expect("pair route marker" in typed_pair.output.get("krea_prompt", ""), "Typed Krea pair route lost extra positive")
+
+    fallback_request = krea_format_route.KreaFormatRequest(
+        source_text="Scene: quiet studio. Pose: seated portrait. Avoid: blur",
+        input_hint="prompt",
+        target="weird",
+        detail_level="verbose",
+        style_mode="invalid",
+        preserve_trigger=False,
+    )
+    typed_fallback = krea_format_route.format_krea2_prompt_result(
+        fallback_request,
+        krea_formatter._krea_format_dependencies(),
+    )
+    public_fallback = krea_formatter.format_krea2_prompt(
+        fallback_request.source_text,
+        input_hint=fallback_request.input_hint,
+        target=fallback_request.target,
+        detail_level=fallback_request.detail_level,
+        style_mode=fallback_request.style_mode,
+        preserve_trigger=fallback_request.preserve_trigger,
+    )
+    _expect(typed_fallback.output == public_fallback, "Typed Krea format route should match public fallback output")
+    _expect(typed_fallback.branch == "fallback", "Typed Krea format route changed fallback branch")
+    _expect(typed_fallback.target == "auto", "Typed Krea format route should normalize invalid target")
+    _expect(typed_fallback.detail_level == "balanced", "Typed Krea format route should normalize invalid detail level")
+    _expect(typed_fallback.style_mode == "preserve", "Typed Krea format route should normalize invalid style mode")
+    _expect("blur" in typed_fallback.output.get("negative_prompt", ""), "Typed Krea fallback route lost Avoid negative")
+
+
 def smoke_formatter_cast_policy() -> None:
     descriptor = (
         "Woman A / primary creator: 25-year-old adult woman, average figure, warm skin, dark hair; "
@@ -5515,6 +5622,7 @@ SMOKE_CASES: list[tuple[str, Callable[[], None]]] = [
     ("row_role_graph_policy", smoke_row_role_graph_policy),
     ("row_assembly_policy", smoke_row_assembly_policy),
     ("formatter_input_policy", smoke_formatter_input_policy),
+    ("krea_format_route_policy", smoke_krea_format_route_policy),
     ("formatter_cast_policy", smoke_formatter_cast_policy),
     ("caption_policy", smoke_caption_policy),
     ("caption_text_policy", smoke_caption_text_policy),