id=str(run.id),
                        name=run.name,
                        type=run.run_type,
                        tags=run.tags or [],
                        metadata=(run.extra or {}).get("metadata", {}),
                        start_time=run.start_time.isoformat(timespec="milliseconds"),
                        streamed_output_str=[],
                        final_output=None,
                        end_time=None,
                    ),
                }
            )
        )
    def _on_run_update(self, run: Run) -> None:
        """Finish a run."""
        try:
            index = self._key_map_by_run_id.get(run.id)
            if index is None:
                return
            self.send_stream.send_nowait(
                RunLogPatch(
                    {
                        "op": "add",
                        "path": f"/logs/{index}/final_output",
                        # to undo the dumpd done by some runnables / tracer / etc
                        "value": load(run.outputs),
                    },
                    {
                        "op": "add",
                        "path": f"/logs/{index}/end_time",
                        "value": run.end_time.isoformat(timespec="milliseconds")
                        if run.end_time is not None
                        else None,
                    },
                )
            )
        finally:
            if run.id == self.root_id:
                if self.auto_close:
                    self.send_stream.close()
    def _on_llm_new_token(
        self,
        run: Run,
        token: str,
        chunk: Optional[Union[GenerationChunk, ChatGenerationChunk]],
    ) -> None:
        """Process new LLM token."""
        index = self._key_map_by_run_id.get(run.id)
        if index is None:
            return
        self.send_stream.send_nowait(
            RunLogPatch(