diff --git a/api/chunking.py b/api/chunking.py index 5660abf..2bf8baf 100644 --- a/api/chunking.py +++ b/api/chunking.py @@ -36,6 +36,14 @@ from datetime import date, datetime CHUNK_TOKEN_TARGET = int(os.environ.get("CHUNK_TOKEN_TARGET", "800")) +# The hard ceiling no chunk may exceed, whatever the input looks like. +# +# Set well under text-embedding-3-small's 8191-token input limit, because +# approx_tokens() is a length/4 estimate and under-counts dense technical text. +# A chunk over the model's limit does not degrade - the embeddings call fails +# and the document cannot be published at all. +MAX_CHUNK_TOKENS = int(os.environ.get("MAX_CHUNK_TOKENS", "6000")) + MARKDOWN_HEADING = re.compile(r"^(#{1,6})\s+(.*\S)\s*$") DOC_NUMBER_RE = re.compile(r"\b(WRPS-[A-Z]{2,4}-\d{3,4})\b") @@ -48,7 +56,13 @@ DATE_RE = re.compile( r"\d{1,2}\s+\w+\s+\d{4})\b", re.IGNORECASE, ) -TITLE_RE = re.compile(r"^\s*title[\s:]+(.+\S)\s*$", re.IGNORECASE | re.MULTILINE) +# [ ], NOT \s: \s includes the newline, so the old form matched a bare +# "Title" line, consumed the line break and captured whatever was on the NEXT +# line. A real cover page flattens to a label column then a value column - +# Title / Revision / Date - and every chunk of WRPS-CTL-001 was stored with +# doc_title = "Revision". Same trap AUTHORISING_ROLE_RE was fixed for. +TITLE_RE = re.compile(r"^[ ]*title[ :]+(.+\S)[ ]*$", + re.IGNORECASE | re.MULTILINE) AUTHORISING_ROLE_RE = re.compile( r"^[ \t]*authoris(?:ing|ed)[ \t]*(?:role|by)?[ \t]*:[ \t]*(.+\S)[ \t]*$", re.IGNORECASE | re.MULTILINE, @@ -136,18 +150,96 @@ def split_sections(text: str) -> list[tuple[int, str, str]]: return [(page, title, "\n".join(body)) for page, title, body in sections] +def _split_on_words(line: str, limit: int) -> list[str]: + """Split one over-long line on word boundaries. + + The bottom of the ladder. A PDF page can extract as a SINGLE line with no + newline anywhere in it, and at that point there is no structure left to + respect - only the limit, which is not negotiable because exceeding the + embedding model's input makes the document unpublishable. + """ + pieces: list[str] = [] + buffer: list[str] = [] + for word in line.split(" "): + candidate = " ".join(buffer + [word]) + if buffer and approx_tokens(candidate) > limit: + pieces.append(" ".join(buffer)) + buffer = [word] + else: + buffer.append(word) + if buffer: + pieces.append(" ".join(buffer)) + # A single "word" longer than the limit is not language - it is a base64 + # blob or a table rendered without spaces. Slice it rather than emit it. + bounded: list[str] = [] + for piece in pieces: + while approx_tokens(piece) > limit: + bounded.append(piece[: limit * 4]) + piece = piece[limit * 4 :] + if piece: + bounded.append(piece) + return bounded + + +def _split_on_lines(text: str, limit: int) -> list[str]: + """Last-resort split, on line boundaries, honouring `limit`. + + Used when paragraph splitting could not get a chunk under the ceiling - + text with no blank lines in it at all, which is exactly what flat PDF + extraction produces. Falls through to word boundaries for a single line + that is itself over the limit. + """ + chunks: list[str] = [] + buffer: list[str] = [] + for line in text.splitlines() or [text]: + if approx_tokens(line) > limit: + if buffer: + chunks.append("\n".join(buffer)) + buffer = [] + chunks.extend(_split_on_words(line, limit)) + continue + candidate = "\n".join(buffer + [line]) + if buffer and approx_tokens(candidate) > limit: + chunks.append("\n".join(buffer)) + buffer = [line] + else: + buffer.append(line) + if buffer: + chunks.append("\n".join(buffer)) + return chunks + + def chunk_section(text: str, doc_type: str) -> list[str]: """Split a section, unless splitting it would break a step sequence. - For procedures the rule is absolute: a section containing numbered steps is - emitted whole, however long it is. An oversized chunk costs tokens. Half a + For procedures the rule still holds: a section containing numbered steps is + emitted whole rather than split. An oversized chunk costs tokens; half a procedure costs more than that. + + THE CEILING, added 2026-08-28. The rule above used to be unbounded, and + that was safe only while sections arrived pre-split by Docling and were + therefore small. They no longer always do: flat PDF extraction can hand + this function an entire document as one section, and an unbounded refusal + then produces one chunk for the whole document - one embedding vector for + eight pages, useless retrieval, and a citation reading "(untitled)". Worse, + a long enough document exceeds the embedding model's input limit and the + publish fails outright. + + So the refusal is now bounded by MAX_CHUNK_TOKENS. Below it, a procedure + section stays whole exactly as before. Above it, splitting is the lesser + harm - an embedding call that fails protects nobody. The gap between + CHUNK_TOKEN_TARGET and MAX_CHUNK_TOKENS is deliberately wide so that a real + step sequence, which is the case the rule exists for, is never near it. """ if approx_tokens(text) <= CHUNK_TOKEN_TARGET: return [text] if doc_type == "procedure" and STEP_RE.search(text): - return [text] + if approx_tokens(text) <= MAX_CHUNK_TOKENS: + return [text] + # Past the ceiling. Fall through and split - and split on lines, since + # a blob this shape usually has no paragraph breaks to use. + return _split_on_lines(text, CHUNK_TOKEN_TARGET) chunks: list[str] = [] buffer: list[str] = [] @@ -160,7 +252,17 @@ def chunk_section(text: str, doc_type: str) -> list[str]: buffer.append(paragraph) if buffer: chunks.append("\n\n".join(buffer)) - return chunks + + # A single paragraph can still be over the ceiling - text with no blank + # lines is one paragraph however long it is. Nothing above this point can + # fix that, so enforce it here rather than trusting the input. + bounded: list[str] = [] + for chunk in chunks: + if approx_tokens(chunk) > MAX_CHUNK_TOKENS: + bounded.extend(_split_on_lines(chunk, CHUNK_TOKEN_TARGET)) + else: + bounded.append(chunk) + return bounded def link_equipment(text: str, equipment_ids: list[str]) -> str | None: diff --git a/api/convert.py b/api/convert.py index 62d86e6..6ac2ad4 100644 --- a/api/convert.py +++ b/api/convert.py @@ -62,6 +62,110 @@ def _clean(text: str) -> str: return text.strip() +# A numbered section heading: "1. Purpose", "3.2 Pump control", "4.1.2 Alarms". +# +# THE HARD PART. A numbered STEP in a procedure looks identical to a numbered +# HEADING - both are ". ". Getting this wrong in the dangerous +# direction would break a step sequence apart, which is the one thing chunking +# must never do. So the test is deliberately narrow: a heading is SHORT, has no +# terminal punctuation, and is a handful of words. "1. Purpose" passes. +# "1. Open the isolation valve and confirm zero pressure." does not - it is +# long, and it ends in a full stop. +_NUMBERED_HEADING = re.compile(r"^(\d+(?:\.\d+)*)[.)]?\s+(\S.*)$") +# A heading in capitals: "SECTION 4 - ALARM PHILOSOPHY". Requires a SPACE, so +# it needs at least two words. Found on the first real document: without that, +# "SCADA", "PLC-001" and every "WRPS-PRO-001" in a reference list became +# headings - which is worse than missing a heading, because each one cut a real +# section short and re-titled the remainder with something meaningless. A +# citation reading "SCADA" helps nobody. +_CAPS_HEADING = re.compile(r"^[A-Z][A-Z0-9&/(),.'-]*(?: +[A-Z0-9&/(),.'-]+)+$") + +# "11 August 2026" is not section 11. Same document: the revision history dates +# parsed as numbered headings and split the header block into fragments. +_DATE_LIKE = re.compile( + r"^\d{1,2}[ .-]+(jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)", + re.IGNORECASE, +) + +_HEADING_MAX_CHARS = 80 +_HEADING_MAX_WORDS = 10 + + +def _looks_like_heading(line: str) -> str | None: + """Return the heading text, or None. Conservative by design - see above.""" + line = line.strip() + if not line or len(line) > _HEADING_MAX_CHARS: + return None + if line.endswith((".", ";", ":", ",")): + return None + + if _DATE_LIKE.match(line): + return None + + numbered = _NUMBERED_HEADING.match(line) + if numbered: + title = numbered.group(2).strip() + if title and title[0].isupper() and len(title.split()) <= _HEADING_MAX_WORDS: + return line + return None + + if (_CAPS_HEADING.match(line) + and 2 <= len(line.split()) <= _HEADING_MAX_WORDS): + return line + return None + + +def _structure(lines: list[str]) -> str: + """Flat PDF text lines -> Markdown with headings and real paragraphs. + + WHY THIS EXISTS. pypdf emits one line per line of the PDF and NO blank + lines at all - a real document came through as 416 lines, none of them + blank. The chunker splits sections on Markdown headings and paragraphs on + blank lines, so without this an entire document is one untitled section and + one paragraph: a single 18,000-character chunk, one embedding vector for + eight pages, and every citation reading "(untitled), page 1". + + Paragraph reflow uses line width. PDF body text wraps at a consistent + measure, so a line noticeably shorter than the running width is the LAST + line of its paragraph. It is a heuristic and it will occasionally join two + paragraphs or split one - which is tolerable, because a person reads this + text before the document can be cited, and because chunk_section now has a + hard ceiling that does not depend on getting paragraphs right. + """ + body = [ln for ln in lines if ln.strip() and not _looks_like_heading(ln)] + widths = sorted(len(ln.rstrip()) for ln in body) + # Median width of body lines, with a floor so a very short document does + # not produce a nonsense threshold. + typical = widths[len(widths) // 2] if widths else 0 + short_line = max(int(typical * 0.75), 30) + + out: list[str] = [] + para: list[str] = [] + + def flush() -> None: + if para: + out.append(" ".join(para)) + para.clear() + + for raw in lines: + line = raw.strip() + if not line: + flush() + continue + heading = _looks_like_heading(line) + if heading: + flush() + out.append(f"## {heading}") + continue + para.append(line) + if len(line) < short_line: + # Short line = end of a wrapped paragraph. + flush() + flush() + + return "\n\n".join(out) + + def _from_pdf(data: bytes) -> Converted: from pypdf import PdfReader @@ -84,13 +188,18 @@ def _from_pdf(data: bytes) -> Converted: except Exception: # One bad page must not lose the other ninety. Mark it so the # reviewer can see exactly what is missing. - text = "" - parts.append(f"\n\n\n") + parts.append(f"") + continue if text.strip(): - parts.append(f"\n\n\n\n{text}") + # Structure each page separately: the width heuristic in + # _structure() is per-page because a landscape table page and a + # portrait text page have different measures, and mixing them + # makes the threshold meaningless for both. + parts.append(f"") + parts.append(_structure(text.splitlines())) return Converted( - markdown=_clean("".join(parts)), + markdown=_clean("\n\n".join(parts)), converter="pypdf", page_count=len(reader.pages), ) diff --git a/api/documents.py b/api/documents.py index 0d70442..3e2b932 100644 --- a/api/documents.py +++ b/api/documents.py @@ -578,9 +578,27 @@ def approve( superseded_count = 0 with conn.cursor() as cur: - # Replace any earlier ingest of this exact file, so re-publishing - # cannot double the chunks. - cur.execute("DELETE FROM doc_chunks WHERE source_file=%s", (stored_path,)) + # Replace every earlier publication of THIS DOCUMENT REVISION, not + # just of this file. + # + # source_file carries the upload_id, which is new on every upload, + # so matching on it alone made re-publishing ADDITIVE: approving + # WRPS-CTL-001 rev A twice left 38 live chunks, the same passage + # retrievable and citable twice over. It was invisible on screen, + # because live_documents groups by (doc_number, revision) and only + # the chunk count moved. + # + # Deleting by (doc_number, revision) also clears any superseded + # chunks of the same revision left by an earlier withdraw. That is + # correct: those are stale copies of what is being republished, not + # history. The history is doc_actions, which records who withdrew + # what and why, and which nothing can delete from. + cur.execute( + "DELETE FROM doc_chunks WHERE source_file=%s" + " OR (doc_number=%s AND revision=%s)", + (stored_path, confirmed_doc_number.strip(), + confirmed_revision.strip())) + replaced = cur.rowcount if supersede: cur.execute(""" @@ -615,9 +633,12 @@ def approve( (len(records), superseded_count, stored_path, upload_id)) conn.commit() - log.info("published %s rev %s: %d chunks, %d superseded, by %s", + # `replaced` is worth logging on its own: a non-zero value means this + # document revision was already in the library and has been overwritten, + # which is invisible on screen. + log.info("published %s rev %s: %d chunks, %d replaced, %d superseded, by %s", confirmed_doc_number, confirmed_revision, len(records), - superseded_count, who.stored) + replaced, superseded_count, who.stored) return RedirectResponse("/documents", status_code=303) diff --git a/api/tests/test_documents.py b/api/tests/test_documents.py index 59f267a..1e9e8fc 100644 --- a/api/tests/test_documents.py +++ b/api/tests/test_documents.py @@ -215,3 +215,116 @@ def test_the_router_can_be_mounted(): for expected in ("/documents", "/documents/upload", "/documents/withdraw", "/documents/restore", "/documents/review/{upload_id}"): assert expected in paths, f"{expected} is not mounted" + + +# --- flat PDF text: the defect found on the first real document ------------- +# +# WRPS-CTL-001, an 8-page control philosophy, published as ONE 18,307-character +# chunk titled "(untitled)". pypdf emits one line per PDF line and no blank +# lines at all - 416 lines, 0 blank - so section splitting found no headings +# and paragraph splitting found no paragraphs. Both halves of the fix are +# locked below. + +FLAT_PDF_TEXT = "\n".join( + ["1. Purpose"] + + ["This document states how the station is to be controlled and why it"] * 40 + + ["3.2 Pump control"] + + ["The duty pump starts on rising level and the assist pumps follow it"] * 40 +) + + +def test_flat_pdf_text_gains_headings_and_paragraphs(): + """The conversion must produce structure, not one undifferentiated wall.""" + result = convert._structure(FLAT_PDF_TEXT.splitlines()) + assert "## 1. Purpose" in result + assert "## 3.2 Pump control" in result + assert "\n\n" in result, "no paragraph breaks were produced" + + +def test_a_numbered_step_is_not_mistaken_for_a_heading(): + """The dangerous direction. A step promoted to a heading splits a step + sequence, which is the one thing chunking must never do.""" + assert convert._looks_like_heading("1. Purpose") == "1. Purpose" + assert convert._looks_like_heading( + "1. Open the isolation valve on PU-301 and confirm zero pressure.") is None + assert convert._looks_like_heading( + "2. Close the discharge valve, then wait sixty seconds before starting.") is None + + +def test_no_chunk_may_exceed_the_ceiling_even_with_no_paragraph_breaks(): + """The backstop. This does not depend on the heading heuristic working.""" + wall = "x" * (chunking.MAX_CHUNK_TOKENS * 4 * 3) # 3x the ceiling, one line + for doc_type in ("design", "procedure", "manual", "rationalisation"): + for chunk in chunking.chunk_section(wall, doc_type): + assert chunking.approx_tokens(chunk) <= chunking.MAX_CHUNK_TOKENS, doc_type + + +def test_a_step_sequence_under_the_ceiling_is_still_never_split(): + """The original rule, unchanged. The ceiling must not weaken it.""" + steps = "\n\n".join(f"{n}. Do the {n}th thing. " + "x" * 400 + for n in range(1, 30)) + assert chunking.approx_tokens(steps) > chunking.CHUNK_TOKEN_TARGET + assert chunking.approx_tokens(steps) < chunking.MAX_CHUNK_TOKENS + assert len(chunking.chunk_section(steps, "procedure")) == 1 + + +def test_the_whole_document_no_longer_becomes_one_chunk(): + """End to end over the shape that actually failed.""" + structured = convert._structure(FLAT_PDF_TEXT.splitlines()) + sections = chunking.split_sections(structured) + assert len(sections) >= 2, "headings did not create sections" + assert [t for _, t, _ in sections][:1] != ["(untitled)"] + + +def test_the_two_chunkers_have_not_drifted(): + """api/chunking.py and ingest/ingest.py must chunk identically. + + The header of api/chunking.py used to say drift in ingest.py could not be + detected from here. That was wrong - the source of both functions is right + there on disk. If they differ, the same document chunks differently + depending on whether it arrived through the UI or the CLI, and the + assistant answers or fails to answer depending on that. + + Compares source text, not behaviour: behaviour can agree on the cases + somebody thought to write down and differ on the one that matters. + """ + import pathlib + + here = pathlib.Path(__file__).resolve().parent.parent + def body(path: pathlib.Path, name: str) -> str: + src = path.read_text(encoding="utf-8") + start = src.index(f"def {name}(") + return src[start:src.index("\ndef ", start + 1)].strip() + + api = here / "chunking.py" + cli = here.parent / "ingest" / "ingest.py" + if not cli.exists(): # api/ checked out on its own + pytest.skip("ingest/ingest.py not present") + + for fn in ("chunk_section", "_split_on_lines", "_split_on_words", + "extract_header", "approx_tokens"): + assert body(api, fn) == body(cli, fn), ( + f"{fn} has drifted between api/chunking.py and ingest/ingest.py") + + +def test_a_bare_Title_label_does_not_capture_the_next_line(): + """The cover-page table trap, found on WRPS-CTL-001. + + A PDF table flattens to a label column then a value column. With `\s` in + the gap - which includes the newline - "Title" swallowed the line break and + captured "Revision" from the line below, and every chunk of an eight-page + document was stored with doc_title = "Revision". + + NULL is the right answer here. A citation falls back to the section title; + a confidently wrong document title does not fall back to anything. + """ + flattened = "Document number\nTitle\nRevision\nDate\nStatus\n" \ + "WRPS-CTL-001\nControl Philosophy\nA\n" + assert chunking.extract_header(flattened).title is None + + +def test_a_real_title_line_is_still_read(): + assert chunking.extract_header( + "Title: Wet Well Interlock Bypass\n").title == "Wet Well Interlock Bypass" + assert chunking.extract_header( + "Title Wet Well Interlock Bypass\n").title == "Wet Well Interlock Bypass" diff --git a/ingest/ingest.py b/ingest/ingest.py index 67bbdf4..3bbd4f7 100644 --- a/ingest/ingest.py +++ b/ingest/ingest.py @@ -55,6 +55,13 @@ log = logging.getLogger("ingest") DOCS_ROOT = Path(os.environ.get("AI_DOCS_ROOT", "/docs")) CHUNK_TOKEN_TARGET = int(os.environ.get("CHUNK_TOKEN_TARGET", "800")) +# The hard ceiling no chunk may exceed. Mirrors api/chunking.py - the two must +# stay in step or the same document chunks differently depending on which path +# loaded it. Set well under text-embedding-3-small's 8191-token input limit, +# because approx_tokens() is a length/4 estimate and under-counts dense text. +# A chunk over the model's limit does not degrade; the embeddings call fails +# and the document cannot be published at all. +MAX_CHUNK_TOKENS = int(os.environ.get("MAX_CHUNK_TOKENS", "6000")) DOC_TYPE_BY_FOLDER = { "procedures": "procedure", @@ -75,7 +82,13 @@ DATE_RE = re.compile( # Header fields that are not safety-critical but ARE facts about the document: # storing them stops the model being asked to read them off whatever chunk # retrieval happened to return, which is how it ended up returning "". -TITLE_RE = re.compile(r"^\s*title[\s:]+(.+\S)\s*$", re.IGNORECASE | re.MULTILINE) +# [ ], NOT \s: \s includes the newline, so the old form matched a bare +# "Title" line, consumed the line break and captured whatever was on the NEXT +# line. A real cover page flattens to a label column then a value column - +# Title / Revision / Date - and every chunk of WRPS-CTL-001 was stored with +# doc_title = "Revision". Same trap AUTHORISING_ROLE_RE was fixed for. +TITLE_RE = re.compile(r"^[ ]*title[ :]+(.+\S)[ ]*$", + re.IGNORECASE | re.MULTILINE) # "Authorising role: X", "Authorised by: X", "Authorising: X". The colon is # required: without it the lazy gap swallowed the field NAME and captured # "role: Station Maintenance Supervisor" as the value. @@ -130,7 +143,12 @@ def parse_date(text: str) -> date | None: def extract_header(text: str) -> Header: - """Pull document identity from the first page. Always confirmed by a human.""" + """Pull document identity from the first page. ALWAYS confirmed by a human. + + Everything here is a proposal shown on the review screen with the field + already filled in. It is not authority. A wrong revision on a procedure is + a safety issue, and a regex is not a person. + """ head = text[:4000] number = DOC_NUMBER_RE.search(head) revision = REVISION_RE.search(head) @@ -268,22 +286,96 @@ def approx_tokens(text: str) -> int: return len(text) // 4 +def _split_on_words(line: str, limit: int) -> list[str]: + """Split one over-long line on word boundaries. + + The bottom of the ladder. A PDF page can extract as a SINGLE line with no + newline anywhere in it, and at that point there is no structure left to + respect - only the limit, which is not negotiable because exceeding the + embedding model's input makes the document unpublishable. + """ + pieces: list[str] = [] + buffer: list[str] = [] + for word in line.split(" "): + candidate = " ".join(buffer + [word]) + if buffer and approx_tokens(candidate) > limit: + pieces.append(" ".join(buffer)) + buffer = [word] + else: + buffer.append(word) + if buffer: + pieces.append(" ".join(buffer)) + # A single "word" longer than the limit is not language - it is a base64 + # blob or a table rendered without spaces. Slice it rather than emit it. + bounded: list[str] = [] + for piece in pieces: + while approx_tokens(piece) > limit: + bounded.append(piece[: limit * 4]) + piece = piece[limit * 4 :] + if piece: + bounded.append(piece) + return bounded + + +def _split_on_lines(text: str, limit: int) -> list[str]: + """Last-resort split, on line boundaries, honouring `limit`. + + Used when paragraph splitting could not get a chunk under the ceiling - + text with no blank lines in it at all, which is exactly what flat PDF + extraction produces. Falls through to word boundaries for a single line + that is itself over the limit. + """ + chunks: list[str] = [] + buffer: list[str] = [] + for line in text.splitlines() or [text]: + if approx_tokens(line) > limit: + if buffer: + chunks.append("\n".join(buffer)) + buffer = [] + chunks.extend(_split_on_words(line, limit)) + continue + candidate = "\n".join(buffer + [line]) + if buffer and approx_tokens(candidate) > limit: + chunks.append("\n".join(buffer)) + buffer = [line] + else: + buffer.append(line) + if buffer: + chunks.append("\n".join(buffer)) + return chunks + + def chunk_section(text: str, doc_type: str) -> list[str]: """Split a section, unless splitting it would break a step sequence. - For procedures the rule is absolute: a section containing numbered steps is - emitted whole, however long it is. An oversized chunk costs tokens. Half a + For procedures the rule still holds: a section containing numbered steps is + emitted whole rather than split. An oversized chunk costs tokens; half a procedure costs more than that. + + THE CEILING, added 2026-08-28. The rule above used to be unbounded, and + that was safe only while sections arrived pre-split by Docling and were + therefore small. They no longer always do: flat PDF extraction can hand + this function an entire document as one section, and an unbounded refusal + then produces one chunk for the whole document - one embedding vector for + eight pages, useless retrieval, and a citation reading "(untitled)". Worse, + a long enough document exceeds the embedding model's input limit and the + publish fails outright. + + So the refusal is now bounded by MAX_CHUNK_TOKENS. Below it, a procedure + section stays whole exactly as before. Above it, splitting is the lesser + harm - an embedding call that fails protects nobody. The gap between + CHUNK_TOKEN_TARGET and MAX_CHUNK_TOKENS is deliberately wide so that a real + step sequence, which is the case the rule exists for, is never near it. """ if approx_tokens(text) <= CHUNK_TOKEN_TARGET: return [text] if doc_type == "procedure" and STEP_RE.search(text): - log.info( - "keeping a %d-token procedure section whole - it contains a step sequence", - approx_tokens(text), - ) - return [text] + if approx_tokens(text) <= MAX_CHUNK_TOKENS: + return [text] + # Past the ceiling. Fall through and split - and split on lines, since + # a blob this shape usually has no paragraph breaks to use. + return _split_on_lines(text, CHUNK_TOKEN_TARGET) chunks: list[str] = [] buffer: list[str] = [] @@ -296,7 +388,17 @@ def chunk_section(text: str, doc_type: str) -> list[str]: buffer.append(paragraph) if buffer: chunks.append("\n\n".join(buffer)) - return chunks + + # A single paragraph can still be over the ceiling - text with no blank + # lines is one paragraph however long it is. Nothing above this point can + # fix that, so enforce it here rather than trusting the input. + bounded: list[str] = [] + for chunk in chunks: + if approx_tokens(chunk) > MAX_CHUNK_TOKENS: + bounded.extend(_split_on_lines(chunk, CHUNK_TOKEN_TARGET)) + else: + bounded.append(chunk) + return bounded def link_equipment(text: str, equipment_ids: list[str]) -> str | None: