fix(ipc): ThreadJobExecutor.aclose() hangs when the runner exits before its ack is read - #7225

Bug description

With JobExecutorType.THREAD, a graceful shutdown of a worker that has idle prewarmed runners logs job did not ack shutdown in time for most of them and then never exits. drain() has already returned by then, so drain_timeout doesn't help. Only a second signal ends the process. On a 16-core host (so 16 idle runners) I've been consistently reproducing this on every shutdown.

Why

ThreadJobExecutor._main_task() cancels _monitor_task as soon as the runner thread joins. An idle runner replies ShutdownRequestAck + ShuttingDown and exits right away, and its join callback (call_soon_threadsafe) is queued ahead of the socket read, so the monitor is cancelled with both messages still unread in the StreamReader. Nothing else resolves _shutting_down_fut, and aclose() awaits it with no timeout right after the ack timeout. SupervisedProc already handles this case (_on_read_ipc_done resolves both futures when its reader ends), but the thread executor never got the equivalent and has no kill path to fall back on. The result is the parent process hangs indefinetly instead of dying.

Fix

Simple change at job_thread_executor.py

• After the join, let _monitor_task finish on its own (bounded to 1 s) before cancelling it. The thread closes its end of the channel before joining, so the monitor reads the pending messages and ends on DuplexClosed.
• Resolve _shutdown_ack_fut and _shutting_down_fut in a finally of _monitor_task, same as SupervisedProc, so a cancelled or closed reader can never leave aclose() waiting.

Tests

tests/test_ipc.py had no ThreadJobExecutor coverage at all. Added test_thread_shutdown_no_job, test_thread_job_graceful_shutdown and test_thread_shutdown_no_job_join_before_ack_is_read, which reproduces the race deterministically by blocking the loop for 0.5 s right after aclose() sends the ShutdownRequest. All three fail on main with TimeoutError and pass with this change.

Minimal reproducible example

The minimal worker below is a proof of the bug: THREAD executor, 32 idle runners, no LiveKit server needed, SIGTERM once the runners are up. On main 27 of the 32 runners log did not ack and the process never exits. With this change it exits in about a second with no errors.

import logging
import sys
from livekit.agents import AgentServer, JobContext, JobExecutorType, cli
from livekit.agents.worker import ServerType

logging.basicConfig(level=logging.INFO)

async def entrypoint(ctx: JobContext) -> None:
    pass

if __name__ == "__main__":
    server = AgentServer(
        job_executor_type=JobExecutorType.THREAD,
        num_idle_processes=32,  # the more idle runners, the more likely one loses the race
        ws_url="ws://127.0.0.1:7880",  # nothing listens here, on purpose
        api_key="devkey",
        api_secret="secret",
        max_retry=sys.maxsize,  # keep retrying the connection instead of giving up
    )
    server.rtc_session(type=ServerType.ROOM, agent_name="repro")(entrypoint)
    cli.run_app(server)

uv run python repro.py start  # wait until "job runner initialized" was logged 32 times
kill -TERM <pid>              # or Ctrl+C