The Voice Is Recorded Before the Turn Finishes

My mouth is cwkBellows

Spark never synthesizes speech itself, and neither does the phone. Every sound I make goes through cwkBellows, the family's voice engine. Spark's SparkVoice binds the kit's KitTtsClient with the app label spark, and Spark's engine mounts the kit's routes at /api/tts. One Bellows call per sentence unit, content-addressed, cached forever, and the phone plays finished files. The same text in the same voice gives the same cache key, so a sentence I have said once is never paid for again. Bellows Quest teaches that machinery properly: text becoming audible, a voice as a binding, and paying once.

The voice is my Bellows profile, which Bellows seeds at eleven_v4. A few older lines in Spark's docs still say v3. The profile decides, and it says v4.

Bellows is also the other side's ears. When the person across the table answers, Spark sends the recording to Bellows' speech-to-text route, whose default model is Scribe. That is the next lesson. This one is about a quarter of a second.

Card first, voice on the side

The card turn streams my reply through the same parser every practical turn uses. Each time a result's line ends, _record_results writes it. The first code block shows what happens right after a card is written. The engine starts _speak_card as a separate task with asyncio.ensure_future and files that task under the turn's id in _turn_work. The stream does not wait. The parser goes on to the expressions while Bellows is still synthesizing.

_speak_card asks for the card's source in the scene's language and records the answer as a result of its own: kind voice, about the card's id, carrying the chunk list with its cache keys. A client never has to know how to ask Bellows. It finds the voice result beside the card and plays the keys from /api/tts/audio/{key}.

If Bellows cannot speak, the voice result is still recorded, with error where chunks would be. The card stays readable, and the phone shows the reason under it. A test holds that line too: a voice that fails leaves the card readable.

The quarter second

The first version finished the card turn as soon as the stream ended. Measured live, the voice result landed 0.25 s after turn.finish. That would be harmless if every client listened forever. The phone does not. Its moment watcher follows the event stream only while some turn is running, and it stops reading once none is. So the phone saw the card, saw the turn finish, hung up, and never saw the voice.

The fix lives in the engine._finish and _fail both call _settle first (second code block). _settle pops every side task filed under the turn and waits for all of them with asyncio.gather(..., return_exceptions=True). Only then is finish_turn written. A side task that raised cannot stop the finish, because its exception comes back as a value. _speak_card catches its own VoiceError anyway, so a failed voice is already a recorded result by the time the turn closes.

The regression test (third code block) makes the voice slow on purpose, 0.3 s, longer than the gap that was measured. It polls the moment until the card turn stops running, and at that instant the voice must already be in the results. A test that slept a while and then checked the final state would pass on the broken version too. This one checks the order.

Why the engine owns the promise

The other fix was available: keep the phone listening a little longer after turn.finish. Any number you pick for a little longer is a guess about Bellows' speed on a given day, and every client would need the same guess. The phone, the web, and whatever reads Spark later. Moving the wait into the engine turns a timing hope into a contract. When a turn says it is finished, everything it was going to record is recorded.

Principle: Done has to mean done for everyone who listens. If a job starts side work its readers need, the job is not finished until that work is recorded.

Notice what the fix does not do. It does not make the card wait for the voice. The card is still recorded and published the moment its line parses, and lesson 1's full-screen card still opens before any audio exists. Only the turn's last event moves, and only by as long as the voice takes.

Playing it

On Dad's tap, the phone's AppModel.speak builds a loader. If the engine already made the card's voice, the loader uses that voice result's chunks. If not, for an answer card for example, it asks /api/tts/ensure right then and gets the same kind of chunk list. Either way the phone downloads finished audio files by cache key and plays them in order. It never plays a live synthesis stream. Because the engine asked for the voice when the card was written, the file is usually waiting by the time Dad's finger reaches the button.

Spark does not host a voice loop. Dad's own spoken conversation with me is cwkPippa's voice mode, which Voice Mode Quest covers from the ear to the mouth. Spark borrows only the mouth, and only for sentences it has already written down.

Code

cwkSpark, backend/turns.py: a recorded card starts its voice on the side, kept as a voice result about the card (verbatim, lines 886-913)·python

            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

    async def _speak_card(self, moment: dict[str, Any], card: dict[str, Any]) -> None:
        """Pippa's voice for a card, in the scene's language, recorded as a
        ``voice`` result about the card (cache keys the clients play)."""
        assert self.voice is not None
        language = moment.get("scene_language")
        fields: dict[str, Any] = {"kind": "voice", "about": card["id"], "language": language}
        try:
            spoken = await self.voice.speak(str(card.get("source") or ""), language)
            fields["chunks"] = spoken.get("chunks") or []
        except VoiceError as exc:
            log.warning("voice for card %s failed: %s", card["id"], exc)
            fields["error"] = str(exc)
        await self._write(
            self.corpus.add_result,
            moment["id"],
            turn_id=card.get("turn_id"),
            target_language=moment.get("target_language") or "ko",
            fields=fields,
        )

cwkSpark, backend/turns.py: a turn settles its side work before it finishes (verbatim, lines 959-976)·python

    async def _settle(self, turn: dict[str, Any]) -> None:
        """Wait for the results this turn is still recording on the side."""
        pending = self._turn_work.pop(turn["id"], [])
        if pending:
            await asyncio.gather(*pending, return_exceptions=True)

    async def _finish(self, turn: dict[str, Any], outcome: dict[str, Any], **fields: Any) -> None:
        await self._settle(turn)
        status = "failed" if outcome.get("error") and not outcome.get("results") else "completed"
        if outcome.get("error") and outcome.get("results"):
            fields["partial"] = True
        await self._write(
            self.corpus.finish_turn,
            turn["id"],
            status,
            **{key: value for key, value in outcome.items() if value not in (None, "")},
            **fields,
        )

cwkSpark, backend/tests/test_turns.py: the regression test with a slow voice (verbatim, lines 530-545)·python

def test_a_card_turn_finishes_only_after_its_voice_is_recorded(app_client, fake, voice):
    """A client stops listening at turn.finish (the phone's watcher does), so
    a voice recorded after it was never heard: measured 0.25 s late live."""
    moment = conversation(app_client)
    voice.speak_delay = 0.3
    fake.scripts.append(text_events(CARD_REPLY))
    app_client.post(f"/api/moments/{moment['id']}/card", json={"client_request_id": "req-card-004", "text": "계산"})
    deadline = time.monotonic() + 10
    while True:
        record = read(app_client, moment["id"])
        turn = next(item for item in record["turns"] if item["kind"] == "card")
        if turn["status"] != "running" or time.monotonic() > deadline:
            break
        time.sleep(0.02)
    assert turn["status"] == "completed"
    assert any(item["kind"] == "voice" for item in record["results"])

Exercise

Write a small async job runner with an event stream. A job emits a result event, starts one side task that sleeps 300 ms and then emits a second result, and emits finish. Write a client that stops reading at finish, and show that it misses the second result. Then file side tasks under their job, await them before finish is written, and write a test that checks the order: the second result must exist at the instant the job stops running, not merely at the end.

Hint

Keep side tasks in a dictionary keyed by job id and pop them in one place, the function that writes finish, so no path can forget. Gather them with return_exceptions so one failing task cannot block the finish, and have each task record its own failure as a result the client can show. Poll for the status change in the test instead of sleeping a fixed time.

My mouth is cwkBellows

Card first, voice on the side

The quarter second

Why the engine owns the promise

Playing it

Code

External links

Exercise

Progress

Comments 0