ronantakizawa/codereview-bench huggingface dataset:
Repository: libsql/libsql
File: bindings/c/src/lib.rs (Rust)

Code before:
) -> std::ffi::c_int {
    if conn.is_null() {
        set_err_msg("Null connection".to_string(), out_err_msg);
        return 1;
    }
    let conn = conn.get_ref();
    if let Err(err) = conn.set_reserved_bytes(reserved_bytes) {
        set_err_msg(err.to_string(), out_err_msg);
    }
    0
}

#[no_mangle]
pub unsafe extern "C" fn libsql_get_reserved_bytes(
    conn: libsql_connection_t,
    reserved_bytes: *mut i32,
    out_err_msg: *mut *const std::ffi::c_char,
) -> std::ffi::c_int {
    if conn.is_null() {
        set_err_msg("Null connection".to_string(), out_err_msg);
        return 1;
    }
    let conn = conn.get_ref();
    match conn.get_reserved_bytes() {
        Ok(v) => *reserved_bytes = v,
        Err(err) => set_err_msg(err.to_string(), out_err_msg),
    }
    0
}

#[no_mangle]
pub unsafe extern "C" fn libsql_reset(
    conn: libsql_connection_t,
    out_err_msg: *mut *const std::ffi::c_char,
) -> std::ffi::c_int {
    if conn.is_null() {
        set_err_msg("Null connection".to_string(), out_err_msg);
        return 1;
    }
    let conn = conn.get_ref();
    RT.block_on(conn.reset());
    0
}

#[no_mangle]
pub unsafe extern "C" fn libsql_disconnect(conn: libsql_connection_t) {
    if conn.is_null() {
        return;
    }
    let conn = unsafe { Box::from_raw(conn.get_ref_mut()) };
    RT.spawn_blocking(|| {

Review comment (suggestion):
we should return `1` if the error is set right? but this returns `0` regardless 

Code after:
) -> std::ffi::c_int {
    if conn.is_null() {
        set_err_msg("Null connection".to_string(), out_err_msg);
        return 1;
    }
    let conn = conn.get_ref();
    if let Err(err) = conn.set_reserved_bytes(reserved_bytes) {
        set_err_msg(err.to_string(), out_err_msg);
        return 1;
    }
    0
}

#[no_mangle]
pub unsafe extern "C" fn libsql_get_reserved_bytes(
    conn: libsql_connection_t,
    reserved_bytes: *mut i32,
    out_err_msg: *mut *const std::ffi::c_char,
) -> std::ffi::c_int {
    if conn.is_null() {
        set_err_msg("Null connection".to_string(), out_err_msg);
        return 1;
    }
    let conn = conn.get_ref();
    match conn.get_reserved_bytes() {
        Ok(v) => *reserved_bytes = v,
        Err(err) => {
            set_err_msg(err.to_string(), out_err_msg);
            return 1;
        }
    }
    0
}

#[no_mangle]
pub unsafe extern "C" fn libsql_reset(
    conn: libsql_connection_t,
    out_err_msg: *mut *const std::ffi::c_char,
) -> std::ffi::c_int {
    if conn.is_null() {
        set_err_msg("Null connection".to_string(), out_err_msg);
        return 1;
    }
    let conn = conn.get_ref();
    RT.block_on(conn.reset());
    0
}

#[no_mangle]
pub unsafe extern "C" fn libsql_disconnect(conn: libsql_connection_t) {
    if conn.is_null() {
Repository: livekit/agents
File: examples/avatar_agents/avatartalk/agent_worker.py (Python)

Code before:
from livekit.agents import Agent, AgentSession, JobContext, WorkerOptions, WorkerType, cli
from livekit.plugins import avatartalk, openai

load_dotenv()


async def entrypoint(ctx: JobContext):
    await ctx.connect()

    session = AgentSession(
        llm=openai.realtime.RealtimeModel(
            voice="ash",
            turn_detection=TurnDetection(
                type="server_vad",
                threshold=0.5,
                prefix_padding_ms=300,
                silence_duration_ms=500,
                create_response=True,
                interrupt_response=True,
            ),
        )
    )

    avatar = avatartalk.AvatarSession(
        api_url=os.getenv("AVATARTALK_API_URL"),
        avatar_id=os.getenv("AVATARTALK_AVATAR_NAME"),
        avatar_expression=os.getenv("AVATARTALK_AVATAR_EXPRESSION"),
        api_secret=os.getenv("AVATARTALK_API_KEY"),
    )

    await avatar.start(session, room=ctx.room)
    await session.start(
        agent=Agent(
            instructions=(
                "You are a helpful AI assistant designed to "
                "communicate with users in multiple languages. "
                "Your primary directive is to always respond in "
                "the same language that the user writes in."
            )
        ),
        room=ctx.room,
    )


if __name__ == "__main__":
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint, worker_type=WorkerType.ROOM))


Review comment (suggestion):
```suggestion
        avatar_name=os.getenv("AVATARTALK_AVATAR_NAME"),
```

Code after:
from livekit.agents import Agent, AgentSession, JobContext, WorkerOptions, WorkerType, cli
from livekit.plugins import avatartalk, openai

load_dotenv()


async def entrypoint(ctx: JobContext):
    await ctx.connect()

    session = AgentSession(
        llm=openai.realtime.RealtimeModel(
            voice="ash",
            turn_detection=TurnDetection(
                type="server_vad",
                threshold=0.5,
                prefix_padding_ms=300,
                silence_duration_ms=500,
                create_response=True,
                interrupt_response=True,
            ),
        )
    )

    avatar = avatartalk.AvatarSession(
        api_url=os.getenv("AVATARTALK_API_URL"),
        avatar=os.getenv("AVATARTALK_AVATAR"),
        emotion=os.getenv("AVATARTALK_EMOTION"),
        api_secret=os.getenv("AVATARTALK_API_KEY"),
    )

    await avatar.start(session, room=ctx.room)
    await session.start(
        agent=Agent(
            instructions=(
                "You are a helpful AI assistant designed to "
                "communicate with users in multiple languages. "
                "Your primary directive is to always respond in "
                "the same language that the user writes in."
            )
        ),
        room=ctx.room,
    )


if __name__ == "__main__":
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint, worker_type=WorkerType.ROOM))
Repository: livekit/agents
File: livekit-agents/livekit/agents/ipc/supervised_proc.py (Python)

