Turns Survive a Restart

The request that asks is not the one that waits

When the phone asks for a practical reading of a photo, the engine answers almost at once with {"turn": ..., "created": true}. The reading itself has not happened yet. Pippa may take many seconds, streaming her lines one by one, and the phone may go to sleep in Dad's pocket halfway through. If the turn lived inside that HTTP request, a dropped socket would kill it.

So turns run as engine tasks. The turns module says it in its opening docstring: turns run as engine tasks, not inside the request that asked for them; the phone may background and drop its socket, and the turn goes on. The first code block is the machinery. launch wraps the work in an asyncio task, files it under the turn's id, and removes it when it ends. The route returns, the task keeps running, and every result it records reaches the phone through the moment's event stream, the subject of the next lesson.

A turn is corpus too

A turn is not a variable in memory. Starting one records turn.start, and the mirror holds it with status running. Every line Pippa's reply yields is recorded as a result.add the moment it parses, carrying the turn's id as its lineage. Ending the turn records turn.finish with one of three outcomes: completed, failed or interrupted. finish_turn closes a turn once; called again on a finished turn, it returns the turn as it is and writes nothing.

That makes the restart question answerable from disk. After a crash, the engine can ask its own mirror which turns were still running, and it already holds every result that arrived before the lights went out. A test pins the second half: a stream cut off midway keeps what arrived.

One more field matters. Each turn carries an external request id derived from its own id, spark-<turn id>, and the engine sends it to cwkPippa with the turn. That shared name is how a restarted engine can later ask cwkPippa what became of the turn.

Shutdown lets go; boot settles

shutdown cancels every task the engine is following and waits for them to unwind. Its docstring is the whole policy: stop following turns; cwkPippa keeps answering; the next boot's recovery records how each one ended.

At boot, the engine's lifespan runs in a fixed order: open the corpus (rebuilding the mirror if needed), seed the purposes, register with the family's brain control (a down sibling only defers this; the boot survives), and then call recover. Only after that does it start serving. The second code block sorts the leftover running turns into four fates:

Turn kind	What recovery does	Why
`live`, `exchange`	Records `interrupted`, error "engine restarted"	One-shot calls with no cwkPippa turn to ask after; an utterance cut by a restart is said again at the table
`version`, `review`	Launches it again	Every finished section is already in the corpus, so it resumes at the next one
`picture`	Launches it again	If the Ember job's id was recorded, it goes back to waiting on that job. If the engine died before writing it, the relaunch submits a new job, and a job Ember had already accepted is never waited on
everything else	Asks cwkPippa (`_recover_one`)	The turn may have finished while the engine was down

The resumable kinds come back in the High Quality, Many Brains track; here the point is that each kind's recovery follows from what it already wrote down.

Asking cwkPippa how it ended

The third code block asks cwkPippa for the turn's status by that external request id, and acts on exactly three kinds of answer:

Unknown (404). cwkPippa never received the turn: the engine died before sending it. Recorded as interrupted, with that sentence as the error.
Completed. cwkPippa holds the full reply. The engine parses it, records only the results this turn has not recorded yet (a line is matched by its Vision number, anything else by its text), and finishes the turn as completed with recovered: true and the conversation it ran in.
Anything else. Still running, or cwkPippa unreachable for the moment: wait 5 seconds and ask again, until those waits add up to 600 seconds. Past that, record interrupted with the last state it heard. The counter adds only the sleeps, so the time each status request takes comes on top: a cwkPippa that answers slowly stretches the real wait past ten minutes.

What it never does is send the turn again. The architecture doc states it as a rule: after a restart the engine first asks cwkPippa for the turn's status and never resubmits a turn whose fate is unknown. A resubmission could produce a second answer in the same thread, a second set of results, and a second bill, all to recover from a turn that may already have succeeded.

The tests

One test builds the hard case by hand. It opens a corpus, starts three turns and closes it without finishing any of them: a practical turn that already recorded one line before the "crash", a second practical turn, and a live one. The fake cwkPippa reports the first as completed with a full reply and has never heard of the second. After boot: the first is completed and recovered, holding three results, and the line recorded before the crash appears once; the second and the live one are interrupted.

A second test shrinks the clock. It patches the poll interval to 0.01 seconds and the wait to 0.03, leaves a turn reported as pending, and asserts the turn ends interrupted with pending in its error. Ten minutes of counted sleep, checked in a few hundredths of a second.

Principle: Give long work a name the other side also knows, and write down progress as it happens. Then a restart becomes a question you can ask.

Code

