Function send_message_async

scripts/test_soft_interrupt.py:59–69

Send message in a thread.

(msg, session_id, result_queue)

        sock.close()

def send_message_async(msg, session_id, result_queue):
    """Send message in a thread."""
    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    try:
        sock.connect(SOCKET_PATH)
        ok, output = send_cmd_blocking(sock, f"message:{msg}", session_id, timeout=180)
        result_queue.put((ok, output))
    except Exception as e:
        result_queue.put((False, str(e)))
    finally:
        sock.close()

def create_test_session(cwd="/tmp"):
    """Create a headless test session."""