Code before:
            yield
        finally:
            signal.pthread_sigmask(signal.SIG_UNBLOCK, [signal.SIGINT])
    else:
        old = signal.signal(signal.SIGINT, signal.SIG_IGN)
        try:
            yield
        finally:
            signal.signal(signal.SIGINT, old)


def _dump_stack_traces_impl() -> None:
    """Implementation of stack trace dumping (callable directly or from signal handler)."""
    import asyncio
    import faulthandler
    import tempfile
    import traceback
    from multiprocessing import current_process
    from pathlib import Path

    import psutil

    if os.getenv("LK_DUMP_STACK_TRACES", "0").lower() in ("0", "false", "no"):
        return

    dir: str = os.getenv("LK_DUMP_DIR", "/tmp")
    Path(dir).mkdir(parents=True, exist_ok=True)

    with tempfile.NamedTemporaryFile(
        mode="w",
        dir=dir,
        delete=False,
        prefix=f"livekit-agents-pid-{current_process().pid}-{time.time_ns()}-",
        suffix=".stacktrace",
    ) as f:
        print(f"\n{'=' * 60}", file=f)
        print(
            f"Process {current_process().name} (pid {current_process().pid}) stack trace dump",
            file=f,
        )
        print(f"{'=' * 60}\n", file=f)

        faulthandler.dump_traceback(file=f, all_threads=True)
        print("\n", file=f)

        try:
            try:
                loop = asyncio.get_running_loop()
            except RuntimeError:
                loop = None


Review comment (suggestion):
```suggestion
    dir = os.getenv("LK_DUMP_DIR") or tempfile.gettempdir()
```

