Fix malformed block parsing in reference pipelines

Malformed or truncated blocks could contaminate following tables, and mismatched replacement indices could raise StopIteration. Recover completed text independently of bad coordinates, preserve raw fragments, and isolate malformed pages and crops. Validated with 10 regression tests, a matched 2,078-page replay, and fresh single/double/triple inference on 16 pages using one H100.

reference/parsebench/pipeline/PACKAGE_MANIFEST.json

@@ -74,28 +74,28 @@
       "sha256": "3d92146e6c713b9de67cb71e92fe254a1344d9054c3925a85cb778ec68c0d075"
     },
     "engine/geometry.py": {
-      "bytes": 13130,
-      "sha256": "76671df686b574a827205acb757eecf702e970470ffa15f81b3129cbcd95e242"
+      "bytes": 12991,
+      "sha256": "ea542d276bf66ed2fb4cc556e0154ac7de95f90d27701cb79b973b9d3d154a92"
     },
     "engine/header_inference_probe.py": {
       "bytes": 2498,
       "sha256": "02950f1aec5e2dc88a96487bd9124c97e5ce2842fbcfc5264ac2d5f697865722"
     },
     "engine/markdown.py": {
-      "bytes": 3557,
-      "sha256": "fc57566255ac632b24825973d9533a13fd717dad4e38d096098bac33ecc79957"
+      "bytes": 3834,
+      "sha256": "308ac27ec325cf7d5f4b5ee11341316e403eabb0773d73e97330e268f4573a5c"
     },
     "engine/merge.py": {
-      "bytes": 20084,
-      "sha256": "b1c2e749e8f772b716c5a8a2ebeba7348474eb9368b00d37f1c8987e26d24e4e"
+      "bytes": 20206,
+      "sha256": "e47d8a21cb53f763d7d26cd27496be07e5e9e4ee346c5915037ef066ff3891d8"
     },
     "engine/navit_geometry.py": {
       "bytes": 13519,
       "sha256": "469c05b5afa19ed5669c8866185a1a149846601c6e45f6c002d55ae80cb6424b"
     },
     "engine/policies.py": {
-      "bytes": 8930,
-      "sha256": "4c1f78be4d8061e4224fb59c3dc5342dfc21d9d37826814ea3573f339777aebd"
+      "bytes": 9212,
+      "sha256": "0115543cfa9656520a858030022820998b0d18cdf2883006fe14c97da27039bb"
     },
     "engine/regions.py": {
       "bytes": 866,
@@ -114,8 +114,8 @@
       "sha256": "525664ca0359029efe8c010ea8a04b1914fb42e7286e63f0cbe30c9b026fac84"
     },
     "postprocess.py": {
-      "bytes": 4211,
-      "sha256": "576db3eb97c5e43ebbb1b5cdd11aef212f7728fca0b4bfa20fc4d2f309e8eb5f"
+      "bytes": 3512,
+      "sha256": "3529d8d38163c2ddb1838fcbe6affb869034d2b7fd3e1ccad8cdb51788f3b5b1"
     },
     "postprocessing_checks.json": {
       "bytes": 3972,
@@ -156,6 +156,22 @@
     "validation_metrics.json": {
       "bytes": 14980,
       "sha256": "1f69fb9cc9860136e93c693f08d0d74f079ce550dc2a8cfbb82be0b4ead38554"
+    },
+    "parse_results.py": {
+      "bytes": 4585,
+      "sha256": "b5b996f374c701b3dd805fed486b00b64ada63b5816f4201380dee4e7f6e2e8f"
+    },
+    "tests/malformed_coordinate_regression.json": {
+      "bytes": 832,
+      "sha256": "46fdb8cdb0e091f1f3aa07b399c7bcb322e163340f0391c82c150055f78c5a97"
+    },
+    "tests/test_parser_general.py": {
+      "bytes": 5380,
+      "sha256": "82bc9366d634405402319c75eacb34468d3a4a05ce7a84e8dccd33962116d12b"
+    },
+    "engine/blocks.py": {
+      "bytes": 4547,
+      "sha256": "de94142d5442205c8a2c728137e1247de3660c04a73e393bda7b3f45975d993e"
     }
   }
 }

