Preserve native token row when assembling semantic poles

Byte BPE can merge final native punctuation with appended newlines, changing an
existing token before any study encode. Keep the exact native prompt/template
and image row authoritative, then insert separately tokenized plain pole payloads
with explicitly owned delimiters before its original suffix.

Require deletion of every owned span to recover the original token objects,
weights, image handles and order. Reject special/media/role pole tokens and keep
native projection, neutral identity and existing node interfaces unchanged.

Validation: 181 host-free tests, required compile, exact saved-tokenizer source
bindings and real byte-BPE construction regressions. No post-fix model or visual
acceptance is claimed.

docs/krea-v12-native-studies.md

@@ -47,8 +47,16 @@ under one identity/revision reject the whole study request.
 
 ## Preparation and ownership
 
-Positive studies append literal pole text before tokenization and use one native
-image/template sequence. The implementation resolves token coordinates,
+Positive studies preserve one exact native image/template/prompt token row.
+Each literal pole is tokenized as text only; its nonempty payload is inserted
+with two explicitly owned Qwen newline tokens before the original user-content
+suffix. The original prompt is never retokenized with appended text: byte BPE
+can otherwise merge a final period with added newlines and change native tokens.
+All pole spans, including their delimiters, are owned before a combined encode.
+Deleting every owned span must recover the original token objects, weights,
+image handles/order and suffix exactly. Pole payloads containing special/control
+or media tokens reject. This is token-level assembly before encoding, not a join
+of finished encodes. The implementation resolves token coordinates,
 expanded input coordinates and coordinates after native template stripping.
 Image indices and extents must match the host's reported preparation. Every
 encode verifies output shape and attention-mask extent. Image gains, when used

kg_krea_v12/ledger.py

@@ -85,6 +85,44 @@ class TokenLedger:
     poles: tuple
     strip: int
 
+    @classmethod
+    def assemble(cls, original_tokens, owners, pole_tokens):
+        """Own added token spans without retokenizing any native prompt prefix.
+
+        Byte BPE may merge original punctuation with appended newlines. Retain
+        the original native row exactly and insert separately tokenized literal
+        payloads before its original suffix. Delimiters are explicitly owned
+        newline tokens, not a requirement that whole-text BPE preserve a prefix.
+        This assembles tokens before encoding, never completed conditionings.
+        """
+        original = cls.build(original_tokens, owners)
+        if not isinstance(pole_tokens, (list, tuple)) or not 1 <= len(pole_tokens) <= 16:
+            raise ValueError('V12 requires one through sixteen native pole payload rows')
+        _, content_end = content_bounds(original.row)
+        additions, spans = [], []
+        for tokens in pole_tokens:
+            row = single_row(tokens)
+            start, end = content_bounds(row)
+            payload = row[start:end]
+            if (not payload or any(isinstance(element(item), dict) for item in row)
+                    or any(not 0 <= element(item) < 151643 for item in payload)):
+                raise RuntimeError('V12 pole payload must be nonempty plain text without native control/media tokens')
+            begin = content_end + len(additions)
+            # Qwen newline198 is also verified in the native user-opening row.
+            # Both separators and payload are owned/muted/projected as one span.
+            additions.extend(((198, 1.0), (198, 1.0)))
+            additions.extend(payload)
+            spans.append((begin, content_end + len(additions)))
+        row = original.row[:content_end] + tuple(additions) + original.row[content_end:]
+        if content_bounds(row) != (original.strip, content_end + len(additions)):
+            raise RuntimeError('V12 pole assembly changed native template ownership')
+        retained = tuple(item for index, item in enumerate(row)
+                         if not any(start <= index < end for start, end in spans))
+        if (len(retained) != len(original.row)
+                or any(item is not native_item for item, native_item in zip(retained, original.row))):
+            raise RuntimeError('V12 removing owned pole spans did not recover the exact native row')
+        return cls(row, original.owners, original.prompt, tuple(spans), original.strip)
+
     @classmethod
     def build(cls, tokens, owners, ladder=()):
         row = single_row(tokens)

kg_krea_v12/studies.py

