SIGTERM does not shut down workers when request_fnc blocks the event loop

[bml1g12]

Bug Description

Worker SIGTERM handler raises _ExitCli inside job-request tasks, preventing graceful drain

Component: livekit-agents worker CLI (livekit/agents/cli/cli.py)

Impact: SIGTERM/SIGINT graceful shutdown is unreliable for workers whose request_fnc does synchronous/blocking I/O (HTTP, database, file locks, etc.). The process ignores orchestrator shutdown, continues to register as available, and may accept new job dispatches until SIGKILL.

We observed this issue in real production ECS self hosted deployment on an old version (1.6.0) but I have provided a minimal example reproducing the issue on latest version. The rest of this ticket I admit is LLM generated text, so please take the suggested solution with a pinch of salt, but the "proof is in the pudding" as they say - the issue is reproducibile.

Summary: _run_worker installs a signal.signal handler that raises _ExitCli (a BaseException subclass) directly from the handler. CPython delivers signal handlers on the main thread at an arbitrary bytecode boundary, so that exception can surface inside whatever coroutine the main thread is currently executing instead of at the loop.run_until_complete(main_task) boundary where it is caught.

When the main thread is inside AgentServer._answer_availability → _job_request_task (very likely whenever a user request_fnc / on_request performs blocking work), _ExitCli propagates out of that asyncio task. asyncio logs Task exception was never retrieved, the worker never enters the graceful drain/server.aclose() path, and shutdown is lost.

Because exit_raised is set to True before the raise, the handler latches: a second signal force-exits via os._exit(1) rather than retrying shutdown. In container environments that send a single SIGTERM and later SIGKILL, the first signal is wasted and the worker keeps accepting dispatches until killed.

worker.py wraps _job_request_task in except Exception (added in 1.6.x), but _ExitCli subclasses BaseException, not Exception, so it is not caught. Behavior is the same on 1.6.0 (original observation), 1.6.6 (re-verified), and 1.6.8 (reproduced against LiveKit Cloud in start mode).

The TCP/console path in the same module already uses the signal-safe pattern: asyncio.run_coroutine_threadsafe(_graceful_shutdown(), loop) instead of raising from the handler.

Observed symptom (framework frames only; reproduced on 1.6.8):

Task exception was never retrieved
future: <Task finished name='job_request' coro=<AgentServer._answer_availability.<locals>._job_request_task() done, defined at livekit/agents/utils/log.py:14> exception=_ExitCli()>
Traceback (most recent call last):
  File "livekit/agents/utils/log.py", line 17, in async_fn_logs
    return await fn(*args, **kwargs)
  File "livekit/agents/worker.py", line 1438, in _job_request_task
    await self._request_fnc(job_req)
  File "<user request_fnc>", line <N>, in request_fnc
    time.sleep(60)
  File "livekit/agents/cli/cli.py", line 304, in _handle_exit
    raise _ExitCli()
livekit.agents.cli.cli._ExitCli

After the first SIGTERM, the worker stayed alive and accepted a new job dispatch ~2s later (same room), confirming it never drained and remained available.

Impact: SIGTERM/SIGINT graceful shutdown is unreliable for workers whose request_fnc / on_request does synchronous/blocking I/O (HTTP, database, file locks, etc.). The process ignores orchestrator shutdown, continues to register as available, and may accept new job dispatches until SIGKILL.

Expected Behavior

Sending SIGTERM or SIGINT to a running worker should reliably trigger graceful shutdown (drain → server.aclose()), regardless of whether request_fnc is currently executing blocking work on the main thread.

Reproduction Steps

Use src/minimal_worker.py in this repo (or the snippet below). Requires LiveKit credentials in the environment (e.g. .env.local).

# minimal_worker.py — run with livekit-agents >= 1.6.0
import asyncio
import os
import time

from dotenv import load_dotenv
from livekit.agents import AgentServer, JobContext, JobRequest, cli

server = AgentServer()

load_dotenv(".env.local")


async def request_fnc(job: JobRequest) -> None:
    # The blocking sleep is the point of this repro: it keeps the main thread inside
    # AgentServer._answer_availability → _job_request_task, which is where the _ExitCli
    # raised by the SIGTERM handler in cli._run_worker lands instead of at the
    # run_until_complete boundary.
    print(f"REPRO: request_fnc entered, blocking for 60s (pid={os.getpid()})", flush=True)
    time.sleep(60)  # noqa: ASYNC251
    print("REPRO: request_fnc finished blocking", flush=True)
    await job.accept()


@server.rtc_session(on_request=request_fnc)
async def entrypoint(ctx: JobContext) -> None:
    await asyncio.sleep(0)


if __name__ == "__main__":
    cli.run_app(server)

Note: On 1.6.x, AgentServer no longer accepts WorkerOptions positionally.
AgentServer(WorkerOptions(...)) raises
TypeError: AgentServer.__init__() takes 1 positional argument but 2 were given.
Register the entrypoint and on_request via @server.rtc_session(...) as above
(or pass a legacy WorkerOptions directly to cli.run_app).

1. Start the worker in production mode:
     uv run python -m src.minimal_worker start