Code after:
        finally:
            signal.pthread_sigmask(signal.SIG_UNBLOCK, [signal.SIGINT])
    else:
        old = signal.signal(signal.SIGINT, signal.SIG_IGN)
        try:
            yield
        finally:
            signal.signal(signal.SIGINT, old)


@dataclass
class _ProcOpts:
    initialize_timeout: float
    close_timeout: float
    memory_warn_mb: float
    memory_limit_mb: float
    ping_interval: float
    ping_timeout: float
    high_ping_threshold: float
    http_proxy: str | None


class SupervisedProc(ABC):
    def __init__(
        self,
        *,
        initialize_timeout: float,
        close_timeout: float,
        memory_warn_mb: float,
        memory_limit_mb: float,
        ping_interval: float,
        ping_timeout: float,
        high_ping_threshold: float,
        http_proxy: str | None,
        mp_ctx: BaseContext,
        loop: asyncio.AbstractEventLoop,
    ) -> None:
        self._loop = loop
        self._mp_ctx = mp_ctx
        self._opts = _ProcOpts(
            initialize_timeout=initialize_timeout,
            close_timeout=close_timeout,
            memory_warn_mb=memory_warn_mb,
            memory_limit_mb=memory_limit_mb,
            ping_interval=ping_interval,
            ping_timeout=ping_timeout,
            high_ping_threshold=high_ping_threshold,
            http_proxy=http_proxy,
        )

        self._exitcode: int | None = None
Repository: livekit/agents
File: livekit-agents/livekit/agents/ipc/job_proc_lazy_main.py (Python)

Code before:
from __future__ import annotations

from multiprocessing import current_process

from .supervised_proc import _dump_stack_traces

if current_process().name == "job_proc":
    import signal

    if hasattr(signal, "SIGUSR1"):
        signal.signal(signal.SIGUSR1, _dump_stack_traces)
    # ignore signals in the jobs process (the parent process will handle them)
    signal.signal(signal.SIGINT, signal.SIG_IGN)
    signal.signal(signal.SIGTERM, signal.SIG_IGN)

import asyncio
import contextlib
import socket
from collections.abc import Awaitable
from dataclasses import dataclass
from typing import Any, Callable, cast

from opentelemetry import trace

from livekit import rtc

from ..job import JobContext, JobExecutorType, JobProcess, _JobContextVar
from ..log import logger
from ..telemetry import trace_types, tracer
from ..utils import aio, http_context, log_exceptions, shortuuid

Review comment (suggestion):
Ideally we import everything lazily inside this file, so
```
    signal.signal(signal.SIGINT, signal.SIG_IGN)
    signal.signal(signal.SIGTERM, signal.SIG_IGN)
```
are executed as soon as possible

Code after:
from __future__ import annotations

from multiprocessing import current_process

if current_process().name == "job_proc":
    import signal

    # ignore signals in the jobs process (the parent process will handle them)
    signal.signal(signal.SIGINT, signal.SIG_IGN)
    signal.signal(signal.SIGTERM, signal.SIG_IGN)

    if hasattr(signal, "SIGUSR1"):
        from .proc_client import _dump_stack_traces

        signal.signal(signal.SIGUSR1, _dump_stack_traces)

import asyncio
import contextlib
import socket
from collections.abc import Awaitable
from dataclasses import dataclass
from typing import Any, Callable, cast

from opentelemetry import trace

from livekit import rtc

from ..job import JobContext, JobExecutorType, JobProcess, _JobContextVar
from ..log import logger
from ..telemetry import trace_types, tracer
Repository: livekit/agents
File: livekit-agents/livekit/agents/llm/llm.py (Python)

Code before:
from ..utils import aio
from .chat_context import ChatContext, ChatRole
from .tool_context import FunctionTool, RawFunctionTool, ToolChoice


