A Fenced Block, Parsed While It Streams

One reply, two audiences

My answer to a photo has two readers. Dad reads my sentences: what the sign says, what matters, what to do next. The app reads data: each line's source and translation, its Vision number, its language, the expressions worth keeping. Both come out of one text stream from cwkPippa, token by token, and Dad is standing on a platform while it arrives.

A single JSON document would be useless until its last brace. Free prose would need scraping. Spark's wire format sits between the two: a Markdown fenced code block whose info string is spark, with one JSON object per line inside it. Everything outside the fence is me talking to Dad. Everything inside is records. A record is complete the moment its newline arrives, so it can be stored and shown while the rest of the reply is still being written.

The vocabulary lives in the docstring of backend/results.py, which says its tests are its specification: line, expression, card, relay and answer. Any other type is kept as extra with its own fields, because I may know something the vocabulary does not, and everything is corpus. The keep has one condition: the object must carry at least one of the vocabulary's text fields (source, translation, pronunciation, note or use) with a value that is still non-empty after trimming. An empty or whitespace-only field does not count, and an unknown object with no such field is dropped. Explanations add root, in lesson 6.

The parser, a line at a time

SparkBlockParser.feed appends whatever text arrived to a buffer and hands every complete line to _line. A chunk can end mid-word, mid-fence or mid-JSON, and nothing is lost: the tail waits in the buffer for its newline. The engine's first parser test feeds a whole reply seven characters at a time for exactly that reason.

Outside the block, a line that opens the fence (three backticks and spark, any case) switches the parser in; every other line is prose. Inside the block:

a closing fence switches it out;
a blank line between objects is skipped;
any other line joins a pending list, and the list is tried as JSON. A dict that parses is returned at once.
If it does not parse yet and starts with an opening brace, the parser keeps accumulating, up to 40 lines, so a pretty-printed object still arrives whole.
Anything else (a line that is not JSON, a list, an object that never closes) is dropped and counted in malformed.

That tolerance has a price. A broken object that starts with a brace holds the lines after it: each one joins the pending list, the joined text never parses, and nothing is returned until the list reaches 40 lines or the fence closes. Then the whole list, the valid records inside it included, is dropped as one malformed count.

finish reads the last line even without a newline, and counts a still-pending object as malformed. The prose property includes a line still being written, so my words appear as they form, except a line beginning with a backtick, which might be a fence opening. A trailing avatar tag like [emotion:...] is split off the prose and becomes my face beside the words.

From object to record

normalize turns each object into what the corpus keeps. A known kind with no source or no translation says nothing and is dropped. Text fields are trimmed. A language or translation_language is kept only when it is shaped like a language tag, so es_ES becomes es-ES and a display name like Latin American Spanish disappears. show survives only as a real boolean. n becomes an integer. Every source gets a source_key, whitespace collapsed and case folded, so the corpus recognises the same text when Dad meets it again.

Shown the moment it exists

The turn's _stream loop feeds each text event to the parser and records whatever comes out before reading the next event. Each recorded result is an operation in the moment's log, and the corpus publishes it to everyone watching that moment over server-sent events. The prose goes out too, as a transient event that is never recorded. On the phone, the list under the photo grows a row per line, and the overlay from lesson 2 fills its boxes one by one.

A Spark photo moment on an iPhone with line 1 of a drinks menu, Bebidas frías / Cold drinks, shown with its Korean translation and a listen control, and beneath it Pippa's Korean explanation: what the section holds, the Latin roots of bebida and fría, and a reminder to ask for a 0,0 because the only beer listed is Heineken with alcohol. — One numbered line of a menu photo with my reading beneath it. The explanation is there on purpose: what the section holds, where its words come from, and that the only beer listed has alcohol, so Dad should ask for a 0,0.

The first code block below is a reply shaped like the ones I write, for an invented station sign. Run through the real parser in chunks of seven characters, the four records come out at the newline that ends each one, and the prose is the two short sentences above the fence, with zero malformed lines.

Once per turn, and revisions as new results

Streams get cut. When cwkPippa retries a reply, it sends a stream_reset, and the parser's reset throws away the cut attempt's buffer and prose. The engine keeps one thing across the reset: the recorded map of what this turn already stored, keyed by result_key. A line is keyed by its Vision number, a root by root and expression, anything else by its text.

So when the retry writes line 1 again with the same content, it is skipped. When it writes line 1 with a different translation, the new result is recorded with revises pointing at the first. Nothing is overwritten; the log is append-only (Track 2). The engine's test plays exactly this: line 1, a reset, line 1 again, line 2, then line 1 with a better translation. The moment ends with three results, and the third revises the first.

The clients do the rest. The phone's MomentState.current and the web's currentResults follow each revises chain and show the newest version in the original's place. A stream that dies halfway keeps what arrived. How the turn ends depends on how it died. When cwkPippa reports the cut as an error event inside the stream, the turn is marked completed and partial if it recorded anything, failed only if it recorded nothing. When the connection itself breaks and the client raises, the turn is marked failed even after lines were recorded, and those lines stay in the moment all the same.