reference/parsebench/pipeline/engine/blocks.py

@@ -0,0 +1,108 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: OpenMDW-1.1
+"""Shared block grammar: an invalid block cannot consume the following block."""
+import re
+
+# A class terminator ends a block even when its coordinates are malformed.
+BODY = r'((?:(?!<class_|<x_[^<>]*><y_[^<>]*>).)*?)'
+NUMBER = r'(\d+(?:\.\d+)?)'
+BLOCK_RE = re.compile(
+    r'<x_' + NUMBER + r'><y_' + NUMBER + r'>' + BODY +
+    r'<x_' + NUMBER + r'><y_' + NUMBER + r'><class_([^>]+)>', re.DOTALL)
+TEXT_BLOCK_RE = re.compile(
+    r'<x_([^>]*)><y_([^>]*)>' + BODY +
+    r'<x_([^>]*)><y_([^>]*)><class_([^>]+)>', re.DOTALL)
+
+CLASS_RE = re.compile(r'<class_([^>]*)>')
+OPEN_PAIR_RE = re.compile(r'^\s*<x_[^<>]*><y_[^<>]*>')
+CLOSE_PAIR_RE = re.compile(r'<x_[^<>]*><y_[^<>]*>\s*$')
+PAIR_RE = re.compile(r'<x_[^<>]*><y_[^<>]*>')
+COORDINATE_RE = re.compile(r'<[xy]_[^<>]*(?:>|(?=<)|$)')
+CONTROL_RE = re.compile(r'</?s>|<pad>|<unk>|<image>|<tbc>|<\|[^>]+\|>|<(?:predict|output)_[^>]*>')
+
+
+def visible_text(raw):
+    """Remove serialization controls, including incomplete coordinate tokens."""
+    return CONTROL_RE.sub('', COORDINATE_RE.sub('', raw)).strip()
+
+
+def text_blocks(raw):
+    """Read text independently of whether coordinates are valid numerical values."""
+    cursor = 0
+    for match in CLASS_RE.finditer(raw or ''):
+        segment = raw[cursor:match.start()]
+        pairs = list(PAIR_RE.finditer(segment))
+        opening = pairs[-2] if len(pairs) >= 2 else None
+        closing = pairs[-1] if pairs and not segment[pairs[-1].end():].strip() else None
+        # A dangling preceding block must not become the body of a later,
+        # completed block (for example after inserting a valid crop result).
+        body = (segment[opening.end():closing.start()].strip()
+                if opening and closing and opening.end() <= closing.start()
+                else visible_text(segment))
+        if body:
+            yield match.group(1), body
+        cursor = match.end()
+    raw_tail = (raw or '')[cursor:]
+    tail = visible_text(raw_tail)
+    # An unfinished serialized block is retained in raw output, not presented as
+    # a successfully decoded text/chart/table block.
+    if tail and cursor == 0 and not any(token in raw_tail for token in ['<x_', '<y_']):
+        yield 'Text', tail
+
+
+def has_unparsed_content(raw):
+    cursor = 0
+    for match in BLOCK_RE.finditer(raw or ''):
+        if visible_text(raw[cursor:match.start()]) or COORDINATE_RE.search(raw[cursor:match.start()]):
+            return True
+        cursor = match.end()
+    gap=(raw or '')[cursor:]
+    return bool(visible_text(gap) or COORDINATE_RE.search(gap))
+
+
+def insert_preserving_unparsed(raw, direct_blocks, merged_blocks):
+    """Insert additions around validated blocks without discarding malformed text."""
+    matches = list(BLOCK_RE.finditer(raw))
+    positions = {id(block):match for block,match in zip(direct_blocks,matches)}
+    pieces, pending, cursor = [], [], 0
+    for block in merged_blocks:
+        match = positions.get(id(block))
+        if match is None:
+            pending.append(block.raw)
+            continue
+        pieces.append(raw[cursor:match.start()])
+        if pending:
+            pieces.append('\n\n'.join(pending)+'\n\n')
+            pending = []
+        pieces.append(raw[match.start():match.end()])
+        cursor = match.end()
+    pieces.append(raw[cursor:])
+    if pending:
+        pieces.append('\n\n'+'\n\n'.join(pending))
+    return ''.join(pieces)
+
+
+def incomplete_tail(raw):
+    matches=list(CLASS_RE.finditer(raw or ''))
+    tail=(raw or '')[matches[-1].end() if matches else 0:]
+    return bool(tail.strip()) and (bool(matches) or any(token in tail for token in ['<x_', '<y_']))
+
+
+def canonical_geometry(raw):
+    """Remove bad coordinate controls only from closed, unvalidated blocks.
+
+    Unfinished tails remain verbatim in raw output and receive a diagnostic.
+    """
+    import math
+    terminals=list(CLASS_RE.finditer(raw or ''))
+    boundary=terminals[-1].end() if terminals else 0
+    closed=(raw or '')[:boundary]
+    pieces,cursor=[],0
+    for match in BLOCK_RE.finditer(closed):
+        pieces.append(COORDINATE_RE.sub('',closed[cursor:match.start()]))
+        coords=[match[i] for i in [1,2,4,5]]
+        pieces.append(match[0] if all(math.isfinite(float(v)) for v in coords) else COORDINATE_RE.sub('',match[0]))
+        cursor=match.end()
+    pieces.append(COORDINATE_RE.sub('',closed[cursor:]))
+    pieces.append((raw or '')[boundary:])
+    return ''.join(pieces)