class CompletionUsage(BaseModel):
    completion_tokens: int
    """The number of tokens in the completion."""
    prompt_tokens: int
    """The number of input tokens used (includes cached tokens)."""
    prompt_cached_tokens: int = 0
    """The number of cached input tokens used."""
    cache_creation_tokens: int = 0
    """The number of tokens used to create the cache."""
    cache_read_tokens: int = 0
    """The number of tokens read from the cache."""
    total_tokens: int
    """The total number of tokens used (completion + prompt tokens)."""


class FunctionToolCall(BaseModel):
    type: Literal["function"] = "function"
    name: str
    arguments: str
    call_id: str


class ChoiceDelta(BaseModel):
    role: ChatRole | None = None
    content: str | None = None
    tool_calls: list[FunctionToolCall] = Field(default_factory=list)


class ChatChunk(BaseModel):
    id: str
    delta: ChoiceDelta | None = None
    usage: CompletionUsage | None = None


class LLMError(BaseModel):
    model_config = ConfigDict(arbitrary_types_allowed=True)
    type: Literal["llm_error"] = "llm_error"
    timestamp: float
    label: str
    error: Exception = Field(..., exclude=True)
    recoverable: bool


TEvent = TypeVar("TEvent")



Review comment (suggestion):
I would still name this `extra` IMO
```suggestion
    extra: dict[str, Any] | None = None
```

Code after:
from ..utils import aio
from .chat_context import ChatContext, ChatRole
from .tool_context import FunctionTool, RawFunctionTool, ToolChoice


class CompletionUsage(BaseModel):
    completion_tokens: int
    """The number of tokens in the completion."""
    prompt_tokens: int
    """The number of input tokens used (includes cached tokens)."""
    prompt_cached_tokens: int = 0
    """The number of cached input tokens used."""
    cache_creation_tokens: int = 0
    """The number of tokens used to create the cache."""
    cache_read_tokens: int = 0
    """The number of tokens read from the cache."""
    total_tokens: int
    """The total number of tokens used (completion + prompt tokens)."""


class FunctionToolCall(BaseModel):
    type: Literal["function"] = "function"
    name: str
    arguments: str
    call_id: str
    extra: dict[str, Any] | None = None
    """Provider-specific extra data (e.g., Google thought signatures)."""


class ChoiceDelta(BaseModel):
    role: ChatRole | None = None
    content: str | None = None
    tool_calls: list[FunctionToolCall] = Field(default_factory=list)
    extra: dict[str, Any] | None = None
    """Provider-specific extra data (e.g., Google thought signatures)."""


class ChatChunk(BaseModel):
    id: str
    delta: ChoiceDelta | None = None
    usage: CompletionUsage | None = None


class LLMError(BaseModel):
    model_config = ConfigDict(arbitrary_types_allowed=True)
    type: Literal["llm_error"] = "llm_error"
    timestamp: float
    label: str
    error: Exception = Field(..., exclude=True)
    recoverable: bool
Repository: livekit/agents
File: livekit-agents/livekit/agents/llm/llm.py (Python)

Code before:
    prompt_tokens: int
    """The number of input tokens used (includes cached tokens)."""
    prompt_cached_tokens: int = 0
    """The number of cached input tokens used."""
    cache_creation_tokens: int = 0
    """The number of tokens used to create the cache."""
    cache_read_tokens: int = 0
    """The number of tokens read from the cache."""
    total_tokens: int
    """The total number of tokens used (completion + prompt tokens)."""


class FunctionToolCall(BaseModel):
    type: Literal["function"] = "function"
    name: str
    arguments: str
    call_id: str


class ChoiceDelta(BaseModel):
    role: ChatRole | None = None
    content: str | None = None
    tool_calls: list[FunctionToolCall] = Field(default_factory=list)


class ChatChunk(BaseModel):
    id: str
    delta: ChoiceDelta | None = None
    usage: CompletionUsage | None = None


class LLMError(BaseModel):
    model_config = ConfigDict(arbitrary_types_allowed=True)
    type: Literal["llm_error"] = "llm_error"
    timestamp: float
    label: str
    error: Exception = Field(..., exclude=True)
    recoverable: bool