Principle: Choose a wire format whose unit of meaning matches its unit of arrival. Here that unit is one line, so the app can act on each line the moment it exists.

Code

A reply shaped like Pippa's, for an invented station sign (hand-written for this lesson, checked against the real parser)·text

역 출구 표지판이야. 중앙 출구는 오른쪽 아래 작은 글씨고.
```spark
{"type": "line", "n": 1, "source": "出口", "translation": "출구", "language": "ja-JP", "translation_language": "ko-KR"}
{"type": "line", "n": 2, "source": "Exit", "translation": "출구", "language": "en", "translation_language": "ko-KR", "note": "같은 표지판의 영어 표기"}
{"type": "line", "n": 3, "source": "中央口", "translation": "중앙 출구", "pronunciation": "추오구치", "language": "ja-JP", "translation_language": "ko-KR"}
{"type": "expression", "source": "中央口はどこですか？", "translation": "중앙 출구가 어디예요?", "pronunciation": "추오구치와 도코데스카", "use": "길을 물을 때", "language": "ja-JP", "translation_language": "ko-KR"}
```

cwkSpark, backend/results.py: feed, finish and the per-line state machine (verbatim, lines 139-188)·python

    def feed(self, text: str) -> list[dict[str, Any]]:
        self._buffer += text
        found: list[dict[str, Any]] = []
        while "\n" in self._buffer:
            line, self._buffer = self._buffer.split("\n", 1)
            found.extend(self._line(line))
        return found

    def finish(self) -> list[dict[str, Any]]:
        """The reply ended: read what is left, including an unterminated line."""
        found: list[dict[str, Any]] = []
        if self._buffer:
            line, self._buffer = self._buffer, ""
            found.extend(self._line(line))
        if self._pending:
            self.malformed += 1
            self._pending = []
        self._in_block = False
        return found

    def _line(self, line: str) -> list[dict[str, Any]]:
        if not self._in_block:
            if _fence_opens(line):
                self._in_block = True
                self.blocks += 1
            else:
                self.prose_lines.append(line)
            return []
        if _fence_closes(line):
            if self._pending:
                self.malformed += 1
                self._pending = []
            self._in_block = False
            return []
        if not line.strip() and not self._pending:
            return []
        self._pending.append(line)
        candidate = "\n".join(self._pending).strip()
        try:
            parsed = json.loads(candidate)
        except json.JSONDecodeError:
            if len(self._pending) >= _MAX_PENDING_LINES or not candidate.startswith("{"):
                self.malformed += 1
                self._pending = []
            return []
        self._pending = []
        if not isinstance(parsed, dict):
            self.malformed += 1
            return []
        return [parsed]

cwkSpark, backend/turns.py: each result recorded once per turn, a change kept as a revision (verbatim, lines 856-893)·python

    async def _record_results(
        self,
        turn: dict[str, Any],
        moment: dict[str, Any],
        objects: list[dict[str, Any]],
        recorded: dict[tuple[str, str], dict[str, Any]],
    ) -> int:
        """Record each new result once per turn. The same result written again
        (a retried stream) is skipped; a changed one is kept as a revision."""
        count = 0
        target = moment.get("target_language") or "ko"
        for obj in objects:
            fields = normalize(obj)
            if fields is None:
                continue
            key = result_key(fields)
            earlier = recorded.get(key)
            if earlier is not None:
                if content_of(earlier) == content_of(fields):
                    continue
                fields["revises"] = earlier["id"]
            if turn.get("target_result"):
                fields.setdefault("about", turn["target_result"])
            record = await self._write(
                self.corpus.add_result,
                moment["id"],
                turn_id=turn["id"],
                target_language=target,
                fields=fields,
            )
            recorded[key] = record
            count += 1
            if record["kind"] == "card" and self.voice is not None:
                # The card is readable now; its voice follows (card first),
                # and the turn finishes only once the voice is recorded.
                voice = asyncio.ensure_future(self._speak_card(moment, record))
                self._turn_work.setdefault(turn["id"], []).append(voice)
        return count

Exercise

Write a streaming parser for your own model output: prose outside a fenced block with a custom info string, one JSON object per line inside. Feed it a canned reply in chunks of 1, 7 and 50 characters and assert the same objects come out each time, and that each object is returned by the feed call that delivered its newline. Then add a reset() and a per-turn map keyed the way your records are identified, and show that a retried stream records nothing twice.

Hint

Keep three pieces of state: a text buffer, whether you are inside the block, and a list of pending lines for an object spread over several lines. Bound that list. For the retry test, script the stream as text, reset, the same text again, then one changed record, and expect one revision pointing at the original.

One reply, two audiences

The parser, a line at a time

From object to record

Shown the moment it exists

Once per turn, and revisions as new results

Code

External links

Exercise

Progress

Comments 0