reference/parsebench/pipeline/engine/geometry.py

@@ -60,11 +60,7 @@ import numpy as np
 from PIL import Image
 
 
-BLOCK_RE = re.compile(
-    r"<x_(\d+(?:\.\d+)?)><y_(\d+(?:\.\d+)?)>(.*?)"
-    r"<x_(\d+(?:\.\d+)?)><y_(\d+(?:\.\d+)?)><class_([^>]+)>",
-    re.MULTILINE | re.DOTALL,
-)
+from .blocks import BLOCK_RE
 
 
 @dataclass(frozen=True)

reference/parsebench/pipeline/engine/markdown.py

@@ -61,10 +61,7 @@ TEXT_BEARING_CLASSES = {
 }
 
 
-_RE_EXTRACT_CLASS_BBOX = re.compile(
-    r"<x_(\d+(?:\.\d+)?)><y_(\d+(?:\.\d+)?)>(.*?)<x_(\d+(?:\.\d+)?)><y_(\d+(?:\.\d+)?)><class_([^>]+)>",
-    re.DOTALL,
-)
+from .blocks import BLOCK_RE as _RE_EXTRACT_CLASS_BBOX, text_blocks
 
 
 def extract_classes_bboxes(text: str) -> tuple[list[str], list[tuple[float, float, float, float]], list[str]]:
@@ -107,9 +104,20 @@ def _simple_latex_table_to_html(text: str) -> str:
     return pattern.sub(repl, text)
 
 
+# Text remains usable when a coordinate token is malformed. Keep numerical
+# coordinate validation in extract_classes_bboxes for callers that need geometry.
+
+
+
 def mmd_to_markdown(prediction: str, text_transform=None) -> str:
-    classes, _boxes, texts = extract_classes_bboxes(prediction or "")
+    parsed = list(text_blocks(prediction or ""))
+    classes = [cls for cls, _ in parsed]
+    texts = [text for _, text in parsed]
     if not texts:
+        # No class-terminated block was completed. Keep the original response
+        # available to callers instead of presenting a partial table as parsed.
+        if any(token in (prediction or "") for token in ['<x_', '<y_']):
+            return ""
         cleaned = re.sub(r"<x_[^>]+>|<y_[^>]+>|<class_[^>]+>", "", prediction or "")
         return convert_latex_tables_in_markdown(cleaned.replace("<tbc>", "").strip())
 
@@ -125,9 +133,6 @@ def mmd_to_markdown(prediction: str, text_transform=None) -> str:
             text = f"# {text}"
         elif cls == "Section-header" and not text.startswith("#"):
             text = f"## {text}"