cwkSpark, backend/turns.py: turns as tasks the engine follows, and writes kept off the loop (verbatim, lines 736-757)·python

    def launch(self, key: str, work: Awaitable[None]) -> None:
        task = asyncio.ensure_future(work)
        self._tasks[key] = task
        task.add_done_callback(lambda _done: self._tasks.pop(key, None))

    async def wait_idle(self) -> None:
        """Every launched turn has ended (tests and a graceful stop)."""
        while self._tasks:
            await asyncio.gather(*list(self._tasks.values()), return_exceptions=True)

    async def shutdown(self) -> None:
        """Stop following turns. cwkPippa keeps answering; the next boot's
        recovery records how each one ended."""
        for task in list(self._tasks.values()):
            task.cancel()
        await asyncio.gather(*list(self._tasks.values()), return_exceptions=True)

    # ── helpers ─────────────────────────────────────────────────────

    async def _write(self, fn: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
        # Corpus writes fsync and take a threading lock: keep them off the loop.
        return await asyncio.to_thread(fn, *args, **kwargs)

cwkSpark, backend/turns.py: each leftover turn, by kind (verbatim, lines 1391-1415)·python

    async def recover(self) -> None:
        """Settle the turns a previous engine process left running."""
        for turn in await asyncio.to_thread(self.corpus.running_turns):
            if turn["id"] in self._tasks:
                continue
            if turn.get("kind") in ("live", "exchange"):
                # One-shot calls with no cwkPippa turn to ask after: an
                # utterance cut by a restart is said again at the table.
                await self._write(
                    self.corpus.finish_turn, turn["id"], "interrupted", error="engine restarted"
                )
                continue
            # Versions and reviews ride stateless doors; every section they
            # finished is in the corpus, so they resume at the next one.
            if turn.get("kind") == "version":
                self.launch(turn["id"], self.run_version(turn))
                continue
            if turn.get("kind") == "review":
                self.launch(turn["id"], self.run_review(turn))
                continue
            # A picture whose Ember job was recorded goes back to waiting on it.
            if turn.get("kind") == "picture":
                self.launch(turn["id"], self.run_picture(turn))
                continue
            self.launch(turn["id"], self._recover_one(turn))

cwkSpark, backend/turns.py: ask cwkPippa, never resubmit (verbatim, lines 1417-1466)·python

    async def _recover_one(self, turn: dict[str, Any]) -> None:
        waited = 0.0
        while True:
            try:
                status = await self.pippa.turn_status(turn["external_request_id"])
            except PippaError as exc:
                status = {"status": "unreachable", "error": str(exc)}
            state = (status or {}).get("status")
            if status is None:
                await self._write(
                    self.corpus.finish_turn,
                    turn["id"],
                    "interrupted",
                    error="the engine restarted before cwkPippa received the turn",
                )
                return
            if state == "completed":
                moment = await asyncio.to_thread(
                    self.corpus.moment, turn["moment_id"], include_deleted=True
                )
                if moment is None:
                    await self._fail(turn, "the moment is gone")
                    return
                recorded = {
                    result_key(item): item
                    for item in moment.get("results") or []
                    if item.get("turn_id") == turn["id"]
                }
                objects, prose = parse_reply(str(status.get("result") or ""))
                await self._record_results(turn, moment, objects, recorded)
                await self._write(
                    self.corpus.finish_turn,
                    turn["id"],
                    "completed",
                    prose=prose,
                    results=len(recorded),
                    recovered=True,
                    conversation_id=status.get("conversation_id"),
                )
                return
            if waited >= RECOVERY_WAIT_SECONDS:
                await self._write(
                    self.corpus.finish_turn,
                    turn["id"],
                    "interrupted",
                    error=f"cwkPippa still reports the turn as {state or 'unknown'}",
                )
                return
            await asyncio.sleep(RECOVERY_POLL_SECONDS)
            waited += RECOVERY_POLL_SECONDS

Exercise

In a web service of yours, move one slow operation out of its request. The route should record a job as running, start a background task, and return the job at once. Give the job an external id derived from its own id and pass it to whatever does the slow work. Then write the recovery: at startup, find jobs still marked running and settle each one by asking the worker about the external id, covering three answers (never received, finished, still going). Test it by starting jobs, stopping the service before they finish, and restarting it.

Hint

Keep the task objects in a dictionary keyed by job id and remove them in a done callback; besides letting recovery skip a job already being followed, it keeps a strong reference so the event loop cannot drop the task. Record partial output as it arrives, and on a finished answer record only what is missing.

The request that asks is not the one that waits

A turn is corpus too

Shutdown lets go; boot settles

Asking cwkPippa how it ended

The tests

Code

External links

Exercise

Progress

Comments 0