2. Dispatch any job so request_fnc runs (automatic dispatch works when agent_name
   is empty; e.g. `lk room join --identity repro-tester <room>`).
3. While request_fnc is inside time.sleep(), send SIGTERM: kill -TERM <worker-pid>
4. Observe: asyncio logs "Task exception was never retrieved" with _ExitCli from
   _job_request_task; the worker process stays alive, keeps heartbeating, and
   continues accepting dispatches until a second SIGTERM (os._exit latch) or SIGKILL.
5. Expected (but does not happen): worker enters drain and shuts down gracefully.

Optionally this bash script reproduces

#!/usr/bin/env bash
set -Eeuo pipefail

# Reproduction driver for the _ExitCli signal race:
# dispatch a job so request_fnc blocks the main thread, then SIGTERM the worker.

cd "$(dirname "$0")/.."
set -a
# shellcheck disable=SC1091
source .env.local
set +a

WORKER_PID="$1"
ROOM="repro-$(date +%s)"

echo "== joining room ${ROOM} to trigger automatic dispatch =="
lk room join --identity repro-tester "${ROOM}" > tmp/lk-join.log 2>&1 &
LK_PID=$!

for _ in $(seq 1 60); do
  if grep -q "REPRO: request_fnc entered" tmp/worker.log; then
    break
  fi
  sleep 1
done

if ! grep -q "REPRO: request_fnc entered" tmp/worker.log; then
  echo "!! request_fnc never ran; dispatch did not reach the worker"
  kill "${LK_PID}" 2>/dev/null || true
  exit 1
fi

echo "== request_fnc is blocking; sending SIGTERM to ${WORKER_PID} =="
date +%T
kill -TERM "${WORKER_PID}"

for i in $(seq 1 20); do
  sleep 1
  if ! kill -0 "${WORKER_PID}" 2>/dev/null; then
    echo "== worker exited ${i}s after SIGTERM =="
    kill "${LK_PID}" 2>/dev/null || true
    exit 0
  fi
done

echo "== worker STILL ALIVE 20s after SIGTERM =="
kill "${LK_PID}" 2>/dev/null || true
exit 0

Reproduction Steps

1. 
2.
3.
...
- Sample code snippet, or a GitHub Gist link -

Operating System

Linux (containerized worker, Python 3.13)

Models Used

N/A (no STT/LLM/TTS providers involved; worker dispatch lifecycle only)

Package Versions

livekit==1.1.13
livekit-agents==1.6.0   # originally observed
livekit-agents==1.6.6   # re-verified still present

Session/Room/Call IDs

N/A — reproduced with a synthetic request_fnc and no real sessions.

Proposed Solution

Align `_run_worker`'s signal handling with the TCP/console path in the same file (`_run_tcp_console`):


# livekit/agents/cli/cli.py — _run_worker (current, problematic)
def _handle_exit(sig: int, frame: FrameType | None) -> None:
    nonlocal exit_raised
    if exit_raised:
        os._exit(1)
    exit_raised = True
    raise _ExitCli()  # can land inside an arbitrary asyncio task



# Proposed: schedule shutdown on the event loop (signal-safe)
async def _graceful_shutdown() -> None:
    if not devmode:
        try:
            await server.drain()
        except asyncio.TimeoutError:
            logger.warning("drain timed out, forcing shutdown")
    await server.aclose()
    if watch_client:
        await watch_client.aclose()
    loop.stop()  # unblock run_until_complete

def _handle_exit(sig: int, frame: FrameType | None) -> None:
    nonlocal exit_raised
    if exit_raised:
        os._exit(1)
    exit_raised = True
    asyncio.run_coroutine_threadsafe(_graceful_shutdown(), loop)


Alternatively, on POSIX, register handlers with `loop.add_signal_handler(sig, ...)` so delivery stays on the event-loop thread.

Either approach avoids raising into asyncio task frames and makes SIGTERM shutdown reliable when `request_fnc` blocks.

Additional Context

• Raising exceptions from signal.signal handlers is a known CPython footgun: the handler runs on the main thread at the next bytecode boundary, which may be deep inside an unrelated call stack.
• Workers behind container orchestrators typically receive exactly one SIGTERM before SIGKILL; a latched one-shot handler that can miss its target is especially dangerous.
• Users cannot fully mitigate this in request_fnc alone — any blocking call on the event loop reopens the race window.

Screenshots and Recordings

No response

[chenghao-mou]

Thanks for the issue. I believe #5856 and PR #6624 is a fix for the same underlying issue. Going to ask the PR owner to add a test for this.

[biztex]

@bml1g12 your diagnosis matches what I found for #5856 — same root cause, one asyncio boundary over. #5856 is the raise landing in a loop callback (Handle._run); yours lands in a coroutine, where Task.__step applies the same rule: only SystemExit/KeyboardInterrupt are re-raised out of the loop, anything else is stored on the task and the loop keeps running.

#6624 makes _ExitCli derive from SystemExit, which covers both, and now carries a test for your case: a job-request task raises while the worker's main task is still awaiting, asserting the exit reaches run_until_complete. It fails with the current BaseException base.

Worth noting your except Exception observation is still true and still correct — SystemExit isn't caught by it either, which is what lets the exit escape the task rather than being handled as a job failure.