-        elif cls not in TEXT_BEARING_CLASSES and cls not in {"Formula", "Picture"}:
-            continue
         blocks.append(text)
     return convert_latex_tables_in_markdown("\n\n".join(blocks).strip())
 
-

reference/parsebench/pipeline/engine/merge.py

@@ -30,11 +30,7 @@ from pathlib import Path
 from typing import Any
 
 
-BLOCK_RE = re.compile(
-    r"<x_(\d+(?:\.\d+)?)><y_(\d+(?:\.\d+)?)>(.*?)"
-    r"<x_(\d+(?:\.\d+)?)><y_(\d+(?:\.\d+)?)><class_([^>]+)>",
-    re.DOTALL,
-)
+from .blocks import BLOCK_RE, has_unparsed_content, insert_preserving_unparsed
 
 
 TOKEN_RE = re.compile(r"</?s>|<pad>|<unk>|<image>|<\|[^>]+\|>")
@@ -566,7 +562,10 @@ def build_direct_plus_ordered_structural_addonly(
             added.append(block)
         merged_blocks = insert_additions_preserving_direct_order(direct_blocks, added)
         row = clone_prediction(label, direct, source_name=f"direct-plus-{args.structural_source}-ordered-structural-addonly")
-        row["raw_out"] = rebuild_blocks(merged_blocks) or prediction_text(direct) or prediction_text(structural)
+        direct_raw = prediction_text(direct)
+        row["raw_out"] = (insert_preserving_unparsed(direct_raw, direct_blocks, merged_blocks)
+                          if has_unparsed_content(direct_raw)
+                          else rebuild_blocks(merged_blocks)) or direct_raw or prediction_text(structural)
         row["hybrid_kept_direct_blocks"] = len(direct_blocks)
         row["hybrid_added_structural_blocks"] = len(added)
         row["hybrid_skipped_duplicate_blocks"] = skipped_duplicates

reference/parsebench/pipeline/engine/policies.py

@@ -168,8 +168,11 @@ def replace_guarded(raw,candidates,merger,similarity=.65,guard='valid',repair=Fa
  for overlap,sim,i,j,body in sorted(pairs,reverse=True):
   if i in chosen or j in used:continue
   chosen[i]=body;used.add(j)
- index=iter(range(len(blocks)))
+ # Match the same coordinate grammar used by parse_blocks. The broader BLOCK
+ # pattern also matches malformed coordinates and cannot share positional indices.
+ spans={(m.start(),m.end()):i for i,m in enumerate(merger.BLOCK_RE.finditer(raw))}
  def f(m):
-  i=next(index)
-  return m[1]+chosen[i]+m[3] if i in chosen else m[0]
- return BLOCK.sub(f,raw)
+  i=spans[(m.start(),m.end())]
+  if i not in chosen:return m[0]
+  return raw[m.start():m.start(3)]+chosen[i]+raw[m.end(3):m.end()]
+ return merger.BLOCK_RE.sub(f,raw)

reference/parsebench/pipeline/parse_results.py

@@ -0,0 +1,92 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: OpenMDW-1.1
+"""Parse available model content and isolate failures; no request or time policies."""
+from __future__ import annotations
+
+import html
+import multiprocessing
+from collections import defaultdict
+from concurrent.futures import ProcessPoolExecutor
+
+import postprocess as renderer
+from engine import merge
+from engine.blocks import canonical_geometry, has_unparsed_content, incomplete_tail, text_blocks
+
+
+def error(stage, exc):
+    return {'stage': stage, 'type': type(exc).__name__, 'message': str(exc)}
+
+
+def fallback(row, errors):
+    raw = row.get('raw_out') or ''
+    return {'key': renderer.key(row), 'raw_out': raw,
+            'source_raw_out': row.get('source_raw_out', raw),
+            'markdown': html.escape(raw), 'coordinate_frame': 'content_normalized',
+            'postprocessing_status': 'raw_fallback', 'postprocessing_errors': errors}
+
+
+def finalize(task):
+    row, candidates, errors = task
+    errors = list(errors)
+    original = row.get('raw_out') or ''
+    if row.get('error'):
+        errors.append({'stage': 'fullpage_generation', 'error': row['error']})
+    try:
+        raw = renderer.repair_empty_prefix(original)
+        if has_unparsed_content(raw) and any(t in raw for t in ['<x_', '<y_', '<class_']):
+            errors.append({'stage': 'coordinate_validation',
+                           'type': 'IncompleteBlock' if incomplete_tail(raw) else 'MalformedCoordinates'})
+        known = renderer.adapter.TEXT_BEARING_CLASSES | {'Table', 'Formula', 'Picture'}
+        unknown = sorted({cls for cls, _ in text_blocks(raw) if cls not in known})
+        if unknown:
+            errors.append({'stage': 'class_validation', 'type': 'UnknownClasses', 'classes': unknown})
+        row = dict(row, raw_out=canonical_geometry(raw))
+        if row['raw_out'] != original:
+            row['source_raw_out'] = original
+        if candidates:
+            row = merge.build_direct_plus_ordered_structural_addonly(
+                renderer.arguments(), [row], {'direct': merge.index_rows([row])},
+                {renderer.key(row): candidates})[0]
+            # Insertions can place a formerly unfinished tail before a new
+            # closed block. Validate that boundary again before rendering.
+            row = dict(row, raw_out=canonical_geometry(row['raw_out']), source_raw_out=original)
+        result = renderer.finalize_one((row, candidates))
+        result['postprocessing_status'] = 'partial' if errors else ('empty' if not original.strip() else 'ok')
+        result['postprocessing_errors'] = errors
+        if row.get('source_raw_out') is not None:
+            result['source_raw_out'] = row['source_raw_out']
+        return result
+    except Exception as exc:
+        return fallback(dict(row, source_raw_out=original), errors + [error('page_rendering', exc)])
+
+
+def process(direct, crop_path=None, workers=8):
+    rows = [dict(row, key=renderer.key(row)) for row in direct]
+    if len({r['key'] for r in rows}) != len(rows):
+        raise ValueError('Duplicate input keys')
+    args = renderer.arguments()
+    classes = merge.parse_class_set(args.structural_classes)
+    labels = merge.parse_label_filter(args.include_detector_labels)
+    candidates, errors = defaultdict(list), defaultdict(list)
+    for crop in renderer.read_rows(crop_path) if crop_path else []:
+        key = str(crop.get('page_key') or '')
+        if not key:
+            continue
+        try:
+            if crop.get('error'):
+                errors[key].append({'stage': 'crop_generation', 'error': crop['error']})
+                continue
+            if labels and merge.normalize_detector_label(crop.get('detector_label') or '') not in labels:
+                continue
+            for block in merge.rebuild_crop_row_blocks(crop, args):
+                if block.cls in classes and not (args.min_structural_area > 0 and merge.block_area(block) < args.min_structural_area):
+                    candidates[key].append(block)
+        except Exception as exc:
+            errors[key].append(error('crop_reconstruction', exc))
+    for blocks in candidates.values():
+        blocks.sort(key=merge.reading_order_key)
+    tasks = [(row, candidates.get(row['key'], []), errors.get(row['key'], [])) for row in rows]
+    if workers == 1:
+        return [finalize(task) for task in tasks]
+    with ProcessPoolExecutor(max_workers=workers, mp_context=multiprocessing.get_context('fork')) as pool:
+        return list(pool.map(finalize, tasks, chunksize=8))

reference/parsebench/pipeline/postprocess.py

@@ -54,18 +54,8 @@ def finalize_one(task):
     return {'key':k,'raw_out':raw,'markdown':markdown(raw),'coordinate_frame':'content_normalized'}
 
 def process(direct,crop_path=None,workers=8):
-    direct=[dict(row,key=key(row)) for row in direct]
-    if len({x['key'] for x in direct})!=len(direct):raise ValueError('Duplicate input keys')
-    if any(x.get('error') for x in direct):raise ValueError('Input contains generation failures')
-    args=arguments(crop_path);candidates=merge.crop_element_blocks_by_page(args)
-    if candidates:
-        indexes={'direct':merge.index_rows(direct)}
-        joined=merge.build_direct_plus_ordered_structural_addonly(args,direct,indexes,candidates)
-    else:joined=direct
-    tasks=[(row,candidates.get(key(row),[])) for row in joined]
-    if workers==1:return list(map(finalize_one,tasks))
-    with ProcessPoolExecutor(max_workers=workers,mp_context=multiprocessing.get_context('fork')) as pool:
-        return list(pool.map(finalize_one,tasks,chunksize=8))
+    from parse_results import process as parse_process
+    return parse_process(direct,crop_path,workers)
 
 def main():
     p=argparse.ArgumentParser(description=__doc__);p.add_argument('--direct',type=Path,required=True)

reference/parsebench/pipeline/tests/malformed_coordinate_regression.json

@@ -0,0 +1,6 @@
+[
+  {
+    "key": "malformed_footer_fixture",
+    "raw_out": "<x_0.1><y_0.1>Text 0<x_0.8><y_0.8><class_Text><x_0.1><y_0.1>Text 1<x_0.8><y_0.8><class_Text><x_0.1><y_0.1>Text 2<x_0.8><y_0.8><class_Text><x_0.1><y_0.1>Text 3<x_0.8><y_0.8><class_Text><x_0.1><y_0.1>Text 4<x_0.8><y_0.8><class_Text><x_0.1><y_0.1>Text 5<x_0.8><y_0.8><class_Text><x_0.1><y_0.1>Text 6<x_0.8><y_0.8><class_Text><x_0.1><y_0.1>Text 7<x_0.8><y_0.8><class_Text><x_0.1><y_0.1>Text 8<x_0.8><y_0.8><class_Text><x_0.1><y_0.1>Text 9<x_0.8><y_0.8><class_Text><x_0.1><y_0.1>Text 10<x_0.8><y_0.8><class_Text><x_0.1><y_0.1>Text 11<x_0.8><y_0.8><class_Text><x_0.1><y_0.1>Text 12<x_0.8><y_0.8><class_Text><x_0.1><y_0.1>Text 13<x_0.8><y_0.8><class_Text><x_0.1><y_0.1>Text 14<x_0.8><y_0.8><class_Text><x_0.1934><y_0.925>Footer text<x_0.org><y_0.9641><class_Page-footer>"
+  }
+]

reference/parsebench/pipeline/tests/test_parser_general.py

@@ -0,0 +1,97 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: OpenMDW-1.1
+import json
+from pathlib import Path
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+import postprocess
+from engine import merge, policies
+import parse_results
+
+
+class ParserFixes(unittest.TestCase):
+    def test_observed_malformed_footer_retains_text_without_invented_box(self):
+        path = Path(__file__).with_name('malformed_coordinate_regression.json')
+        for row in json.loads(path.read_text()):
+            result = postprocess.process([row], workers=1)[0]
+            self.assertEqual(result['source_raw_out'], row['raw_out'])
+            self.assertEqual(result['postprocessing_status'], 'partial')
+            self.assertIn('Footer text', result['markdown'])
+            self.assertEqual(len(merge.parse_blocks(result['raw_out'])), 15)
+
+    def test_malformed_block_does_not_shift_table_replacement(self):
+        malformed = '<x_0.1><y_0.1>footer text<x_0.org><y_0.2><class_Page-footer>'
+        valid = merge.make_block((.1, .2, .8, .9), 'Table', 'old table')
+        candidate = merge.parse_blocks(merge.make_block((.1, .2, .8, .9), 'Table', 'replacement table'))[0]
+        result = policies.replace_guarded(malformed+valid, [candidate], merge, similarity=0)
+        self.assertTrue(result.startswith(malformed))
+        self.assertIn('replacement table', result)
+
+    def test_orphan_coordinates_do_not_swallow_next_valid_block(self):
+        raw = '<x_0.1><y_0.1>broken prefix' + merge.make_block((.2, .2, .8, .3), 'Title', 'A valid title')
+        blocks = merge.parse_blocks(raw)
+        self.assertEqual(len(blocks), 1)
+        self.assertEqual(blocks[0].body, 'A valid title')
+
+    def test_bad_page_keeps_other_page_and_original_output(self):
+        original = postprocess.finalize_one
+        def faulty(task):
+            if task[0]['key'] == 'bad':
+                raise ValueError('invalid table')
+            return original(task)
+        with patch.object(postprocess, 'finalize_one', side_effect=faulty):
+            rows = postprocess.process([{'key': 'good', 'raw_out': 'readable'}, {'key': 'bad', 'raw_out': 'original text'}], workers=1)
+        self.assertEqual(rows[0]['markdown'], 'readable')
+        self.assertEqual(rows[1]['postprocessing_status'], 'raw_fallback')
+        self.assertEqual(rows[1]['source_raw_out'], 'original text')
+
+    def test_bad_crop_preserves_full_page(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            p = Path(tmp)/'crops.jsonl'
+            p.write_text(json.dumps({'page_key': 'page', 'detector_label': 'table', 'raw_out': 'bad crop', 'detector_bbox_norm': ['invalid']})+'\n')
+            row = postprocess.process([{'key': 'page', 'raw_out': 'full page survives'}], p, workers=1)[0]
+        self.assertEqual(row['markdown'], 'full page survives')
+        self.assertEqual(row['postprocessing_status'], 'partial')
+
+    def test_unfinished_chart_not_claimed_as_complete(self):
+        raw = merge.make_block((.1, .1, .8, .2), 'Text', 'finished text') + '\n<x_0.1><y_0.3><chart type="bar">unfinished data'
+        row = postprocess.process([{'key': 'page', 'raw_out': raw}], workers=1)[0]
+        self.assertEqual(row['postprocessing_status'], 'partial')
+        self.assertIn('finished text', row['markdown'])
+        self.assertEqual(row['raw_out'], raw)
+
+    def test_unfinished_table_cannot_contaminate_next_completed_table(self):
+        unfinished = '<x_0.1><y_0.1>\\begin{tabular}{cc}\nunfinished & data'
+        complete = merge.make_block((.2, .3, .8, .9), 'Table', '\\begin{tabular}{cc}\nleft & right\\\\\n\\end{tabular}')
+        from engine.blocks import text_blocks
+        self.assertEqual(list(text_blocks(unfinished+complete)), [('Table', '\\begin{tabular}{cc}\nleft & right\\\\\n\\end{tabular}')])
+        result = postprocess.process([{'key': 'page', 'raw_out': unfinished+complete}], workers=1)[0]
+        self.assertNotIn('unfinished', result['markdown'])
+        self.assertIn('left', result['markdown'])
+        self.assertIn('unfinished', result['source_raw_out'])
+
+    def test_unterminated_trailing_fragment_is_retained_only_in_raw(self):
+        raw = merge.make_block((.1, .1, .8, .2), 'Text', 'completed text') + '\nunfinished trailing text'
+        result = postprocess.process([{'key': 'page', 'raw_out': raw}], workers=1)[0]
+        self.assertEqual(result['markdown'], 'completed text')
+        self.assertIn('unfinished trailing text', result['raw_out'])
+        self.assertEqual(result['postprocessing_status'], 'partial')
+
+    def test_plain_input_without_serialized_blocks_still_works(self):
+        result = postprocess.process([{'key': 'page', 'raw_out': 'ordinary unstructured text'}], workers=1)[0]
+        self.assertEqual(result['markdown'], 'ordinary unstructured text')
+
+    def test_unfinished_serialized_table_only_not_reported_as_parsed(self):
+        raw = '<x_0.1><y_0.1>\\begin{tabular}{cc}\nunfinished & data'
+        result = postprocess.process([{'key': 'page', 'raw_out': raw}], workers=1)[0]
+        self.assertEqual(result['markdown'], '')
+        self.assertEqual(result['raw_out'], raw)
+        self.assertEqual(result['postprocessing_status'], 'partial')
+
+
+if __name__ == '__main__':
+    unittest.main()