def METHOD_NAME(
    rule_runner: RuleRunner | None = None, *, notification_names: set[str] | None = None
):
    rule_runner = rule_runner or RuleRunner(rules=bsp_rules())
    notification_names = notification_names or set()
    thread_locals = PyThreadLocals.get_for_current_thread()
    with setup_pipes() as pipes:
        context = BSPContext()
        rule_runner.set_session_values({BSPContext: context})
        conn = BSPConnection(
            rule_runner.scheduler,
            rule_runner.union_membership,
            context,
            pipes.server_reader,
            pipes.server_writer,
        )
        def run_bsp_server():
            thread_locals.set_for_current_thread()
            conn.run()
        bsp_thread = Thread(target=run_bsp_server)
        bsp_thread.daemon = True
        bsp_thread.start()
        client_reader = JsonRpcStreamReader(pipes.client_reader)
        client_writer = JsonRpcStreamWriter(pipes.client_writer)
        notifications = Notifications([])
        endpoint = Endpoint(
            {name: functools.partial(notifications._record, name) for name in notification_names},
            lambda msg: client_writer.write(msg),
        )
        def run_client():
            client_reader.listen(lambda msg: endpoint.consume(msg))
        client_thread = Thread(target=run_client)
        client_thread.daemon = True
        client_thread.start()
        try:
            yield endpoint, notifications
        finally:
            client_reader.close()
            client_writer.close()