TEvent = TypeVar("TEvent")


class LLM(
    ABC,
    rtc.EventEmitter[Union[Literal["metrics_collected", "error"], TEvent]],
    Generic[TEvent],
):
    def __init__(self) -> None:
        super().__init__()
        self._label = f"{type(self).__module__}.{type(self).__name__}"

Review comment (suggestion):
Same here
```suggestion
    extra: dict[str, Any] | None = None
```

Code after:
    prompt_tokens: int
    """The number of input tokens used (includes cached tokens)."""
    prompt_cached_tokens: int = 0
    """The number of cached input tokens used."""
    cache_creation_tokens: int = 0
    """The number of tokens used to create the cache."""
    cache_read_tokens: int = 0
    """The number of tokens read from the cache."""
    total_tokens: int
    """The total number of tokens used (completion + prompt tokens)."""


class FunctionToolCall(BaseModel):
    type: Literal["function"] = "function"
    name: str
    arguments: str
    call_id: str
    extra: dict[str, Any] | None = None
    """Provider-specific extra data (e.g., Google thought signatures)."""


class ChoiceDelta(BaseModel):
    role: ChatRole | None = None
    content: str | None = None
    tool_calls: list[FunctionToolCall] = Field(default_factory=list)
    extra: dict[str, Any] | None = None
    """Provider-specific extra data (e.g., Google thought signatures)."""


class ChatChunk(BaseModel):
    id: str
    delta: ChoiceDelta | None = None
    usage: CompletionUsage | None = None


class LLMError(BaseModel):
    model_config = ConfigDict(arbitrary_types_allowed=True)
    type: Literal["llm_error"] = "llm_error"
    timestamp: float
    label: str
    error: Exception = Field(..., exclude=True)
    recoverable: bool


TEvent = TypeVar("TEvent")


