include = include and run.name not in self.exclude_names
        if self.exclude_types is not None:
            include = include and run.run_type not in self.exclude_types
        if self.exclude_tags is not None:
            include = include and all(tag not in self.exclude_tags for tag in run_tags)
        return include
    def _persist_run(self, run: Run) -> None:
        # This is a legacy method only called once for an entire run tree
        # therefore not useful here
        pass
    def _on_run_create(self, run: Run) -> None:
        """Start a run."""
        if self.root_id is None:
            self.root_id = run.id
            self.send_stream.send_nowait(
                RunLogPatch(
                    {
                        "op": "replace",
                        "path": "",
                        "value": RunState(
                            id=str(run.id),
                            streamed_output=[],
                            final_output=None,
                            logs={},
                        ),
                    }
                )
            )
        if not self.include_run(run):
            return
        # Determine previous index, increment by 1
        with self.lock:
            self._counter_map_by_name[run.name] += 1
            count = self._counter_map_by_name[run.name]
            self._key_map_by_run_id[run.id] = (
                run.name if count == 1 else f"{run.name}:{count}"
            )
        # Add the run to the stream
        self.send_stream.send_nowait(
            RunLogPatch(
                {
                    "op": "add",
                    "path": f"/logs/{self._key_map_by_run_id[run.id]}",
                    "value": LogEntry(
                        id=str(run.id),
                        name=run.name,