턴은 재시작을 견뎌

    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)

    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))

    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