speak the prompt syntax every Stable-Diffusion client already uses (v0.4.89)

A CLIP text encoder reads 77 tokens, two of them markers. Handing a longer
prompt to a diffusers pipeline does not fail: the tokenizer truncates it and
the model renders what fits. The booru-tag fine-tune families (Pony,
Illustrious, NoobAI) open with a run of quality tags, so the part that got
dropped was the subject -- and the request answered 200 with an image, which
is why it was easy to miss.

Image prompts are now encoded here rather than by the pipeline, which is what
makes the rest possible:

  * as many 75-token windows as the prompt needs, concatenated -- no limit
  * the A1111 weight grammar: (tag:1.3), [tag], bare (tag) = 1.1, nesting
    multiplies, \( escapes. Per-token scaling with the window mean restored,
    so emphasis does not drag the brightness along with it
  * BREAK, which closes the current window instead of blending tags meant
    for different subjects
  * clip_skip, mapped to diffusers' own convention -- SDXL counts from the
    penultimate layer, SD 1.x from the final output and needs the layer norm
    applied by hand. Getting that backwards shifts every image

Three places were dropping the negative prompt outright. The sd.cpp path
hardcoded negative_prompt='', so for every GGUF image model whatever the
caller asked to avoid stopped at the API -- half the idiom, since the quality
tags have a negative half (score_4, score_5, score_6). /v1/images/edits and
/v1/images/inpaint had no field for one at all, so an edit could only ever be
steered positively. All three now pass it, and take clip_skip and
prompt_weighting with it.

clip_skip and prompt_weighting are also per-model and server-wide settings,
because a client speaking the plain OpenAI image API has no field to ask for
clip_skip and a model trained at 2 needs it on every request. Request beats
model beats global.

Weighting defaults on, so a literal '(' in prose now weighs 1.1 -- the
ecosystem's convention, with \( and prompt_weighting:false as the way out.
What that cannot be allowed to change is prompts that already worked: against
a real CLIPTextModel a short unweighted prompt encodes bit-identically to what
diffusers produced, because CLIP pads with its own end-of-text token, so
[bos]+toks+[eos]+pad... and [bos]+toks+pad...+[eos] are the same ids. The test
records that reasoning, so a tokenizer where it stops holding is caught.

Flux, SD3 and Z-Image carry a T5 with its own limit and conventions and keep
their pipeline's native text path.

49 tests; full suite 4410 passed with the same 15 pre-existing failures as
0.4.88 (diff empty).
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01S8LAP3ueNzQJtwzhgVbq66

README.md

@@ -838,6 +838,11 @@ persisted per `config_id`, so the configs don't overwrite each other.
 | `POST /v1/images/from3d` | 3D model → rendered 2D image |
 | `GET /v1/images/progress` | Current generation progress (step/total) |
 
+Generations, edits and inpaint take `negative_prompt`, `clip_skip` and the
+`(tag:1.3)` / `[tag]` / `BREAK` prompt grammar, and encode prompts of any
+length (CLIP's 77-token limit does not apply) — see
+[docs/image-prompts.md](docs/image-prompts.md).
+
 ### Video
 
 | Endpoint | Description |

codai/__init__.py

@@ -16,7 +16,7 @@
 
 # Canonical product version for CoderAI — single source of truth. Both the API
 # metadata and the admin web UI read from here.
-__version__ = "0.4.88"
+__version__ = "0.4.89"
 __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.

codai/admin/routes.py

@@ -2927,6 +2927,11 @@ async def api_model_configure(request: Request, username: str = Depends(require_
                 "lora_scale",
                 "lora_train_base_model", "lora_train_nodes",
                 "max_vram", "sdcpp_flash_attn", "sdcpp_diffusion_flash_attn", "vae_tiling",
+                # Image prompt handling: which CLIP layer to read, and whether
+                # the A1111 weight grammar is parsed. Per model because the
+                # booru-tag fine-tunes need clip_skip 2 whatever the caller
+                # sends, and an OpenAI-API client has no field to send it in.
+                "clip_skip", "prompt_weighting",
                 "component_quantization", "output_crf", "force_vram_update",
                 "balanced_gpu_percent", "acceleration",
                 "cache_type_k", "cache_type_v", "kv_offload", "n_batch", "n_ubatch", "n_seq_max",

codai/admin/templates/models.html

@@ -1541,6 +1541,18 @@ window.__DEFAULT_WHISPER_SERVER_PATH__ = {{ default_whisper_server_path|tojson }
         <label style="display:flex;align-items:center;gap:.5rem;cursor:pointer;font-size:13px"><input type="checkbox" id="cfg-vae-tiling"> VAE Tiling <span class="muted">(reduces VRAM at decode)</span></label>
       </div>
 
+      <!-- Prompt handling — image models (SD 1.x / SDXL text encoders) -->
+      <div class="card-title" style="margin-top:1.25rem">Prompt Handling <span class="muted">(image models)</span></div>
+      <div class="form-row" style="margin-bottom:.5rem;max-width:340px">
+        <label class="form-label">CLIP skip <span class="muted">(blank = the pipeline's own)</span></label>
+        <input type="number" id="cfg-clip-skip" class="form-input" min="1" max="12" step="1" placeholder="e.g. 2">
+        <span class="form-hint">Takes the text embedding this many layers from the end of CLIP. The booru-tag SDXL fine-tune families — Pony, Illustrious, NoobAI and their derivatives — are trained at <b>2</b> and look washed out and off-prompt without it. Set it here because a client speaking the OpenAI image API has no field to ask for it; a request's own <code>clip_skip</code> still wins.</span>
+      </div>
+      <div style="display:flex;gap:1.5rem;flex-wrap:wrap;margin-top:.5rem">
+        <label style="display:flex;align-items:center;gap:.5rem;cursor:pointer;font-size:13px"><input type="checkbox" id="cfg-prompt-weighting" checked> Prompt weighting <span class="muted">((tag:1.3), [tag], BREAK)</span></label>
+      </div>
+      <span class="form-hint">On, the A1111 grammar is parsed: <code>(tag:1.3)</code> emphasises, <code>[tag]</code> attenuates, bare <code>(tag)</code> is 1.1, nesting multiplies, <code>\(</code> is a literal bracket, and <code>BREAK</code> starts a new 75-token window. Off, brackets reach the model as text. Either way the prompt is encoded in as many windows as it needs, so nothing is cut at 77 tokens.</span>
+
       <!-- video output quality (CRF) — only relevant for video models -->
       <div class="form-row" style="margin-top:.75rem;max-width:340px">
         <label class="form-label">Video output quality — CRF
@@ -4291,6 +4303,8 @@ function openCfgModal(idx, cfgIdx){
   document.getElementById('cfg-sdcpp-flash').checked = !!s.sdcpp_flash_attn;
   document.getElementById('cfg-sdcpp-diff-flash').checked = !!s.sdcpp_diffusion_flash_attn;
   document.getElementById('cfg-vae-tiling').checked = !!s.vae_tiling;
+  document.getElementById('cfg-clip-skip').value = (s.clip_skip != null ? s.clip_skip : '');
+  document.getElementById('cfg-prompt-weighting').checked = (s.prompt_weighting !== false);
   document.getElementById('cfg-output-crf').value = (s.output_crf != null ? s.output_crf : '');
   document.getElementById('cfg-balanced-gpu-pct').value = (s.balanced_gpu_percent != null ? s.balanced_gpu_percent : '');
   _populateCompSelects();
@@ -5393,6 +5407,9 @@ async function saveModelConfig(){
                           ? null : parseFloat(document.getElementById('cfg-balanced-gpu-pct').value)),
     acceleration:      _collectAccel(),
     turboquant:        _collectTurboQuant(),
+    clip_skip:         (document.getElementById('cfg-clip-skip').value.trim() === ''
+                          ? null : parseInt(document.getElementById('cfg-clip-skip').value, 10)),
+    prompt_weighting:  document.getElementById('cfg-prompt-weighting').checked,
   };
   try{
     const r = await fetch(ROOT_PATH + '/admin/api/model-configure',{

codai/api/images.py

@@ -887,7 +887,75 @@ def _apply_loras(pipeline, loras):
         _log.warning("Could not apply LoRA weights: %s", e)
 
 
-async def _generate_with_diffusers(pipeline, request, global_args, http_request=None):
+def _resolve_clip_skip(request, model_config, args=None):
+    """``clip_skip`` for this request: asked for, else the model's, else global.
+
+    None means "whatever the pipeline does by default", which is the last CLIP
+    layer on SD 1.x and the penultimate one on SDXL. The booru-tag fine-tunes
+    want 2, and a client speaking the OpenAI image API has no field to say so,
+    so the model's own config has to be able to.
+    """
+    for v in (getattr(request, 'clip_skip', None),
+              (model_config or {}).get('clip_skip'),
+              getattr(args or global_args, 'image_clip_skip', None)):
+        if v is None:
+            continue
+        try:
+            v = int(v)
+        except (TypeError, ValueError):
+            continue
+        return v if v > 0 else None
+    return None
+
+
+def _resolve_prompt_weighting(request, model_config, args=None) -> bool:
+    """Whether to parse the A1111 weight grammar. On unless something says no."""
+    for v in (getattr(request, 'prompt_weighting', None),
+              (model_config or {}).get('prompt_weighting'),
+              getattr(args or global_args, 'image_prompt_weighting', None)):
+        if v is not None:
+            return bool(v)
+    return True
+
+
+def _weighted_embed_kwargs(pipeline, prompt, neg_prompt, clip_skip, weighting):
+    """Encode prompt + negative ourselves, or None to use the pipeline's path.
+
+    Keeps the whole prompt (as many 75-token windows as it takes) and applies
+    per-token weights, which is the only way the booru-tag prompting style
+    survives contact with CLIP's 77-token limit.
+    """
+    try:
+        from codai.models.prompt_weighting import (
+            encode_weighted_prompt, has_weighting_syntax, supports_weighted_embeds,
+            window_count,
+        )
+        if not supports_weighted_embeds(pipeline):
+            return None
+        if not weighting and has_weighting_syntax(prompt):
+            # Worth saying: the caller wrote the grammar and asked for it not to
+            # be parsed, so those brackets are about to reach CLIP as text.
+            _log.info("prompt_weighting is off: '(tag:1.3)' / BREAK in this "
+                      "prompt are being sent to the model literally")
+        kwargs = encode_weighted_prompt(
+            pipeline, prompt or "", neg_prompt or "",
+            clip_skip=clip_skip, plain=not weighting,
+            dtype=getattr(getattr(pipeline, 'text_encoder', None), 'dtype', None),
+        )
+        if kwargs and _log.isEnabledFor(logging.INFO):
+            n = window_count(pipeline.tokenizer, prompt or "", plain=not weighting)
+            if n > 1:
+                _log.info("Prompt encoded in %d windows: %d tokens of room, "
+                          "none truncated (CLIP alone holds 75)", n, n * 75)
+        return kwargs
+    except Exception as e:
+        _log.warning("Weighted prompt encode failed (%s) — falling back to the "
+                     "pipeline's own encoder", e)
+        return None
+
+
+async def _generate_with_diffusers(pipeline, request, global_args, http_request=None,
+                                   model_config=None):
     """Generate images using a diffusers pipeline (with prompt-embedding cache)."""
     import torch
     import numpy as np
@@ -971,10 +1039,17 @@ async def _generate_with_diffusers(pipeline, request, global_args, http_request=
     # cache miss instead of feeding stale GPU tensors into a fresh pipeline.
     _pipe_cls = str(type(pipeline).__name__)
     _model_name = getattr(pipeline, 'model_name_or_path', None) or _pipe_cls
-    model_id = f"{_model_name}#{id(pipeline)}"
     neg_prompt = getattr(request, 'negative_prompt', None) or ""
     do_cfg = cfg_scale > 1.0
 
+    # clip_skip and the weight grammar both change what the embeddings ARE, so
+    # they belong in the cache key: without them the first request's layer
+    # choice would be served to every later one.
+    _clip_skip = _resolve_clip_skip(request, model_config, global_args)
+    _weighting = _resolve_prompt_weighting(request, model_config, global_args)
+    model_id = (f"{_model_name}#{id(pipeline)}"
+                f"|cs={_clip_skip}|w={int(_weighting)}")
+
     # Some pipelines (Z-Image) return prompt embeddings as *per-sample lists*
     # rather than stacked tensors, and the pipeline consumes/extends those
     # lists during a run.  Reusing them on a later request corrupts the batch
@@ -996,7 +1071,17 @@ async def _generate_with_diffusers(pipeline, request, global_args, http_request=
         # native encode + batching (the only path that survives reuse here).
         pass
     else:
-        # Try to encode and cache
+        # First choice: encode it ourselves, in as many 75-token windows as the
+        # prompt needs and with per-token weights. The pipeline's own encoder
+        # cannot do either — it truncates at 77 and reads the brackets as text.
+        embed_kwargs = _weighted_embed_kwargs(
+            pipeline, request.prompt, neg_prompt, _clip_skip, _weighting) or {}
+        if embed_kwargs and _embed_cacheable:
+            _embed_cache.put(request.prompt, neg_prompt, model_id, embed_kwargs)
+
+    if not embed_kwargs and not cache_hit and _embed_cacheable:
+        # Not a CLIP-pair pipeline (Flux, SD3), or the encode failed: the
+        # pipeline's own path, as before.
         try:
             if hasattr(pipeline, 'encode_prompt'):
                 import inspect as _inspect
@@ -1311,31 +1396,41 @@ async def _generate_with_sdcpp(sd_model, request, global_args, http_request=None
     # Use request seed if provided, otherwise use CLI default seed
     seed = request.seed if request.seed is not None else getattr(global_args, 'image_seed', None)
 
+    # sd.cpp parses the weight grammar and the token windows itself, inside the
+    # engine, so the prompt goes through as written. What it was never given is
+    # the negative prompt: this call hardcoded '' and silently discarded
+    # whatever the caller asked to avoid -- the half of the booru-tag idiom that
+    # holds quality down (score_4, score_5, score_6) stopped at the API.
+    _sd_kwargs = dict(
+        prompt=prompt,
+        negative_prompt=getattr(request, 'negative_prompt', None) or '',
+        width=width,
+        height=height,
+        cfg_scale=cfg_scale,
+        sample_steps=steps,
+        seed=seed if seed is not None else 42,
+        batch_count=request.n if request.n else 1,
+        progress_callback=_sdcpp_progress,
+    )
+    _cs = _resolve_clip_skip(request, model_config, global_args)
+    if _cs:
+        _sd_kwargs['clip_skip'] = _cs
+    # Drop anything this build of the binding does not take, rather than losing
+    # the whole call to a TypeError and retrying blind.
     try:
-        result = await asyncio.to_thread(
-            sd_model.generate_image,
-            prompt=prompt,
-            negative_prompt='',
-            width=width,
-            height=height,
-            cfg_scale=cfg_scale,
-            sample_steps=steps,
-            seed=seed if seed is not None else 42,
-            batch_count=request.n if request.n else 1,
-            progress_callback=_sdcpp_progress,
-        )
+        import inspect as _inspect
+        _accepted = set(_inspect.signature(sd_model.generate_image).parameters)
+        if 'kwargs' not in _accepted:
+            _sd_kwargs = {k: v for k, v in _sd_kwargs.items() if k in _accepted}
+    except (TypeError, ValueError):
+        pass
+
+    try:
+        result = await asyncio.to_thread(sd_model.generate_image, **_sd_kwargs)
     except TypeError:
-        result = await asyncio.to_thread(
-            sd_model.generate_image,
-            prompt=prompt,
-            negative_prompt='',
-            width=width,
-            height=height,
-            cfg_scale=cfg_scale,
-            sample_steps=steps,
-            seed=seed if seed is not None else 42,
-            batch_count=request.n if request.n else 1,
-        )
+        _sd_kwargs.pop('progress_callback', None)
+        _sd_kwargs.pop('clip_skip', None)
+        result = await asyncio.to_thread(sd_model.generate_image, **_sd_kwargs)
     except Exception as e:
         task_registry.finish(_tid, "error", str(e)[:200])
         _progress_done()
@@ -1619,7 +1714,10 @@ async def create_image_generation(request: ImageGenerationRequest, http_request:
             else:
                 # Assume it's a diffusers pipeline
                 print(f"Using cached diffusers pipeline for generation")
-                return await _generate_with_diffusers(pipeline, request, global_args, http_request)
+                _diff_cfg = (multi_model_manager.config.get(model_key)
+                             or multi_model_manager.config.get(model_name) or {})
+                return await _generate_with_diffusers(pipeline, request, global_args,
+                                                      http_request, model_config=_diff_cfg)
         
         # =====================================================================
         # Step 4: Model not loaded - try to load it
@@ -1656,7 +1754,8 @@ async def create_image_generation(request: ImageGenerationRequest, http_request:
                     print(f"Loaded diffusers model: {model_name}")
 
                     _treg.finish(_ltid, "done")
-                    return await _generate_with_diffusers(pipeline, request, global_args, http_request)
+                    return await _generate_with_diffusers(pipeline, request, global_args,
+                                                          http_request, model_config=_diff_cfg)
                     
             except ImportError as e:
                 diffusers_error = str(e)
@@ -1732,6 +1831,9 @@ class ImageEditRequest(BaseModel):
     prompt: str
     image: str              # /v1/files path, sha256: id, data URI, URL or base64
     mask: Optional[str] = None   # accepted but unused here -- see /v1/images/inpaint
+    negative_prompt: Optional[str] = None   # what to avoid; CFG needs it as much as a t2i call does
+    clip_skip: Optional[int] = None         # None = the model's configured default
+    prompt_weighting: Optional[bool] = None # None = on; see codai/models/prompt_weighting.py
     n: int = 1
     #: Omitted means "keep the source's shape" (see _default_edit_size), NOT a
     #: square: defaulting to 1024x1024 silently distorted every non-square photo.
@@ -1788,6 +1890,24 @@ def _decode_b64_image(data: str, field: str = "image"):
     return PILImage.open(io.BytesIO(_image_bytes(data, field))).convert("RGB")
 
 
+def _prompt_call_kwargs(pipeline, request, model_config=None):
+    """Prompt kwargs for an img2img / inpaint call: embeddings, or plain text.
+
+    The weighted encoder keeps the whole prompt and the weights; if this is not
+    a CLIP-pair pipeline it returns None and the text goes through as before.
+    Either way the negative prompt is passed -- the edit endpoints used to have
+    nowhere to put one, so an edit could only ever be steered positively.
+    """
+    neg = getattr(request, 'negative_prompt', None) or ""
+    kw = _weighted_embed_kwargs(
+        pipeline, request.prompt, neg,
+        _resolve_clip_skip(request, model_config, global_args),
+        _resolve_prompt_weighting(request, model_config, global_args))
+    if kw:
+        return kw
+    return {"prompt": request.prompt, "negative_prompt": neg or None}
+
+
 def _load_img2img_pipeline(model_name: str, global_args):
     """Load a diffusers img2img pipeline."""
     import torch
@@ -1891,16 +2011,18 @@ async def create_image_edit(request: ImageEditRequest, http_request: Request = N
         num_steps = request.steps or (30 if quality == "standard" else 50)
         cfg_scale = request.guidance_scale or (getattr(global_args, 'image_cfg_scale', 7.5) if quality == "standard" else 9.0)
 
+        _edit_cfg = (multi_model_manager.config.get(model_key)
+                     or multi_model_manager.config.get(model_name) or {})
         result = await asyncio.get_event_loop().run_in_executor(
             None,
             lambda: pipe(
-                prompt=request.prompt,
                 image=source_img,
                 strength=request.strength,
                 num_inference_steps=num_steps,
                 guidance_scale=cfg_scale,
                 num_images_per_prompt=request.n,
                 generator=generator,
+                **_prompt_call_kwargs(pipe, request, _edit_cfg),
             )
         )
     except Exception as e:
@@ -1925,6 +2047,9 @@ class ImageInpaintRequest(BaseModel):
     prompt: str
     image: str              # base64 source image
     mask: str               # base64 mask (white = inpaint region)
+    negative_prompt: Optional[str] = None
+    clip_skip: Optional[int] = None
+    prompt_weighting: Optional[bool] = None
     n: int = 1
     size: Optional[str] = "1024x1024"
     response_format: Optional[str] = "url"
@@ -2020,10 +2145,11 @@ async def create_image_inpaint(request: ImageInpaintRequest, http_request: Reque
         quality = request.quality or "standard"
         num_steps = request.steps or (30 if quality == "standard" else 50)
         cfg_scale = request.guidance_scale or (getattr(global_args, 'image_cfg_scale', 7.5) if quality == "standard" else 9.0)
+        _inp_cfg = (multi_model_manager.config.get(model_key)
+                    or multi_model_manager.config.get(model_name) or {})
         result = await asyncio.get_event_loop().run_in_executor(
             None,
             lambda: pipe(
-                prompt=request.prompt,
                 image=source_img,
                 mask_image=mask_img,
                 strength=request.strength,
@@ -2031,6 +2157,7 @@ async def create_image_inpaint(request: ImageInpaintRequest, http_request: Reque
                 guidance_scale=cfg_scale,
                 num_images_per_prompt=request.n,
                 generator=generator,
+                **_prompt_call_kwargs(pipe, request, _inp_cfg),
             )
         )
     except Exception as e:

codai/config.py

@@ -303,6 +303,13 @@ class ImageConfig:
     seed: Optional[int] = None
     vae_tiling: bool = False
     clip_on_cpu: bool = False
+    # Which CLIP layer the text embedding comes from, counted from the end.
+    # None = the pipeline's own choice. The booru-tag SDXL fine-tunes (Pony,
+    # Illustrious, NoobAI) are trained at 2; a model can override this.
+    clip_skip: Optional[int] = None
+    # Parse the A1111 prompt grammar -- (tag:1.3), [tag], BREAK. Off keeps
+    # brackets literal; the prompt is windowed past 77 tokens either way.
+    prompt_weighting: bool = True
 
 
 @dataclass
@@ -2128,7 +2135,9 @@ class ConfigManager:
                 "cpu_offload": self.config.image.cpu_offload,
                 "seed": self.config.image.seed,
                 "vae_tiling": self.config.image.vae_tiling,
-                "clip_on_cpu": self.config.image.clip_on_cpu
+                "clip_on_cpu": self.config.image.clip_on_cpu,
+                "clip_skip": self.config.image.clip_skip,
+                "prompt_weighting": self.config.image.prompt_weighting
             },
             "archive": {
                 "enabled": self.config.archive.enabled,

codai/main.py

@@ -1225,6 +1225,8 @@ def main():
     global_args.image_seed = config.image.seed
     global_args.vae_tiling = config.image.vae_tiling
     global_args.clip_on_cpu = config.image.clip_on_cpu
+    global_args.image_clip_skip = config.image.clip_skip
+    global_args.image_prompt_weighting = config.image.prompt_weighting
     global_args.system_prompt = config.system_prompt
     global_args.tools_closer_prompt = config.tools_closer_prompt
     global_args.grammar_guided_gen = config.grammar_guided

codai/models/prompt_weighting.py

+# CoderAI - OpenAI-compatible API server
+# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+"""The prompt syntax every Stable-Diffusion client already speaks.
+
+A CLIP text encoder takes 77 tokens, two of which are the start and end
+markers. Handing a longer prompt to a diffusers pipeline does not fail: the
+tokenizer truncates it and the model renders what fits, silently. That is
+fatal for the booru-tag prompting style the SDXL fine-tune families are
+trained on (Pony, Illustrious, NoobAI and their derivatives), where a prompt
+opens with a run of quality tags -- ``score_9, score_8_up, score_7_up, ...``
+-- and spends twenty-odd tokens before reaching the subject. Truncated at 77,
+the subject is what gets dropped.
+
+Those models also expect two things the plain text path cannot express:
+
+* **weights** -- ``(tag:1.3)`` to emphasise, ``[tag]`` to attenuate, bare
+  ``(tag)`` for 1.1 and nesting that multiplies. Without a parser the brackets
+  and the colon reach CLIP as literal tokens, i.e. as noise in the prompt.
+* **BREAK** -- start the next 75-token window here, so tags meant to describe
+  different things are not blended by sharing one window.
+
+This module implements both, and the ``clip_skip`` those fine-tunes are
+trained with, by encoding the prompt ourselves: split it into weighted
+fragments, tokenize to as many 75-token windows as it needs, encode each
+window, scale per token, and concatenate. The result goes to the pipeline as
+``prompt_embeds`` instead of ``prompt``, so no caller loses tokens and no
+pipeline has to support any of it.
+
+The grammar is A1111's, which ComfyUI, InvokeAI and every prompt a user is
+likely to paste also use. One consequence worth stating plainly: a literal
+parenthesis in prose now carries a weight of 1.1. ``\\(`` escapes it, and
+``prompt_weighting: false`` on a request (or ``prompt_weighting`` in the
+model's config) turns the parser off while keeping the long-prompt windowing.
+
+Only the CLIP-encoder families are handled -- SD 1.x/2.x (one encoder) and
+SDXL (two, with a pooled vector from the second). Flux, SD3 and Z-Image carry
+a T5 with its own much longer limit and their own conventions;
+:func:`supports_weighted_embeds` returns False for those and the caller keeps
+the pipeline's native path.
+"""
+
+import re
+from typing import List, Optional, Tuple
+
+#: Tokens per encoded window, excluding the start and end markers that bring
+#: each window to CLIP's 77.
+WINDOW = 75
+
+#: Parsed fragments use this sentinel, with weight -1, for an explicit BREAK.
+BREAK = "BREAK"
+
+_re_attention = re.compile(r"""
+\\\(|
+\\\)|
+\\\[|
+\\]|
+\\\\|
+\\|
+\(|
+\[|
+:\s*([+-]?[.\d]+)\s*\)|
+\)|
+]|
+[^\\()\[\]:]+|
+:
+""", re.X)
+
+_re_break = re.compile(r"\s*\bBREAK\b\s*", re.S)
+
+_ROUND = 1.1
+_SQUARE = 1 / 1.1
+
+
+def parse_prompt_attention(text: str) -> List[list]:
+    """``text`` as ``[[fragment, weight], ...]``, plus ``[BREAK, -1]`` markers.
+
+    ``(a)`` weighs 1.1, ``[a]`` weighs 1/1.1, ``(a:1.3)`` weighs exactly 1.3,
+    and nesting multiplies, so ``((a))`` is 1.21. A backslash escapes a
+    bracket. Unclosed brackets still apply to everything after them, which is
+    what a user who forgot the closing one meant.
+    """
+    res: List[list] = []
+    round_pos: List[int] = []
+    square_pos: List[int] = []
+
+    def multiply_from(start: int, factor: float) -> None:
+        for p in range(start, len(res)):
+            res[p][1] *= factor
+
+    for m in _re_attention.finditer(text or ""):
+        token = m.group(0)
+        weight = m.group(1)
+        if token.startswith("\\"):
+            res.append([token[1:], 1.0])
+        elif token == "(":
+            round_pos.append(len(res))
+        elif token == "[":
+            square_pos.append(len(res))
+        elif weight is not None and round_pos:
+            multiply_from(round_pos.pop(), float(weight))
+        elif token == ")" and round_pos:
+            multiply_from(round_pos.pop(), _ROUND)
+        elif token == "]" and square_pos:
+            multiply_from(square_pos.pop(), _SQUARE)
+        else:
+            parts = _re_break.split(token)
+            for i, part in enumerate(parts):
+                if i > 0:
+                    res.append([BREAK, -1])
+                res.append([part, 1.0])
+
+    for pos in round_pos:
+        multiply_from(pos, _ROUND)
+    for pos in square_pos:
+        multiply_from(pos, _SQUARE)
+
+    if not res:
+        return [["", 1.0]]
+
+    # Merge neighbours that ended up with the same weight: fewer, longer
+    # fragments tokenize closer to how the model saw the text in training
+    # (CLIP's BPE merges across what would otherwise be a fragment boundary).
+    i = 0
+    while i + 1 < len(res):
+        if res[i][1] == res[i + 1][1] and BREAK not in (res[i][0], res[i + 1][0]):
+            res[i][0] += res[i + 1][0]
+            res.pop(i + 1)
+        else:
+            i += 1
+    return res
+
+
+def has_weighting_syntax(text: str) -> bool:
+    """True when ``text`` uses anything the parser would change.
+
+    Only used for reporting: the encoder runs the same either way.
+    """
+    if not text:
+        return False
+    if _re_break.search(text):
+        return True
+    return any(w != 1.0 for _, w in parse_prompt_attention(text))
+
+
+# --------------------------------------------------------------------------- #
+#  tokenizing to windows
+# --------------------------------------------------------------------------- #
+
+def _pad_id(tokenizer) -> int:
+    pad = getattr(tokenizer, "pad_token_id", None)
+    if pad is None:
+        pad = getattr(tokenizer, "eos_token_id", None)
+    return 0 if pad is None else int(pad)
+
+
+def tokenize_weighted(tokenizer, text: str,
+                      plain: bool = False) -> Tuple[List[List[int]], List[List[float]]]:
+    """``text`` as equal-length windows of token ids and per-token weights.
+
+    Each window is ``WINDOW`` ids long: the last one is padded, and a BREAK
+    closes the current window early. The start/end markers are added by
+    :func:`_wrap`, not here, so a window can be re-wrapped per encoder.
+
+    ``plain`` skips the weight grammar entirely (every token weighs 1.0) for a
+    caller that wants the long-prompt windowing without it.
+    """
+    fragments = [[text or "", 1.0]] if plain else parse_prompt_attention(text)
+    pad = _pad_id(tokenizer)
+
+    windows: List[List[int]] = []
+    weights: List[List[float]] = []
+    ids: List[int] = []
+    ws: List[float] = []
+
+    def close(force: bool) -> None:
+        """Push the current window, padded out, if there is one (or ``force``)."""
+        nonlocal ids, ws
+        if not ids and not force:
+            return
+        ids = ids + [pad] * (WINDOW - len(ids))
+        ws = ws + [1.0] * (WINDOW - len(ws))
+        windows.append(ids)
+        weights.append(ws)
+        ids, ws = [], []
+
+    for fragment, weight in fragments:
+        if fragment == BREAK and weight < 0:
+            close(force=True)
+            continue
+        if not fragment:
+            continue
+        # add_special_tokens=False: the markers belong to the window, not to
+        # each fragment. truncation=False because the whole point is to keep
+        # what a single window cannot hold.
+        chunk = tokenizer(fragment, add_special_tokens=False,
+                          truncation=False).input_ids
+        for tid in chunk:
+            if len(ids) == WINDOW:
+                close(force=True)
+            ids.append(int(tid))
+            ws.append(float(weight))
+
+    close(force=not windows)
+    return windows, weights
+
+
+def _wrap(tokenizer, window: List[int]) -> List[int]:
+    """One window with CLIP's start and end markers, i.e. 77 ids."""
+    bos = getattr(tokenizer, "bos_token_id", None)
+    eos = getattr(tokenizer, "eos_token_id", None)
+    pad = _pad_id(tokenizer)
+    bos = pad if bos is None else int(bos)
+    eos = pad if eos is None else int(eos)
+    return [bos] + list(window) + [eos]
+
+
+# --------------------------------------------------------------------------- #
+#  encoding
+# --------------------------------------------------------------------------- #
+
+def supports_weighted_embeds(pipeline) -> bool:
+    """True when this pipeline's text encoders are the CLIP pair we handle.
+
+    A second encoder that is not CLIP means a T5 (Flux, SD3) whose limit and
+    conventions are different, so those keep their native path.
+    """
+    tok = getattr(pipeline, "tokenizer", None)
+    enc = getattr(pipeline, "text_encoder", None)
+    if tok is None or enc is None:
+        return False
+    if "CLIP" not in type(enc).__name__:
+        return False
+    enc2 = getattr(pipeline, "text_encoder_2", None)
+    if enc2 is not None and "CLIP" not in type(enc2).__name__:
+        return False
+    if enc2 is not None and getattr(pipeline, "tokenizer_2", None) is None:
+        return False
+    return True
+
+
+def _is_sdxl(pipeline) -> bool:
+    return getattr(pipeline, "text_encoder_2", None) is not None
+
+
+def _hidden(out, text_encoder, clip_skip: Optional[int], sdxl: bool):
+    """The hidden state a diffusers pipeline would have used.
+
+    Mirrors diffusers exactly, because anything else shifts every output:
+    SDXL always takes the penultimate layer, so its ``clip_skip`` counts from
+    there, while SD 1.x takes the final (already normalised) output and a
+    ``clip_skip`` there needs the final layer norm applied by hand.
+    """
+    if sdxl:
+        layer = -2 if not clip_skip else -(int(clip_skip) + 2)
+        return out.hidden_states[layer]
+    if not clip_skip:
+        return out[0]
+    emb = out.hidden_states[-(int(clip_skip) + 1)]
+    try:
+        return text_encoder.text_model.final_layer_norm(emb)
+    except AttributeError:
+        return emb
+
+
+def _apply_weights(emb, weights):
+    """Scale each token's embedding, keeping the window's overall magnitude.
+
+    Multiplying alone changes the mean of the window, which shows up as a
+    brightness/contrast shift that has nothing to do with the emphasis asked
+    for. Restoring the original mean afterwards is what A1111 does and what
+    the weights were tuned against.
+    """
+    import torch
+    if all(w == 1.0 for w in weights):
+        return emb
+    w = torch.tensor(weights, dtype=emb.dtype, device=emb.device)
+    w = w.unsqueeze(0).unsqueeze(-1)            # (1, tokens, 1)
+    before = emb.mean()
+    emb = emb * w
+    after = emb.mean()
+    if float(after) != 0.0:
+        emb = emb * (before / after)
+    return emb
+
+
+def encode_weighted_prompt(pipeline, prompt: str, negative_prompt: str = "",
+                           clip_skip: Optional[int] = None,
+                           plain: bool = False,
+                           device=None, dtype=None) -> Optional[dict]:
+    """Pipeline kwargs for ``prompt`` with no 77-token limit and real weights.
+
+    Returns ``{'prompt_embeds', 'negative_prompt_embeds'}``, plus the two
+    pooled vectors on SDXL -- ready to pass straight to the pipeline instead
+    of ``prompt``/``negative_prompt``. Returns None when the pipeline is not
+    one of the CLIP families, so the caller can fall back.
+
+    The positive and negative prompts are padded to the same number of
+    windows: classifier-free guidance subtracts one from the other, so they
+    have to be the same length whatever the user typed.
+    """
+    import torch
+
+    if not supports_weighted_embeds(pipeline):
+        return None
+
+    sdxl = _is_sdxl(pipeline)
+    pairs = [(pipeline.tokenizer, pipeline.text_encoder)]
+    if sdxl:
+        pairs.append((pipeline.tokenizer_2, pipeline.text_encoder_2))
+
+    if device is None:
+        device = getattr(pipeline, "device", None)
+
+    # One window count for both sides, per tokenizer. The two SDXL tokenizers
+    # are the same BPE with different padding, so they agree in practice, but
+    # taking the max keeps a mismatch from truncating anything.
+    tokenized = []
+    n_windows = 1
+    for tokenizer, _ in pairs:
+        pos = tokenize_weighted(tokenizer, prompt or "", plain=plain)
+        neg = tokenize_weighted(tokenizer, negative_prompt or "", plain=plain)
+        tokenized.append((tokenizer, pos, neg))
+        n_windows = max(n_windows, len(pos[0]), len(neg[0]))
+
+    def _windows_to_embeds(tokenizer, text_encoder, windows, weights):
+        """Encode every window and lay them end to end -> (embeds, pooled)."""
+        pad = _pad_id(tokenizer)
+        embeds = []
+        pooled = None
+        for i in range(n_windows):
+            window = windows[i] if i < len(windows) else [pad] * WINDOW
+            w = weights[i] if i < len(weights) else [1.0] * WINDOW
+            ids = torch.tensor([_wrap(tokenizer, window)], dtype=torch.long,
+                               device=device)
+            out = text_encoder(ids, output_hidden_states=True)
+            emb = _hidden(out, text_encoder, clip_skip, sdxl)
+            # The pooled vector summarises a whole 77-token pass, so there is
+            # no meaningful way to pool across windows: SDXL gets the first
+            # one, which holds the start of the prompt.
+            if pooled is None and sdxl and hasattr(out, "text_embeds"):
+                pooled = out.text_embeds
+            embeds.append(_apply_weights(emb, [1.0] + list(w) + [1.0]))
+        return torch.cat(embeds, dim=1), pooled
+
+    pos_parts, neg_parts = [], []
+    pooled_pos = pooled_neg = None
+    with torch.no_grad():
+        for (tokenizer, pos, neg), (_, text_encoder) in zip(tokenized, pairs):
+            emb, pooled = _windows_to_embeds(tokenizer, text_encoder, *pos)
+            pos_parts.append(emb)
+            if pooled is not None:
+                pooled_pos = pooled
+            emb, pooled = _windows_to_embeds(tokenizer, text_encoder, *neg)
+            neg_parts.append(emb)
+            if pooled is not None:
+                pooled_neg = pooled
+
+    # SDXL conditions on both encoders' states concatenated on the feature
+    # axis (768 + 1280); SD 1.x has only the one.
+    prompt_embeds = torch.cat(pos_parts, dim=-1) if len(pos_parts) > 1 else pos_parts[0]
+    negative_embeds = torch.cat(neg_parts, dim=-1) if len(neg_parts) > 1 else neg_parts[0]
+
+    if dtype is not None:
+        prompt_embeds = prompt_embeds.to(dtype)
+        negative_embeds = negative_embeds.to(dtype)
+
+    out = {
+        "prompt_embeds": prompt_embeds,
+        "negative_prompt_embeds": negative_embeds,
+    }
+    if sdxl and pooled_pos is not None and pooled_neg is not None:
+        if dtype is not None:
+            pooled_pos = pooled_pos.to(dtype)
+            pooled_neg = pooled_neg.to(dtype)
+        out["pooled_prompt_embeds"] = pooled_pos
+        out["negative_pooled_prompt_embeds"] = pooled_neg
+    return out
+
+
+def window_count(tokenizer, prompt: str, plain: bool = False) -> int:
+    """How many 75-token windows ``prompt`` needs. For logging and tests."""
+    windows, _ = tokenize_weighted(tokenizer, prompt, plain=plain)
+    return len(windows)

codai/pydantic/imagerequest.py

@@ -53,6 +53,15 @@ class ImageGenerationRequest(BaseModel):
         "Null out the diffusers safety_checker so uncensored fine-tunes are not blocked. "
         "Only affects SD 1.x/2.x (SDXL/Flux ship no checker)."))
     negative_prompt: Optional[str] = Field(None, description="What to avoid in the output.")
+    clip_skip: Optional[int] = Field(None, description=(
+        "Take the text embedding this many CLIP layers from the end instead of the last. "
+        "The booru-tag SDXL fine-tune families (Pony, Illustrious, NoobAI) are trained at 2. "
+        "Omitted uses the model's configured default, then the pipeline's own."))
+    prompt_weighting: Optional[bool] = Field(None, description=(
+        "Parse the A1111 prompt grammar: '(tag:1.3)' to emphasise, '[tag]' to attenuate, "
+        "'(tag)' for 1.1, nesting multiplies, '\\\\(' escapes, and BREAK starts a new "
+        "75-token window. False keeps brackets literal. Either way the prompt is encoded "
+        "in as many windows as it needs, so nothing is truncated at 77 tokens."))
 
     # Per-request component overrides
     vae_model: Optional[str] = Field(None, description="Override the VAE for this request.")

docs/README.md

@@ -28,6 +28,7 @@ reference; these are the deep dives.
 | [ocr.md](ocr.md) | The `/v1/ocr` document-transcription subsystem |
 | [chunking.md](chunking.md) | `POST /v1/chunk`: splitting a document for a retrieval index, with the structure decided per request by a template |
 | [expressive-tts.md](expressive-tts.md) | Expressive text-to-speech |
+| [image-prompts.md](image-prompts.md) | Image prompts: the `(tag:1.3)` / `[tag]` / `BREAK` grammar, prompts past CLIP's 77 tokens, `clip_skip`, negative prompts |
 | [zimage-lora-training.md](zimage-lora-training.md) | LoRA training for image models |
 | [dtype-auto-selection.md](dtype-auto-selection.md) | How model dtype is chosen automatically |
 

docs/image-prompts.md

+# Image prompts: weights, BREAK, long prompts and CLIP skip
+
+A CLIP text encoder reads 77 tokens, two of which are the start and end
+markers. That limit is the reason the Stable Diffusion ecosystem grew a prompt
+syntax of its own, and the reason a server that ignores it renders something
+other than what was asked for — without failing, which is the awkward part.
+
+CoderAI encodes image prompts itself rather than handing the text to the
+pipeline. Three things follow from that, on every CLIP-family image model
+(SD 1.x, SD 2.x, SDXL and their fine-tunes):
+
+* **No 77-token limit.** The prompt is split into as many 75-token windows as
+  it needs, each encoded and concatenated. Nothing is truncated and nothing is
+  dropped quietly.
+* **Weights.** `(tag:1.3)` emphasises, `[tag]` attenuates, bare `(tag)` is
+  1.1, nesting multiplies, and `\(` is a literal bracket.
+* **BREAK.** Starts the next window, so tags describing different subjects are
+  not blended by sharing one.
+
+And `clip_skip`, which decides how deep in CLIP the embedding is read from.
+
+## Why it matters
+
+The booru-tag fine-tune families — Pony Diffusion, Illustrious, NoobAI and the
+hundreds of models derived from them — are trained on prompts that open with a
+run of quality tags and are read at `clip_skip` 2:
+
+```
+score_9, score_8_up, score_7_up, score_6_up, rating_safe, source_anime,
+(masterpiece:1.2), 1girl, red hair, standing in a field of wheat, golden hour,
+(detailed background:1.3) BREAK
+wide shot, depth of field, film grain
+```
+
+Against a server that passes that text straight to a pipeline, three separate
+things go wrong at once: the prompt is cut off partway through the quality tags
+so the subject is never seen; the brackets and colons arrive as literal tokens,
+i.e. as noise; and the embedding is read from the wrong layer, which washes the
+output out. All three answer `200` with an image, which is why they are easy to
+miss and worth stating.
+
+## Requests
+
+`POST /v1/images/generations`, `/v1/images/edits` and `/v1/images/inpaint` all
+take:
+
+| Field | Meaning |
+|---|---|
+| `negative_prompt` | What to avoid. The counterpart of the quality tags (`score_4, score_5, score_6, worst quality`) and as load-bearing as the prompt itself. |
+| `clip_skip` | Read the embedding this many CLIP layers from the end. Omitted uses the model's configured value, then the pipeline's own default. |
+| `prompt_weighting` | `false` keeps brackets literal, for a caller sending prose. The windowing stays on either way, so turning it off never costs tokens. |
+
+```json
+{
+  "model": "pony-xl",
+  "prompt": "score_9, score_8_up, (masterpiece:1.2), 1girl, red hair BREAK wide shot",
+  "negative_prompt": "score_4, score_5, score_6, worst quality, blurry",
+  "clip_skip": 2,
+  "size": "832x1216"
+}
+```
+
+## Per model, for clients that cannot ask
+
+A client speaking the plain OpenAI image API has no `clip_skip` field to send,
+and a model that needs 2 needs it on every request. So both settings are part
+of the model's own configuration — **Models → Prompt Handling** in the admin
+UI, or `clip_skip` / `prompt_weighting` on the model's entry in `models.json`.
+A request that sends either one still wins; the config is the default for
+everyone who does not.
+
+There is also a server-wide default under `image` in `config.json`
+(`clip_skip`, `prompt_weighting`), which the model's entry overrides in turn.
+
+## GGUF models (sd.cpp)
+
+sd.cpp parses the weight grammar and manages the token windows itself, inside
+the engine, so a prompt reaches it as written and `clip_skip` is passed
+through. Until v0.4.89 the negative prompt was not: the call hardcoded an empty
+string, so for every GGUF image model whatever the caller asked to avoid was
+discarded between the API and the engine.
+
+## What is not covered
+
+Flux, SD3 and Z-Image carry a T5 encoder with a much longer limit and
+conventions of its own. CoderAI leaves those on their pipeline's native text
+path — the weight grammar is a CLIP-era convention and a T5 reads it as text,
+which is what those models expect. Negative prompts work there as the pipeline
+supports them.
+
+Textual-inversion embeddings (`easynegative` and friends) are not loaded by
+name yet; a LoRA is the supported way to add learned tokens
+(see [zimage-lora-training.md](zimage-lora-training.md)).
+
+## Implementation
+
+`codai/models/prompt_weighting.py` holds the parser, the windowing and the
+encoder, and documents how `clip_skip` maps to a hidden-state index — SDXL
+takes the penultimate layer by default and counts from there, SD 1.x takes the
+final output and needs the layer norm applied by hand. Getting that mapping
+wrong shifts every image, so it mirrors diffusers exactly.
+`tests/test_prompt_weighting_and_clip_skip.py` pins the grammar table, the
+window arithmetic, both layer conventions and the three defects above.

tests/test_prompt_weighting_and_clip_skip.py

+"""The prompt syntax the SD ecosystem speaks, and the 77-token wall.
+
+Three defects are pinned here, all of them silent -- each one answered 200 with
+a plausible-looking image:
+
+1. A prompt longer than 77 CLIP tokens was truncated by the tokenizer and the
+   rest was never seen. The booru-tag fine-tune families (Pony, Illustrious,
+   NoobAI) open with a run of quality tags, so the part that got dropped was
+   the subject.
+2. ``(tag:1.3)``, ``[tag]`` and ``BREAK`` reached CLIP as literal text, i.e.
+   as noise, because nothing parsed them.
+3. The sd.cpp path hardcoded ``negative_prompt=''``: for every GGUF image
+   model the caller's negative prompt was discarded between the API and the
+   engine, which is half of the idiom gone.
+
+Plus ``clip_skip``, which those models are trained at and which an
+OpenAI-image-API client has no field to ask for -- so the model's own config
+has to be able to.
+"""
+import types
+
+import pytest
+
+from codai.models.prompt_weighting import (
+    BREAK,
+    WINDOW,
+    encode_weighted_prompt,
+    has_weighting_syntax,
+    parse_prompt_attention,
+    supports_weighted_embeds,
+    tokenize_weighted,
+    window_count,
+)
+
+
+# --------------------------------------------------------------------------- #
+#  the grammar
+# --------------------------------------------------------------------------- #
+
+@pytest.mark.parametrize("text,expected", [
+    ("a (b) c",      [["a ", 1.0], ["b", 1.1], [" c", 1.0]]),
+    ("(a:1.3)",      [["a", 1.3]]),
+    ("(a:0.5)",      [["a", 0.5]]),
+    ("[a]",          [["a", 1 / 1.1]]),
+    ("((a))",        [["a", 1.1 * 1.1]]),
+    ("[[a]]",        [["a", (1 / 1.1) ** 2]]),
+    ("",             [["", 1.0]]),
+])
+def test_the_a1111_grammar(text, expected):
+    got = parse_prompt_attention(text)
+    assert len(got) == len(expected)
+    for (gt, gw), (et, ew) in zip(got, expected):
+        assert gt == et
+        assert gw == pytest.approx(ew)
+
+
+def test_a_backslash_escapes_a_bracket():
+    """Prose with parentheses in it has an out, and it is the ecosystem's."""
+    assert parse_prompt_attention(r"a \(b\) c") == [["a (b) c", 1.0]]
+
+
+def test_an_unclosed_bracket_still_applies():
+    """What a user who forgot the closing bracket meant."""
+    assert parse_prompt_attention("a (b c") == [["a ", 1.0], ["b c", 1.1]]
+
+
+def test_break_is_a_marker_not_a_word():
+    parsed = parse_prompt_attention("a BREAK b")
+    assert [p[0] for p in parsed] == ["a", BREAK, "b"]
+    assert parsed[1][1] == -1
+
+
+def test_break_is_never_merged_into_the_text_around_it():
+    """Both neighbours weigh 1.0; merging equal weights must not swallow it."""
+    parsed = parse_prompt_attention("one BREAK two BREAK three")
+    assert [p[0] for p in parsed] == ["one", BREAK, "two", BREAK, "three"]
+
+
+def test_weights_nest_through_a_sentence():
+    parsed = parse_prompt_attention("score_9, (masterpiece:1.4), [blurry]")
+    weights = {t.strip(" ,"): w for t, w in parsed}
+    assert weights["score_9"] == pytest.approx(1.0)
+    assert weights["masterpiece"] == pytest.approx(1.4)
+    assert weights["blurry"] == pytest.approx(1 / 1.1)
+
+
+@pytest.mark.parametrize("text,flagged", [
+    ("a plain prompt", False),
+    ("score_9, score_8_up, a cat", False),
+    ("(a cat:1.2)", True),
+    ("[blurry]", True),
+    ("a cat BREAK a dog", True),
+    (r"a \(parenthetical\) aside", False),
+])
+def test_reporting_whether_the_grammar_was_used(text, flagged):
+    assert has_weighting_syntax(text) is flagged
+
+
+# --------------------------------------------------------------------------- #
+#  windows: the 77-token wall
+# --------------------------------------------------------------------------- #
+
+class _FakeTokenizer:
+    """One id per whitespace-separated word, which makes token counts countable."""
+    bos_token_id = 49406
+    eos_token_id = 49407
+    pad_token_id = 49407
+
+    def __call__(self, text, add_special_tokens=True, truncation=True, **kw):
+        assert add_special_tokens is False, "the markers belong to the window"
+        assert truncation is False, "truncating here is the whole defect"
+        ids = [1000 + (hash(w) % 1000) for w in str(text).split()]
+        return types.SimpleNamespace(input_ids=ids)
+
+
+def test_a_short_prompt_is_one_window():
+    tok = _FakeTokenizer()
+    windows, weights = tokenize_weighted(tok, "a cat on a mat")
+    assert len(windows) == 1
+    assert len(windows[0]) == WINDOW == len(weights[0])
+
+
+def test_a_long_prompt_is_not_truncated_but_windowed():
+    """The regression: 200 tokens used to become 75 and nobody was told."""
+    tok = _FakeTokenizer()
+    prompt = " ".join(f"tag{i}" for i in range(200))
+    windows, weights = tokenize_weighted(tok, prompt)
+    assert len(windows) == 3                     # 75 + 75 + 50 padded
+    assert window_count(tok, prompt) == 3
+    assert all(len(w) == WINDOW for w in windows)
+    # every one of the 200 tokens is somewhere in those windows
+    flat = [t for w in windows for t in w]
+    assert sum(1 for t in flat if t != tok.pad_token_id) == 200
+
+
+def test_break_closes_the_window_early():
+    """Two tags separated by BREAK do not share a window, which is the point."""
+    tok = _FakeTokenizer()
+    windows, _ = tokenize_weighted(tok, "cat BREAK dog")
+    assert len(windows) == 2
+    assert windows[0][1] == tok.pad_token_id     # padded out after one token
+    assert windows[1][1] == tok.pad_token_id
+
+
+def test_weights_line_up_with_their_tokens():
+    tok = _FakeTokenizer()
+    windows, weights = tokenize_weighted(tok, "plain (heavy:2.0) plain2")
+    assert weights[0][0] == pytest.approx(1.0)
+    assert weights[0][1] == pytest.approx(2.0)
+    assert weights[0][2] == pytest.approx(1.0)
+    # the padding weighs nothing in particular
+    assert weights[0][3] == pytest.approx(1.0)
+
+
+def test_plain_mode_keeps_the_windows_and_drops_the_grammar():
+    """`prompt_weighting: false` is an escape hatch, not a return to truncation."""
+    tok = _FakeTokenizer()
+    prompt = " ".join(f"tag{i}" for i in range(100)) + " (heavy:2.0)"
+    windows, weights = tokenize_weighted(tok, prompt, plain=True)
+    assert len(windows) == 2
+    assert all(w == 1.0 for win in weights for w in win)
+
+
+# --------------------------------------------------------------------------- #
+#  encoding
+# --------------------------------------------------------------------------- #
+
+HIDDEN = 8
+HIDDEN_2 = 12
+LAYERS = 6
+
+
+class _Out:
+    """What a CLIP text encoder returns, indexable the way diffusers reads it."""
+
+    def __init__(self, hidden_states, last_hidden_state, text_embeds=None):
+        self.hidden_states = hidden_states
+        self.last_hidden_state = last_hidden_state
+        if text_embeds is not None:
+            self.text_embeds = text_embeds
+        self._first = last_hidden_state if text_embeds is None else text_embeds
+
+    def __getitem__(self, i):
+        return self._first if i == 0 else self.hidden_states[i]
+
+
+class CLIPTextModel:
+    """A text encoder whose every layer is identifiable by its own constant."""
+
+    projection = False
+
+    def __init__(self, hidden_size):
+        import torch
+        self.hidden_size = hidden_size
+        self.dtype = torch.float32
+        # A real final_layer_norm would change the values; marking them is
+        # enough to prove it was applied on the SD 1.x clip_skip path.
+        self.text_model = types.SimpleNamespace(
+            final_layer_norm=lambda t: t * 100.0)
+
+    def __call__(self, ids, output_hidden_states=False, **kw):
+        import torch
+        n = ids.shape[1]
+        # hidden_states[i] is filled with i, so which layer was taken is
+        # readable straight off the returned tensor.
+        hs = tuple(torch.full((1, n, self.hidden_size), float(i))
+                   for i in range(LAYERS))
+        last = torch.full((1, n, self.hidden_size), -1.0)
+        pooled = (torch.full((1, self.hidden_size), 7.0)
+                  if self.projection else None)
+        return _Out(hs, last, pooled)
+
+
+class CLIPTextModelWithProjection(CLIPTextModel):
+    """SDXL's second encoder: output[0] is the pooled vector, not the states."""
+
+    projection = True
+
+
+class T5EncoderModel:
+    """What Flux and SD3 carry instead, with its own limit and conventions."""
+
+
+def _fake_sd15_pipeline():
+    pipe = types.SimpleNamespace()
+    pipe.tokenizer = _FakeTokenizer()
+    pipe.text_encoder = CLIPTextModel(HIDDEN)
+    pipe.device = "cpu"
+    return pipe
+
+
+def _fake_sdxl_pipeline():
+    pipe = _fake_sd15_pipeline()
+    pipe.tokenizer_2 = _FakeTokenizer()
+    pipe.text_encoder_2 = CLIPTextModelWithProjection(HIDDEN_2)
+    return pipe
+
+
+def test_sd15_is_supported_and_sdxl_is_supported():
+    assert supports_weighted_embeds(_fake_sd15_pipeline())
+    assert supports_weighted_embeds(_fake_sdxl_pipeline())
+
+
+def test_a_t5_second_encoder_is_left_alone():
+    """Flux and SD3 carry a T5 with its own limit and conventions."""
+    pipe = _fake_sdxl_pipeline()
+    pipe.text_encoder_2 = T5EncoderModel()
+    assert not supports_weighted_embeds(pipe)
+    assert encode_weighted_prompt(pipe, "a cat") is None
+
+
+def test_a_pipeline_with_no_text_encoder_is_left_alone():
+    assert not supports_weighted_embeds(types.SimpleNamespace())
+
+
+def test_sd15_shape_grows_with_the_prompt():
+    """77 tokens of room per window, concatenated -- not a cap."""
+    pipe = _fake_sd15_pipeline()
+    long_prompt = " ".join(f"tag{i}" for i in range(200))
+    out = encode_weighted_prompt(pipe, long_prompt, "")
+    assert out["prompt_embeds"].shape == (1, 3 * (WINDOW + 2), HIDDEN)
+
+
+def test_the_negative_is_padded_to_the_same_length():
+    """Guidance subtracts one from the other, so they must match."""
+    pipe = _fake_sd15_pipeline()
+    out = encode_weighted_prompt(
+        pipe, " ".join(f"tag{i}" for i in range(200)), "blurry")
+    assert out["prompt_embeds"].shape == out["negative_prompt_embeds"].shape
+
+
+def test_sd15_clip_skip_picks_the_layer_and_normalises_it():
+    pipe = _fake_sd15_pipeline()
+    plain = encode_weighted_prompt(pipe, "a cat", "")
+    # No clip_skip: the encoder's final output (marked -1), untouched.
+    assert plain["prompt_embeds"].flatten()[0].item() == pytest.approx(-1.0)
+    # clip_skip=2 on SD 1.x: hidden_states[-3] (layer 3 of 6), then the final
+    # layer norm, which the fake multiplies by 100.
+    skipped = encode_weighted_prompt(pipe, "a cat", "", clip_skip=2)
+    assert skipped["prompt_embeds"].flatten()[0].item() == pytest.approx(300.0)
+
+
+def test_sdxl_counts_clip_skip_from_the_penultimate_layer():
+    """SDXL pipelines always take hidden_states[-2], so a clip_skip there is
+    relative to that -- getting this wrong shifts every image."""
+    pipe = _fake_sdxl_pipeline()
+    default = encode_weighted_prompt(pipe, "a cat", "")
+    assert default["prompt_embeds"].flatten()[0].item() == pytest.approx(4.0)  # hs[-2]
+    skipped = encode_weighted_prompt(pipe, "a cat", "", clip_skip=2)
+    assert skipped["prompt_embeds"].flatten()[0].item() == pytest.approx(2.0)  # hs[-4]
+
+
+def test_sdxl_concatenates_both_encoders_and_pools_the_second():
+    pipe = _fake_sdxl_pipeline()
+    out = encode_weighted_prompt(pipe, "a cat", "blurry")
+    assert out["prompt_embeds"].shape == (1, WINDOW + 2, HIDDEN + HIDDEN_2)
+    assert out["pooled_prompt_embeds"].shape == (1, HIDDEN_2)
+    assert out["negative_pooled_prompt_embeds"].shape == (1, HIDDEN_2)
+
+
+def test_a_weight_changes_the_tokens_but_not_the_window_mean():
+    """Scaling alone would shift the whole window's magnitude, which shows up
+    as a brightness change that nobody asked for."""
+    import torch
+    pipe = _fake_sd15_pipeline()
+    flat = encode_weighted_prompt(pipe, "cat dog", "")["prompt_embeds"]
+    weighted = encode_weighted_prompt(pipe, "(cat:1.5) dog", "")["prompt_embeds"]
+    assert not torch.allclose(flat, weighted)
+    assert weighted.mean().item() == pytest.approx(flat.mean().item(), rel=1e-4)
+
+
+def test_plain_mode_ignores_the_weights_end_to_end():
+    import torch
+    pipe = _fake_sd15_pipeline()
+    a = encode_weighted_prompt(pipe, "(cat:1.5) dog", "", plain=True)["prompt_embeds"]
+    b = encode_weighted_prompt(pipe, "cat dog", "", plain=True)["prompt_embeds"]
+    # Same token count either way, and no scaling applied.
+    assert a.shape == b.shape
+    assert torch.allclose(a, b)
+
+
+# --------------------------------------------------------------------------- #
+#  the API: what a request can ask for, and what reaches the engine
+# --------------------------------------------------------------------------- #
+
+def test_the_image_request_takes_clip_skip_and_weighting():
+    from codai.pydantic.imagerequest import ImageGenerationRequest
+    r = ImageGenerationRequest(model="m", prompt="p", clip_skip=2,
+                               prompt_weighting=False, negative_prompt="score_4")
+    assert r.clip_skip == 2
+    assert r.prompt_weighting is False
+    assert r.negative_prompt == "score_4"
+
+
+def test_defaults_leave_every_decision_to_the_model():
+    from codai.pydantic.imagerequest import ImageGenerationRequest
+    r = ImageGenerationRequest(model="m", prompt="p")
+    assert r.clip_skip is None and r.prompt_weighting is None
+
+
+@pytest.mark.parametrize("cls_name", ["ImageEditRequest", "ImageInpaintRequest"])
+def test_the_edit_endpoints_accept_a_negative_prompt(cls_name):
+    """They had no field for one, so an edit could only be steered positively."""
+    from codai.api import images as images_mod
+    cls = getattr(images_mod, cls_name)
+    kw = {"model": "m", "prompt": "p", "image": "data:,x"}
+    if cls_name == "ImageInpaintRequest":
+        kw["mask"] = "data:,y"
+    r = cls(negative_prompt="score_4, blurry", clip_skip=2, **kw)
+    assert r.negative_prompt == "score_4, blurry"
+    assert r.clip_skip == 2
+    assert r.prompt_weighting is None
+
+
+def test_the_sdcpp_path_no_longer_discards_the_negative_prompt():
+    """The defect, as source: `negative_prompt=''` passed to the engine."""
+    src = open("codai/api/images.py").read()
+    assert "negative_prompt=''" not in src
+    assert 'negative_prompt=""' not in src
+
+
+def test_resolve_clip_skip_precedence():
+    from codai.api import images as images_mod
+    req = types.SimpleNamespace(clip_skip=3)
+    model = {"clip_skip": 2}
+    args = types.SimpleNamespace(image_clip_skip=4)
+    assert images_mod._resolve_clip_skip(req, model, args) == 3
+    assert images_mod._resolve_clip_skip(
+        types.SimpleNamespace(clip_skip=None), model, args) == 2
+    assert images_mod._resolve_clip_skip(
+        types.SimpleNamespace(clip_skip=None), {}, args) == 4
+    assert images_mod._resolve_clip_skip(
+        types.SimpleNamespace(clip_skip=None), {}, types.SimpleNamespace()) is None
+
+
+@pytest.mark.parametrize("value", [0, -1, "", "two", None])
+def test_a_meaningless_clip_skip_means_the_default(value):
+    from codai.api import images as images_mod
+    assert images_mod._resolve_clip_skip(
+        types.SimpleNamespace(clip_skip=value), {}, types.SimpleNamespace()) is None
+
+
+def test_resolve_prompt_weighting_precedence():
+    from codai.api import images as images_mod
+    none = types.SimpleNamespace(prompt_weighting=None)
+    assert images_mod._resolve_prompt_weighting(
+        types.SimpleNamespace(prompt_weighting=False), {"prompt_weighting": True},
+        types.SimpleNamespace()) is False
+    assert images_mod._resolve_prompt_weighting(
+        none, {"prompt_weighting": False}, types.SimpleNamespace()) is False
+    assert images_mod._resolve_prompt_weighting(
+        none, {}, types.SimpleNamespace(image_prompt_weighting=False)) is False
+    # nothing says otherwise: on, as the ecosystem assumes
+    assert images_mod._resolve_prompt_weighting(
+        none, {}, types.SimpleNamespace()) is True
+
+
+def test_clip_skip_is_part_of_the_embedding_cache_key():
+    """Embeddings taken from different layers are different embeddings: sharing
+    one cache entry would serve the first request's layer to all the rest."""
+    from codai.api.images import _PromptEmbedCache
+    c = _PromptEmbedCache()
+    c.put("p", "n", "model#1|cs=2|w=1", {"prompt_embeds": "a"})
+    assert c.get("p", "n", "model#1|cs=2|w=1") is not None
+    assert c.get("p", "n", "model#1|cs=None|w=1") is None
+    assert c.get("p", "n", "model#1|cs=2|w=0") is None
+
+
+def test_both_knobs_are_configurable_per_model():
+    """An OpenAI-API client cannot send clip_skip, so the model must carry it."""
+    src = open("codai/admin/routes.py").read()
+    assert '"clip_skip", "prompt_weighting",' in src
+    ui = open("codai/admin/templates/models.html").read()
+    assert "cfg-clip-skip" in ui and "cfg-prompt-weighting" in ui
+
+
+def test_the_server_wide_defaults_round_trip_through_config():
+    from codai.config import ImageConfig
+    cfg = ImageConfig()
+    assert cfg.clip_skip is None
+    assert cfg.prompt_weighting is True
+
+
+# --------------------------------------------------------------------------- #
+#  the compatibility guarantee
+# --------------------------------------------------------------------------- #
+
+def test_a_short_plain_prompt_encodes_exactly_as_diffusers_did():
+    """Nothing that worked before renders differently now.
+
+    This is the whole reason the windowing is safe to turn on for everyone: for
+    a prompt that fits, encoding it ourselves has to be *bit-identical* to what
+    the pipeline produced, or every existing caller's images change under them.
+
+    It is identical for a reason worth recording. diffusers tokenizes with
+    ``padding="max_length"``, giving ``[bos] + tokens + [eos] + [pad]...``;
+    these windows are ``[bos] + tokens + [pad]... + [eos]``. For a CLIP
+    tokenizer the pad token *is* the end-of-text token, so the two id
+    sequences are the same -- and if a tokenizer ever turns up where they
+    differ, this test is what notices.
+
+    Run against a real ``CLIPTextModel`` (tiny, random weights, built from a
+    config so it needs no download), because the point is the model's own
+    attention over the padding, which a stub cannot stand in for.
+    """
+    import torch
+    from transformers import CLIPTextConfig, CLIPTextModel
+
+    cfg = CLIPTextConfig(hidden_size=32, intermediate_size=37,
+                         num_hidden_layers=5, num_attention_heads=4,
+                         vocab_size=1000, max_position_embeddings=77,
+                         bos_token_id=0, eos_token_id=999)
+    torch.manual_seed(0)
+    encoder = CLIPTextModel(cfg).eval()
+
+    class _ClipLikeTokenizer(_FakeTokenizer):
+        bos_token_id = 0
+        eos_token_id = 999
+        pad_token_id = 999          # CLIP pads with end-of-text, which is the key
+
+        def __call__(self, text, add_special_tokens=True, truncation=True, **kw):
+            ids = [1 + (abs(hash(w)) % 900) for w in str(text).split()]
+            return types.SimpleNamespace(input_ids=ids)
+
+    tok = _ClipLikeTokenizer()
+    pipe = types.SimpleNamespace(tokenizer=tok, text_encoder=encoder, device="cpu")
+    prompt = "a cat sitting on a mat"
+
+    mine = encode_weighted_prompt(pipe, prompt, "")["prompt_embeds"]
+
+    ids = tok(prompt, add_special_tokens=False, truncation=False).input_ids
+    as_diffusers_does = [tok.bos_token_id] + ids + \
+        [tok.pad_token_id] * (WINDOW + 2 - 1 - len(ids))
+    with torch.no_grad():
+        theirs = encoder(torch.tensor([as_diffusers_does]))[0]
+
+    assert mine.shape == theirs.shape == (1, WINDOW + 2, 32)
+    assert torch.equal(mine, theirs)