@@ -139,12 +139,13 @@ def build_studies(context, cards, split=0.4, budget=2.0, max_passes=20, max_toke
             pixels += shape[1] * shape[2]
         if pixels > max_image_pixels:
             raise RuntimeError('V12 prepared-image pixel budget exceeded before vision encoding')
-        texts = [context.prompt]
-        for group in groups:
-            for pole in (group['plus'], group['minus']):
-                texts.append(texts[-1] + '\n\n' + pole)
-        rows = [prepared.tokenize(context.clip, text) for text in texts]
-        ledger = TokenLedger.build(rows[-1], prepared.owners, rows[:-1])
+        original_tokens = prepared.tokenize(context.clip, context.prompt)
+        # Keep the native image/template/prompt row authoritative. Tokenize only
+        # each literal text payload with the native text tokenizer; do not
+        # retokenize an ever-growing prompt or prepare its images repeatedly.
+        pole_tokens = [context.clip.tokenize(pole) for group in groups
+                       for pole in (group['plus'], group['minus'])]
+        ledger = TokenLedger.assemble(original_tokens, prepared.owners, pole_tokens)
         if len(ledger.row) > max_tokens:
             raise RuntimeError('V12 token-row budget exceeded before encoding')
         tokens = ledger.tagged_tokens()
@@ -159,7 +160,7 @@ def build_studies(context, cards, split=0.4, budget=2.0, max_passes=20, max_toke
             _validate_output(result, expanded)
             if observed and expanded != observed[0]:
                 raise RuntimeError('V12 native preparation ledger changed between studies')
-            positions = ledger.native_output_positions(rows[0], expanded)
+            positions = ledger.native_output_positions(original_tokens, expanded)
             if retained_positions and positions != retained_positions[0]:
                 raise RuntimeError('V12 native output projection changed between studies')
             retained_positions.append(positions)

tests/test_krea_v12_token_assembly.py

@@ -0,0 +1,154 @@
+"""Actual Qwen byte-BPE rows; encoder/vision outputs remain authored scalar fixtures."""
+import json
+from pathlib import Path
+from types import SimpleNamespace as NS
+import unittest
+from unittest.mock import patch
+
+import test_krea_v12 as fixture
+from native_v12_tests import hooks, ledger, native, studies
+
+DATA = json.loads((Path(__file__).parent / 'fixtures/krea_v12_bpe_boundary.json').read_text(encoding='utf-8'))
+weighted = lambda values: {ledger.TOKEN_KEY: [[(value, 1.0) for value in values]]}
+
+
+class BPEClip(fixture.Clip):
+    """Use captured exact byte-BPE rows, not a character/word tokenizer stub."""
+    def __init__(self, record):
+        super().__init__()
+        self.record, self.raw_inputs = record, []
+
+    def tokenize(self, text, images=(), llama_template=None):
+        self.tokenizations += 1
+        self.raw_inputs.append((text, tuple(images), llama_template))
+        if text in (self.record['plus'], self.record['minus']):
+            if images or llama_template is not None:
+                raise AssertionError('pole tokenization must be text-only')
+            key = 'plus_native' if text == self.record['plus'] else 'minus_native'
+            ids = self.record[key]
+        elif text == self.record['prompt'] or text == '':
+            if images:
+                if llama_template != DATA['ground_templates'][str(len(images))]:
+                    raise AssertionError('actual native grounding template changed')
+                ids = (self.record['grounded'] if text else DATA['negative_grounded'])[str(len(images))]
+            else:
+                ids = self.record['native'] if text else DATA['negative']
+        else:
+            raise AssertionError('native prompt/pole text was unexpectedly changed or concatenated')
+        image_iter = iter(images)
+        row = [(dict(type='image', data=next(image_iter), original_type='image') if value == 151655 else value, 1.0)
+               for value in ids]
+        return {ledger.TOKEN_KEY: [row]}
+
+
+class BPEGrounder(fixture.Grounder):
+    def _template(self, count, system):
+        if system:
+            raise AssertionError('this fixture binds the original native default')
+        return DATA['ground_templates'][str(count)]
+
+
+def make_context(record, image_count=0):
+    clip, grounder = BPEClip(record), BPEGrounder()
+    source = fixture.Image(7, 3) if image_count else None
+    donor = fixture.Image(11, 5) if image_count == 2 else None
+    model = NS(patches_uuid='native-bpe-fixture-v1')
+    target = {'samples': object(), 'noise_mask': object()}
+    with patch.object(native, 'load_grounder', return_value=grounder):
+        context = native.prepare_context(clip, model, record['prompt'], source=source, donor=donor,
+            source_latent=object(), target_latent=target, revision='bpe-row-fixture-v1')
+    return context, grounder
+
+
+class TokenAssemblyTests(unittest.TestCase):
+    def test_actual_failing_qwen_period_boundary_is_reproduced_without_models(self):
+        record = DATA['cases'][0]
+        with self.assertRaisesRegex(RuntimeError, 'boundary merged'):
+            ledger.TokenLedger.build(weighted(record['grown']), (), [weighted(record['native'])])
+        point = DATA['mismatch']['index']
+        self.assertEqual(record['native'][point], 13)
+        self.assertEqual(record['grown'][point], 382)
+        self.assertEqual(DATA['mismatch']['grown_piece'], '.\n\n')
+
+    def test_multiple_owned_poles_recover_exact_bpe_native_row_and_weights(self):
+        for record in DATA['cases']:
+            with self.subTest(prompt=record['prompt']):
+                original = weighted(record['native'])
+                base = ledger.single_row(original)
+                result = ledger.TokenLedger.assemble(original, (), [weighted(record['plus_native']), weighted(record['minus_native'])] * 2)
+                self.assertEqual(len(result.poles), 4)
+                kept = [item for index, item in enumerate(result.row)
+                        if not any(a <= index < b for a, b in result.poles)]
+                self.assertEqual(len(kept), len(base))
+                for left, right in zip(kept, base):
+                    self.assertIs(left, right)
+                self.assertTrue(all(item[1] == 1.0 for item in result.row))
+
+    def test_source_and_donor_identity_order_and_metadata_survive_assembly(self):
+        for count in (1, 2):
+            context, grounder = make_context(DATA['cases'][0], count)
+            with patch.object(native, 'load_grounder', return_value=grounder):
+                prepared = context.prepare_inputs()
+                original = prepared.tokenize(context.clip, context.prompt)
+                result = ledger.TokenLedger.assemble(original, prepared.owners,
+                    [context.clip.tokenize(DATA['cases'][0]['plus']), context.clip.tokenize(DATA['cases'][0]['minus'])])
+            images = [item for item in ledger.single_row(original) if isinstance(ledger.element(item), dict)]
+            actual = [item for item in result.row if isinstance(ledger.element(item), dict)]
+            self.assertEqual(len(actual), count)
+            for left, right in zip(actual, images):
+                self.assertIs(left, right)
+                self.assertEqual(left[0]['original_type'], 'image')
+            self.assertIs(actual[0][0]['data'], context.sources[0])
+            if count == 2:
+                self.assertIs(actual[1][0]['data'], context.sources[1])
+            self.assertNotIn(ledger.OWNER_KEY, actual[0][0])
+
+    def test_special_media_role_and_weighted_pole_payloads_reject(self):
+        record = DATA['cases'][0]
+        original = weighted(record['native'])
+        for token in (151643, 151644, 151645, 151652, 151653, 151655, 151665, 200000):
+            payload = weighted(record['plus_native'])
+            start, _ = ledger.content_bounds(ledger.single_row(payload))
+            payload[ledger.TOKEN_KEY][0].insert(start, (token, 1.0))
+            with self.subTest(token=token), self.assertRaises(RuntimeError):
+                ledger.TokenLedger.assemble(original, (), [payload])
+        payload = weighted(record['plus_native'])
+        start, _ = ledger.content_bounds(ledger.single_row(payload))
+        payload[ledger.TOKEN_KEY][0][start] = (payload[ledger.TOKEN_KEY][0][start][0], 2.0)
+        with self.assertRaisesRegex(RuntimeError, 'unweighted'):
+            ledger.TokenLedger.assemble(original, (), [payload])
+        payload = weighted(record['plus_native'])
+        payload[ledger.TOKEN_KEY][0].insert(start, (dict(type='image', data=object()), 1.0))
+        with self.assertRaisesRegex(RuntimeError, 'plain text'):
+            ledger.TokenLedger.assemble(original, (), [payload])
+
+    def test_complete_studies_use_single_native_bpe_image_row_and_project_only_owned_tokens(self):
+        record = DATA['cases'][0]
+        for count in (0, 1, 2):
+            with self.subTest(image_count=count):
+                context, grounder = make_context(record, count)
+                before_image_calls = sum(bool(images) for _, images, _ in context.clip.raw_inputs)
+                with patch.object(native, 'load_grounder', return_value=grounder):
+                    result = studies.build_studies(context,
+                        [fixture.card(1, plus=record['plus'], minus=record['minus']),
+                         fixture.card(-0.5, axis='another', plus=record['plus'], minus=record['minus'])])
+                self.assertEqual(result.encode_passes, 5)
+                self.assertEqual(sum(bool(images) for _, images, _ in context.clip.raw_inputs) - before_image_calls, int(count > 0))
+                self.assertIs(result.phases[1]['negative'], context.native['negative'])
+                self.assertIs(result.phases[1]['target_latent'], context.native['target_latent'])
+                self.assertIs(result.phases[1]['noise_mask'], context.native['noise_mask'])
+                self.assertEqual(result.phases[1]['positive'][0][0].shape, context.native['positive'][0][0].shape)
+                self.assertEqual(len(result.retained_output_positions), context.native['positive'][0][0].shape[1])
+                self.assertFalse(hooks._busy)
+                self.assertNotIn('process_tokens', vars(context.clip.encoder))
+
+    def test_bpe_neutral_does_not_tokenize_prepare_or_encode_studies(self):
+        context, grounder = make_context(DATA['cases'][0], 2)
+        before = context.clip.tokenizations, context.clip.encodes, len(grounder.preps)
+        result = studies.build_studies(context, [fixture.card(0)])
+        self.assertIs(result.sampler_conditioning(), context.native)
+        self.assertEqual(before, (context.clip.tokenizations, context.clip.encodes, len(grounder.preps)))
+
+
+if __name__ == '__main__':
+    unittest.main()