# Call the runnable in streaming mode,
        # add each chunk to the output stream
        async def consume_astream() -> None:
            try:
                prev_final_output: Optional[Output] = None
                final_output: Optional[Output] = None
                async for chunk in self.astream(input, config, **kwargs):
                    prev_final_output = final_output
                    if final_output is None:
                        final_output = chunk
                    else:
                        try:
                            final_output = final_output + chunk  # type: ignore
                        except TypeError:
                            final_output = chunk
                    patches: List[Dict[str, Any]] = []
                    if with_streamed_output_list:
                        patches.append(
                            {
                                "op": "add",
                                "path": "/streamed_output/-",
                                # chunk cannot be shared between
                                # streamed_output and final_output
                                # otherwise jsonpatch.apply will
                                # modify both
                                "value": deepcopy(chunk),
                            }
                        )
                    for op in jsonpatch.JsonPatch.from_diff(
                        prev_final_output, final_output, dumps=dumps
                    ):
                        patches.append({**op, "path": f"/final_output{op['path']}"})
                    await stream.send_stream.send(RunLogPatch(*patches))
            finally:
                await stream.send_stream.aclose()
        # Start the runnable in a task, so we can start consuming output
        task = asyncio.create_task(consume_astream())
        try:
            # Yield each chunk from the output stream
            if diff:
                async for log in stream:
                    yield log
            else:
                state = RunLog(state=None)  # type: ignore[arg-type]
                async for log in stream:
                    state = state + log
                    yield state
        finally: