Streaming Markdown code parser

Model output arrives as chunks of arbitrary size: a chunk can end in the middle of a word, in the middle of a fence, or between the \r and \n of a line break. Ask:

• What counts as a code block? A fenced block: a line of three or more backticks (or tildes), an optional info string such as python or ts file=app.ts, the code lines, and a closing fence of the same character that is at least as long as the opening one. Indented (4-space) code blocks are usually out of scope.
• What do we emit, and when? At minimum a complete CodeBlock(lang, info, code) the moment its closing fence line ends. Often also the prose between blocks, and code lines as they arrive so the editor can render them live.
• What happens at end of stream with an open block? Emit it, marked unclosed, so nothing the model wrote is lost.

Brute force: append every chunk to a buffer and re-parse the whole buffer each time. Correct, but O(n²) over a long response, and it re-emits blocks unless you track what was already sent.

Line-based state machine. Fences are line constructs, so the parser only needs complete lines:

1. Keep a buffer for the partial last line. On each chunk, split on \n. Every piece except the last is a complete line; the last piece stays in the buffer.
2. Run each complete line through a two-state machine:
   • Outside a block. If the line is an opening fence (up to 3 spaces, then 3+ backticks or tildes, then an info string; a backtick fence's info string may not contain a backtick), switch to inside and remember the fence character and length. Otherwise it is prose.
   • Inside a block. If the line is a closing fence (up to 3 spaces, 3+ of the same character, at least the opening length, nothing but spaces after), emit the block and switch back. Otherwise it is a code line: append it and, optionally, emit it right away.
3. finish() flushes the buffered partial line and closes any open block as unclosed.

Chunk invariance is the property to test: any way of splitting the same text into chunks must produce the same events. Fuzz it.

Each character is examined a constant number of times, so the whole stream costs O(n). One detail keeps that true: a very long line that arrives in many chunks must not be rebuilt by string concatenation on every chunk. Keep the pieces in a list and join once, when the newline arrives.

import re
from dataclasses import dataclass, field
from typing import List, Optional, Tuple

OPEN = re.compile(r"^ {0,3}(`{3,}|~{3,})(.*)$")
CLOSE = re.compile(r"^ {0,3}(`{3,}|~{3,})[ \t]*$")


@dataclass
class CodeBlock:
    lang: str
    info: str
    code: str
    closed: bool = True


@dataclass
class _Open:
    char: str
    length: int
    info: str
    lines: List[str] = field(default_factory=list)


Event = Tuple[str, object]  # ("text", str) | ("code_line", str) | ("code", CodeBlock)


class StreamingCodeParser:
    def __init__(self, emit_lines: bool = False) -> None:
        self.emit_lines = emit_lines
        self._pieces: List[str] = []   # the partial current line, unjoined
        self._block: Optional[_Open] = None

    def feed(self, chunk: str) -> List[Event]:
        events: List[Event] = []
        parts = chunk.split("\n")
        for part in parts[:-1]:
            self._pieces.append(part)
            line = "".join(self._pieces)
            self._pieces = []
            self._line(line.rstrip("\r"), events)
        if parts[-1]:
            self._pieces.append(parts[-1])
        return events

    def finish(self) -> List[Event]:
        events: List[Event] = []
        if self._pieces:
            line = "".join(self._pieces)
            self._pieces = []
            self._line(line.rstrip("\r"), events)
        if self._block is not None:
            events.append(("code", self._make(self._block, closed=False)))
            self._block = None
        return events

    # ---- one complete line ----------------------------------------------
    def _line(self, line: str, events: List[Event]) -> None:
        block = self._block
        if block is None:
            m = OPEN.match(line)
            if m and not (m.group(1)[0] == "`" and "`" in m.group(2)):
                self._block = _Open(m.group(1)[0], len(m.group(1)), m.group(2).strip())
            else:
                events.append(("text", line))
            return
        m = CLOSE.match(line)
        if m and m.group(1)[0] == block.char and len(m.group(1)) >= block.length:
            events.append(("code", self._make(block, closed=True)))
            self._block = None
            return
        block.lines.append(line)
        if self.emit_lines:
            events.append(("code_line", line))

    @staticmethod
    def _make(block: _Open, closed: bool) -> CodeBlock:
        lang = block.info.split()[0] if block.info else ""
        return CodeBlock(lang=lang, info=block.info, code="\n".join(block.lines), closed=closed)


def parse_all(chunks: List[str], emit_lines: bool = False) -> List[Event]:
    p = StreamingCodeParser(emit_lines)
    out: List[Event] = []
    for c in chunks:
        out.extend(p.feed(c))
    out.extend(p.finish())
    return out


if __name__ == "__main__":
    doc = "Here is the fix:\n```python file=app.py\nprint('hi')\n```\nDone."
    for ev in parse_all([doc[i:i + 3] for i in range(0, len(doc), 3)]):
        print(ev)

• feed is O(len(chunk)) amortized; each character is split once and joined once.
• Memory is O(longest line + size of the open block). If blocks can be huge and the caller only needs streaming lines, stop storing lines and keep only the count.

• A block split across many chunks, including a split inside the opening fence and inside the closing fence.
• Prose before, between and after blocks comes out as text events in order.
• A longer opening fence (four backticks) contains a three-backtick line as code.
• Tildes: a ~~~ block is not closed by backticks.
• An unclosed block at end of stream comes out with closed=False.
• Windows line endings (\r\n) split between chunks.
• Every split of a document gives the same events (chunk invariance).

import random as _random

_DOC = (
    "Intro line\n"
    "```python file=app.py\n"
    "def f():\n"
    "    return 1\n"
    "```\n"
    "between\n"
    "```\n"
    "plain\n"
    "```\n"
    "end"
)


def _codes(events):
    return [e[1] for e in events if e[0] == "code"]


def test_one_shot():
    events = parse_all([_DOC])
    codes = _codes(events)
    assert [(c.lang, c.info, c.code, c.closed) for c in codes] == [
        ("python", "python file=app.py", "def f():\n    return 1", True),
        ("", "", "plain", True),
    ]
    texts = [e[1] for e in events if e[0] == "text"]
    assert texts == ["Intro line", "between", "end"]


def test_chunk_invariance_every_two_way_split():
    expected = parse_all([_DOC])
    for i in range(len(_DOC) + 1):
        assert parse_all([_DOC[:i], _DOC[i:]]) == expected


def test_chunk_invariance_random_splits():
    expected = parse_all([_DOC])
    rng = _random.Random(1)
    for _ in range(200):
        cuts = sorted(rng.sample(range(1, len(_DOC)), 8))
        pieces = [_DOC[a:b] for a, b in zip([0] + cuts, cuts + [len(_DOC)])]
        assert parse_all(pieces) == expected


def test_block_is_emitted_when_closing_line_ends():
    p = StreamingCodeParser()
    assert _codes(p.feed("```js\nx = 1\n``")) == []
    assert _codes(p.feed("`")) == []           # fence line not finished yet
    assert [c.code for c in _codes(p.feed("\n"))] == ["x = 1"]

def test_longer_fence_contains_shorter():
    doc = "````md\n```\ninner\n```\n````\n"
    codes = _codes(parse_all([doc]))
    assert len(codes) == 1 and codes[0].code == "```\ninner\n```"


def test_tilde_not_closed_by_backticks():
    doc = "~~~\na\n```\nb\n~~~\n"
    codes = _codes(parse_all([doc]))
    assert codes[0].code == "a\n```\nb"


def test_unclosed_block_and_crlf():
    events = parse_all(["```py\r", "\nx = 1\r\ny = 2"])
    (block,) = _codes(events)
    assert block.code == "x = 1\ny = 2" and block.closed is False


def test_streaming_lines():
    events = parse_all(["```py\na\nb\n```\n"], emit_lines=True)
    assert [e for e in events if e[0] == "code_line"] == [("code_line", "a"), ("code_line", "b")]


def test_backtick_info_with_backtick_is_not_a_fence():
    events = parse_all(["``` a`b\ntext\n"])
    assert _codes(events) == [] and events[0] == ("text", "``` a`b")

1. Show a partial line while it streams. Users want to see characters, not whole lines. Emit the buffered partial line as a provisional code_partial event that the UI replaces when the line completes. Never treat a provisional line as a fence until it ends.
2. Fences inside lists and quotes. CommonMark lets a fence sit inside a list item (indented to the item's content column) or a block quote (`> ```). Track a container stack and strip each container's prefix before running the fence rules.
3. File targets. Editors read file=path or path:lines from the info string to know where to apply the code. Parse it into a dict at open time so the apply step can start loading the target file before the block finishes.
4. Back-pressure. If the consumer (syntax highlighter, diff applier) is slower than the model, buffer events in a bounded queue and coalesce code_line events into batches.
5. Full Markdown. Headings, lists and inline code need more states, but the design is the same: complete lines go through a block-level machine, and inline parsing runs once a block closes.

Mid-level (L4)

• Buffers the partial line so fences split across chunks are recognized.
• Two-state machine with correct open and close rules for the basic three-backtick case, plus the info string.
• Handles end of stream and writes a test that splits the input in several places.

Senior (L5)

• Gets the CommonMark details right without being led: fence length and character must match, up to three leading spaces, \r\n, backticks in a backtick info string.
• States and tests chunk invariance, ideally with a fuzz test over random splits.
• Keeps the parser O(n) and explains the quadratic trap of rebuilding a long partial line.

Staff (L6+)

• Designs the event API around the consumer: live code lines, provisional partial lines, file targets parsed early so applying can start.
• Talks through nested containers, back-pressure and how this feeds an apply-edit pipeline.
• Asks what latency the UI needs and shapes the events to meet it.