class LLM(
    ABC,
    rtc.EventEmitter[Union[Literal["metrics_collected", "error"], TEvent]],
    Generic[TEvent],
Repository: livekit/agents
File: examples/voice_agents/session_close_callback.py (Python)

Code before:
import logging

from dotenv import load_dotenv

from livekit.agents import Agent, AgentServer, AgentSession, CloseEvent, JobContext, cli
from livekit.agents.beta.tools import EndCallTool
from livekit.plugins import silero

logger = logging.getLogger("my-worker")
logger.setLevel(logging.INFO)

load_dotenv()


# This example shows how to close the agent session when the linked participant disconnects
# or when the worker is shutting down. When closing the session, agent will be interrupted
# and the last agent message will be added to the chat context.


class MyAgent(Agent):
    def __init__(self):
        super().__init__(
            instructions="You are a helpful assistant.",
            tools=[EndCallTool(end_instructions="thanks the user for calling and goodbye")],
        )

    async def on_enter(self) -> None:
        self.session.generate_reply(instructions="say hello to the user")


server = AgentServer()


@server.rtc_session()
async def entrypoint(ctx: JobContext):
    session = AgentSession(
        stt="assemblyai/universal-streaming",
        llm="openai/gpt-4.1-mini",
        tts="elevenlabs",
        vad=silero.VAD.load(),
    )

    await session.start(agent=MyAgent(), room=ctx.room)

    @session.on("close")
    def on_close(ev: CloseEvent):
        print(f"Agent Session closed, reason: {ev.reason}")
        print("=" * 20)
        print("Chat History:")

Review comment (suggestion):
```suggestion
            tools=[EndCallTool(end_instructions="thanks the user for calling and tell them goodbye")],
```

Code after:
import logging

from dotenv import load_dotenv

from livekit.agents import Agent, AgentServer, AgentSession, CloseEvent, JobContext, cli
from livekit.agents.beta.tools import EndCallTool
from livekit.plugins import google, silero  # noqa: F401

logger = logging.getLogger("my-worker")
logger.setLevel(logging.INFO)

load_dotenv()


# This example shows how to close the agent session when the linked participant disconnects
# or when the worker is shutting down. When closing the session, agent will be interrupted
# and the last agent message will be added to the chat context.


class MyAgent(Agent):
    def __init__(self):
        super().__init__(
            instructions="You are a helpful assistant.",
            stt="assemblyai/universal-streaming",
            llm="openai/gpt-4.1-mini",
            tts="cartesia/sonic-3",
            # llm=google.realtime.RealtimeModel(),
            tools=[
                EndCallTool(
                    end_instructions="thanks the user for calling and tell them goodbye",
                    delete_room=True,  # this will disconnect all remote participants, including SIP callers
                )
            ],
        )

    async def on_enter(self) -> None:
        self.session.generate_reply(instructions="say hello to the user")


server = AgentServer()


@server.rtc_session()
async def entrypoint(ctx: JobContext):
    session = AgentSession(
        vad=silero.VAD.load(),
    )

    await session.start(agent=MyAgent(), room=ctx.room)
Repository: livekit/agents
File: livekit-agents/livekit/agents/utils/http_server.py (Python)

Code before:
        self._app = web.Application()
        self._runner: web.AppRunner | None = None
        self._site: web.TCPSite | None = None

    @property
    def app(self) -> web.Application:
        return self._app

    @property
    def host(self) -> str:
        return self._host

    @property
    def port(self) -> int:
        return self._port

    async def start(self) -> None:
        self._runner = web.AppRunner(self._app)
        await self._runner.setup()
        self._site = web.TCPSite(self._runner, self._host, self._port)
        await self._site.start()

        if self._port == 0:
            server = self._site._server
            if isinstance(server, asyncio.Server) and server.sockets:
                self._port = server.sockets[0].getsockname()[1]

    async def aclose(self) -> None:
        if self._runner is not None:
            await self._runner.cleanup()


Review comment (suggestion):
```suggestion
            addresses = self._runner.addresses
            if addresses:
                self._port = addresses[0][1]
```

then maybe we don't need to store `self._site`

Code after:

    @property
    def app(self) -> web.Application:
        return self._app

    @property
    def host(self) -> str:
        return self._host

    @property
    def port(self) -> int:
        return self._port

    async def start(self) -> None:
        self._runner = web.AppRunner(self._app)
        await self._runner.setup()
        site = web.TCPSite(self._runner, self._host, self._port)
        await site.start()

        if self._port == 0:
            address = self._runner.addresses
            if address:
                self._port = address[0][1]

    async def aclose(self) -> None:
        if self._runner is not None:
            await self._runner.cleanup()
            self._runner = None
Repository: livekit/agents
File: livekit-agents/livekit/agents/utils/http_server.py (Python)

Code before:
        self._runner: web.AppRunner | None = None
        self._site: web.TCPSite | None = None

    @property
    def app(self) -> web.Application:
        return self._app

    @property
    def host(self) -> str:
        return self._host

    @property
    def port(self) -> int:
        return self._port

    async def start(self) -> None:
        self._runner = web.AppRunner(self._app)
        await self._runner.setup()
        self._site = web.TCPSite(self._runner, self._host, self._port)
        await self._site.start()

        if self._port == 0:
            server = self._site._server
            if isinstance(server, asyncio.Server) and server.sockets:
                self._port = server.sockets[0].getsockname()[1]

    async def aclose(self) -> None:
        if self._runner is not None:
            await self._runner.cleanup()


Review comment (suggestion):
nitpicking:  
```suggestion
            await self._runner.cleanup()
            self._runner = None
```

Code after:
    @property
    def app(self) -> web.Application:
        return self._app

    @property
    def host(self) -> str:
        return self._host

    @property
    def port(self) -> int:
        return self._port

    async def start(self) -> None:
        self._runner = web.AppRunner(self._app)
        await self._runner.setup()
        site = web.TCPSite(self._runner, self._host, self._port)
        await site.start()

        if self._port == 0:
            address = self._runner.addresses
            if address:
                self._port = address[0][1]

    async def aclose(self) -> None:
        if self._runner is not None:
            await self._runner.cleanup()
            self._runner = None