Spaces:

fhueni
/

HERO

Running

HERO / src /requestManager.js

initial clean commit

92a645b 3 months ago

21.2 kB

import {measureAsync, sleep} from './utils.js';


/**
 * RequestManager routes inference requests to on-device or cloud services based on a routing strategy and configurations.
 * The manager does orchestrate the inference requests, collects statistics, evaluates the results and returns the final statistic.
 *
 * We provide different routing strategies:
 * - always_cloud: all requests go to cloud
 * - always_device: all requests go to device
 * - probabilistic: each request goes to cloud with a defined probability
 * - roundrobin: requests alternate between cloud and device
 * - hero: routes to the server with the shortest expected queue time
 */
export class RequestManager {
    constructor({
                    deviceService,
                    cloudService,
                    evaluator,
                    logger = null,
                    routeStrategy = 'roundrobin',
                    cloudProb = 0.5,
                    devicePerfModel = {slope: 0, intercept: 0},
                    cloudPerfModel = {slope: 0, intercept: 0}
                } = {}) {

        /**
         * On-device inference service
         */
        this.device = deviceService;

        /**
         * Cloud inference service
         */
        this.cloud = cloudService;

        /**
         * Evaluator instance for evaluating inference results
         */
        this.evaluator = evaluator;

        /**
         * Optional logger callback function
         * @type {null}
         */
        this.logger = logger;

        /**
         * Routing strategy (always_cloud, always_device, probabilistic, roundrobin)
         * @type {string}
         */
        this.routeStrategy = routeStrategy;

        /**
         * Probability of routing to cloud when using 'probabilistic' strategy
         * @type {number}
         */
        this.cloudProb = cloudProb;

        /**
         * Performance model for the device {slope, intercept}
         * @type {{slope: number, intercept: number}}
         */
        this.devicePerfModel = devicePerfModel;

        /**
         * Performance model for the cloud {slope, intercept}
         * @type {{slope: number, intercept: number}}
         */
        this.cloudPerfModel = cloudPerfModel;

        // Initialize Thompson samplers using offline fits as priors
        const deviceMu0 = [this.devicePerfModel.intercept || 0, this.devicePerfModel.slope || 0];
        const cloudMu0 = [this.cloudPerfModel.intercept || 0, this.cloudPerfModel.slope || 0];

        // Hyperparameters: priorLambda and sigma2 (tune as needed)
        const priorLambda = 1e-3;      // prior regularization (fast adaptation: 1e-6 … 1e-3, balanced: 1e-2 … 1.0, slow adaptation (anchor to offline fit): 10 … 1e3)
        const observationSigma2 = 1e4; // noise variance (ms^2) => sd ~100ms (low-noise (sd ≈ 20–50 ms): 400 … 2500, medium-noise (sd ≈ 50–100 ms): 2500 … 10000, high-noise (sd ≈ 100–200 ms): 10000 … 40000)

        this.tsDevice = new LinearThompsonSampler(deviceMu0, priorLambda, observationSigma2);
        this.tsCloud = new LinearThompsonSampler(cloudMu0, priorLambda, observationSigma2);

        /**
         * Internal round robin counter (even = cloud, odd = device)
         * @type {number}
         * @private
         */
        this._rrCounter = 0;

        /**
         * Statistics about routing and evaluations of this job run
         * @type {{cloud: number, evaluations: *[], count: number, device: number, totalLatencyMs: number}}
         */
        this.stats = {count: 0, cloud: 0, device: 0, totalLatencyMs: 0, results: []};

        /**
         * Cloud job queue
         * @type {*[]}
         */
        this.cloud_queue = [];

        /**
         * Device job queue
         * @type {*[]}
         */
        this.device_queue = [];

        // start processing jobs from the queues
        this.runOnDeviceJobsFromQueue();
        this.runCloudJobsFromQueue();
    }

    /**
     * Push a job to the appropriate queue based on routing strategy.
     *
     * @param job - The job to be processed
     */
    pushJob(job) {
        // get routing strategy and inference service
        const route = this._choose(job);
        console.log(`Device Queue Length: ${this.device_queue.length}, \nCloud Queue Length: ${this.cloud_queue.length}`);

        if (route === 'cloud') {
            this.cloud_queue.push(job);
        } else {
            this.device_queue.push(job);
        }
    }


    /**
     * Update routing configuration
     *
     * @param routeStrategy - New routing strategy
     * @param cloudProb - New cloud probability for 'probabilistic' strategy
     * @param devicePerfModel
     * @param cloudPerfModel
     */
    updateRouting({routeStrategy, cloudProb, devicePerfModel, cloudPerfModel}) {
        if (routeStrategy) this.routeStrategy = routeStrategy;
        if (cloudProb !== undefined) this.cloudProb = cloudProb;
        if (devicePerfModel) this.devicePerfModel = devicePerfModel;
        if (cloudPerfModel) this.cloudPerfModel = cloudPerfModel;
    }


    /**
     * Handle device jobs by routing it to the appropriate service, as long as there are jobs in the queue.
     *
     * @returns {Promise<void>}
     */
    async runOnDeviceJobsFromQueue() {
        while (true) {
            if (this.device_queue.length > 0) {
                const job = this._getNextJobFromQueue(this.device_queue, 'fifo');
                const service = this.device;
                const route = 'device';

                // run the job and await until compteted
                await this._runJob(job, route, service);
            }

            // sleep for 10ms to not run into memory leak
            await sleep(10);
        }
    }

    /**
     * Handle cloud jobs by routing it to the appropriate service, as long as there are jobs in the queue.
     *
     * @returns {Promise<void>}
     */
    async runCloudJobsFromQueue() {
        while (true) {
            if (this.cloud_queue.length > 0) {
                const job = this._getNextJobFromQueue(this.cloud_queue, 'fifo');
                const service = this.cloud;
                const route = 'cloud';

                // run the job and await until it completes
                await this._runJob(job, route, service);
            }

            // sleep for 10ms to not run into memory leak
            await sleep(10);
        }
    }


    /**
     * Run the given job on the specified service and record statistics.
     *
     * @param job - The job object containing prompt and ground truth
     * @param route - The selected route ('cloud' or 'device')
     * @param service - The inference service to use
     * @returns {Promise<void>}
     * @private
     */
    async _runJob(job, route, service) {
        let full_prompt = job.prompt; // ensure string input

        // this is a little workaround to disable the thinking mode in qwen models
        if (service.getModelName().toLowerCase().includes("qwen3".toLowerCase())) {
            full_prompt = full_prompt; // + "/no_think";
            console.log("ℹ️ \"/no_think\" was added to the prompt to avoid thinking")
        }

        let response, latencyMs, cleanedResponse; // response is object with .answer and .stats
        try {
            // Mark inference start
            job.timestamps.inferenceStart = Date.now();

            const {res, ms} = await measureAsync(() => service.infer(full_prompt));
            response = res;
            latencyMs = ms;

            // Mark inference end
            job.timestamps.inferenceEnd = Date.now();
        } catch (err) {
            response = `__error__:${err.message}`;
            latencyMs = -1;
            job.timestamps.inferenceEnd = Date.now();
        }

        // Calculate timing metrics
        const queueingTime = job.timestamps.inferenceStart - job.timestamps.jobStart;
        const inferenceTime = job.timestamps.inferenceEnd - job.timestamps.inferenceStart;
        const totalLatency = job.timestamps.inferenceEnd - job.timestamps.jobStart;


        // clean response
        cleanedResponse = this._cleanResponse(response);

        // evaluate result and store results
        const evalRes = this.evaluator.evaluate(cleanedResponse, job.groundTruth, latencyMs);
        this._record(route, latencyMs, evalRes, job, cleanedResponse, {queueingTime, inferenceTime, totalLatency});

        if (this.logger) {
            try {
                this.logger({job, route, latency: latencyMs, evalRes, response: cleanedResponse.answer, queueingTime, inferenceTime, totalLatency});
            } catch (error) {
                console.error("Logger encountered an error:", error);
            }
        }

        // update Thompson sampler with observed inference time (ms)
        try {
            if (latencyMs > 0) {
                const x = [1, job.prompt.length];
                const y = inferenceTime; // ms
                if (route === 'device') {
                    this.tsDevice.update(x, y);
                } else {
                    this.tsCloud.update(x, y);
                }
                // update public linear models (used by _decideHERO UI)
                this._updateLinearModelsForHERO();
            }
        } catch (err) {
            console.warn("TS update failed:", err);
        }

        // logging on console
        console.log(cleanedResponse)
        console.log("🎯 Models Answer: " + response.answer +
            "; \nCleaned Answer: " + cleanedResponse.answer +
            '; \nGround Truth: ' + job.groundTruth +
            "; \nInference Time: " + inferenceTime.toFixed(2) + "ms" +
            "; \nQueueing Time: " + queueingTime.toFixed(2) + "ms" +
            "; \nTotal Latency: " + totalLatency.toFixed(2) + "ms");
    }

    _getNextJobFromQueue(queue, policy) {
        // currently only FIFO is implemented
        return queue.shift();
    }

    /**
     * Choose a route based on the configured strategy.
     *
     * @returns {string} 'cloud' or 'device'
     * @private
     */
    _choose(job) {
        switch (this.routeStrategy) {
            case 'always_cloud':
                return 'cloud';
            case 'always_device':
                return 'device';
            case 'probabilistic':
                return Math.random() < this.cloudProb ? 'cloud' : 'device';
            case 'roundrobin':
                this._rrCounter++;
                return this._rrCounter % 2 === 0 ? 'cloud' : 'device';
            case 'hero':
                return this._decideHERO(job);
            default:
                return 'device';
        }
    }

    /**
     * Decide route based on our HERO policy.
     *
     * @param job
     * @returns {string}
     * @private
     */
    _decideHERO(job) {
        const now = Date.now();
        const input_size = job.prompt.length;

        // Thompson sample thetas (intercept, slope) in ms
        const thetaDevice = this.tsDevice.sampleTheta();
        const thetaCloud = this.tsCloud.sampleTheta();

        // predict inference time for the new job on both servers
        const device_predicted_inference_time = thetaDevice[0] + thetaDevice[1] * input_size;
        const cloud_predicted_inference_time = thetaCloud[0] + thetaCloud[1] * input_size;


        // Get the last job in the queue for both servers, to estimate when they will be free
        const lastDeviceJob = this.device_queue.length > 0 ? this.device_queue[this.device_queue.length - 1] : null;
        const lastCloudJob = this.cloud_queue.length > 0 ? this.cloud_queue[this.cloud_queue.length - 1] : null;

        // Calculate when each server will be free based on the last job in the queue
        const device_free_at = Math.max(now, lastDeviceJob?.hero_predictions?.device?.expectedFinishTime || 0);
        const cloud_free_at = Math.max(now, lastCloudJob?.hero_predictions?.cloud?.expectedFinishTime || 0);

        // Calculate expected finish time for the new job on both servers
        const device_expected_finish_time = device_free_at + device_predicted_inference_time;
        const cloud_expected_finish_time = cloud_free_at + cloud_predicted_inference_time;

        // calculate expected total time for both servers (predicted inference time + time until server is free)
        const device_expected_total_time = device_expected_finish_time - now;
        const cloud_expected_total_time = cloud_expected_finish_time - now;

        // store the predicted values in the job for logging and analysis
        job.hero_predictions = {
            device: {
                predictedInferenceTime: device_predicted_inference_time,
                predictedTotalTime: device_expected_total_time,
                expectedFinishTime: device_expected_finish_time,
                numberOfJobsInQueue: this.device_queue.length || 0
            },
            cloud: {
                predictedInferenceTime: cloud_predicted_inference_time,
                predictedTotalTime: cloud_expected_total_time,
                expectedFinishTime: cloud_expected_finish_time,
                numberOfJobsInQueue: this.cloud_queue.length || 0
            }
        };


        // Choose the server with the earlier expected finish time
        if (device_expected_finish_time <= cloud_expected_finish_time) {
            return 'device';
        } else {
            return 'cloud';
        }
    }

    /**
     * Update public linear model values from sampler posteriors (used by UI & logging)
     */
    _updateLinearModelsForHERO() {
        const devMean = this.tsDevice.posteriorMean();
        const cloudMean = this.tsCloud.posteriorMean();

        // posteriorMean returns [intercept, slope]
        this.devicePerfModel = { intercept: devMean[0], slope: devMean[1] };
        this.cloudPerfModel = { intercept: cloudMean[0], slope: cloudMean[1] };

        // dispatch update event for the frontend
        if (typeof window !== 'undefined') {
            window.dispatchEvent(new CustomEvent('perfModelsUpdated', {detail: { device: this.devicePerfModel, cloud: this.cloudPerfModel }}));
        }
    }

    /**
     * Record statistics for the given job evaluation.
     * Increases counters for total requests and cloud/device usage.
     * Updates the total latency.
     *
     * @param route - The route taken ('cloud' or 'device')
     * @param latency - Latency in milliseconds
     * @param evalRes - Evaluation result object
     * @param job - The job object
     * @param text - The inference result text
     * @param timingMetrics - Object containing queueingTime, inferenceTime, and totalLatency
     * @private
     */
    _record(route, latency, evalRes, job, text, timingMetrics) {
        this.stats.count++;
        if (route === 'cloud') this.stats.cloud++; else this.stats.device++;
        if (latency > 0) this.stats.totalLatencyMs += latency;
        this.stats.results.push({
            job: job,
            route,
            latency,
            evalRes,
            text,
            queueingTime: timingMetrics.queueingTime,
            inferenceTime: timingMetrics.inferenceTime,
            totalLatency: timingMetrics.totalLatency,
            timestamps: job.timestamps
        });
    }

    /**
     * Remove reasoning/thinking tokens and sections from a model's response.
     * Supports various formats (XML tags, special tokens, markdown, etc.) used by reasoning models.
     * Returns a cleaned response object with only the final answer for evaluation.
     *
     * @param response - The uncleaned response object (may include reasoning/thinking sections)
     * @return {object|string} - Cleaned response object or original error string
     * @private
     */
    _cleanResponse(response) {
        // If response is an error string, return as-is
        if (typeof response === 'string' && response.startsWith('__error__')) {
            return response;
        }

        // Clone the response object to avoid mutating the original
        const cleanedResponse = { ...response };
        
        if (!cleanedResponse.answer || typeof cleanedResponse.answer !== 'string') {
            return cleanedResponse;
        }

        let cleanedAnswer = cleanedResponse.answer;

        // Define patterns for thinking tokens (common formats)
        const thinkingPatterns = [
            // XML-style tags
            /<think>[\s\S]*?<\/think>/gi,
            /<thinking>[\s\S]*?<\/thinking>/gi,
            /<reasoning>[\s\S]*?<\/reasoning>/gi,
            /<thought>[\s\S]*?<\/thought>/gi,
            
            // Special tokens
            /<\|startofthinking\|>[\s\S]*?<\|endofthinking\|>/gi,
            /<\|reasoning_start\|>[\s\S]*?<\|reasoning_end\|>/gi,
            
            // Markdown-style
            /\[THINKING\][\s\S]*?\[\/THINKING\]/gi,
            /\[REASONING\][\s\S]*?\[\/REASONING\]/gi,
            /\[THOUGHT\][\s\S]*?\[\/THOUGHT\]/gi,
            
            // Other common patterns
            /\*\*Thinking:\*\*[\s\S]*?(?=\*\*Answer:\*\*|$)/gi,
            /\*\*Reasoning:\*\*[\s\S]*?(?=\*\*Answer:\*\*|$)/gi,
        ];

        // Apply all patterns to remove thinking sections
        for (const pattern of thinkingPatterns) {
            cleanedAnswer = cleanedAnswer.replace(pattern, '');
        }

        // Clean up extra whitespace
        cleanedAnswer = cleanedAnswer.trim();
        
        // If we removed everything, keep original (safety check)
        if (cleanedAnswer.length === 0 && cleanedResponse.answer.length > 0) {
            console.warn('⚠️ Thinking token removal resulted in empty answer. Keeping original.');
            return cleanedResponse;
        }

        cleanedResponse.answer = cleanedAnswer;
        return cleanedResponse;
    }
}




/**
 * 2-D Bayesian linear regressor for Thompson Sampling.
 * Models y = x^T theta + noise, x = [1, n]
 * Maintains A = Lambda + (1/sigma2) X^T X  (2x2), b = Lambda*mu0 + (1/sigma2) X^T y (2x1)
 * Posterior: theta | data ~ N(mu = A^{-1} b, cov = A^{-1})
 */
class LinearThompsonSampler {
    constructor(mu0 = [0, 0], priorLambda = 1e-3, sigma2 = 1e4) {
        // mu0: prior mean [intercept, slope]
        // priorLambda: scalar multiplied with identity (regularization)
        // sigma2: observation noise variance (ms^2)
        this.mu0 = [...mu0];
        this.priorLambda = priorLambda;
        this.sigma2 = sigma2;

        // A (2x2) initialized to priorLambda * I
        this.A = [
            [priorLambda, 0],
            [0, priorLambda]
        ];

        // b (2x1) initialized to priorLambda * mu0
        this.b = [
            priorLambda * this.mu0[0],
            priorLambda * this.mu0[1]
        ];
    }

    /**
     * Sample theta ~ N(mu, cov)
     * Uses Cholesky decomposition for sampling from multivariate normal.
     *
     * @returns {*[]}
     */
    sampleTheta() {
        const cov = invert2x2(this.A); // cov = A^{-1}
        const mu = matVecMul(cov, this.b);

        // Cholesky of cov (2x2) for sampling
        const L = cholesky2x2(cov);
        const z0 = randn(), z1 = randn();
        // theta = mu + L * z
        const theta0 = mu[0] + L[0][0] * z0;
        const theta1 = mu[1] + L[1][0] * z0 + L[1][1] * z1;
        return [theta0, theta1];
    }

    /**
     * Update with one observation (x: [1, n], y: observed_time_ms)
     *
     * @param x
     * @param y
     */
    update(x, y) {
        // A += (1/sigma2) * x x^T
        const factor = 1.0 / this.sigma2;
        this.A[0][0] += factor * x[0] * x[0];
        this.A[0][1] += factor * x[0] * x[1];
        this.A[1][0] += factor * x[1] * x[0];
        this.A[1][1] += factor * x[1] * x[1];

        // b += (1/sigma2) * x * y
        this.b[0] += factor * x[0] * y;
        this.b[1] += factor * x[1] * y;
    }

    /**
     * Return posterior mean [intercept, slope]
     *
     * @returns {*}
     */
    posteriorMean() {
        const cov = invert2x2(this.A);
        return matVecMul(cov, this.b);
    }
}

/* ===== Helper linear algebra (small, 2x2 implementations) ===== */

function matVecMul(mat, vec) {
    return [
        mat[0][0] * vec[0] + mat[0][1] * vec[1],
        mat[1][0] * vec[0] + mat[1][1] * vec[1]
    ];
}

function invert2x2(m) {
    // returns inverse of 2x2 matrix m
    const a = m[0][0], b = m[0][1], c = m[1][0], d = m[1][1];
    const det = a * d - b * c;
    const eps = 1e-12;
    const detSafe = Math.abs(det) < eps ? (det >= 0 ? eps : -eps) : det;
    const invDet = 1.0 / detSafe;
    return [
        [ d * invDet, -b * invDet ],
        [ -c * invDet, a * invDet ]
    ];
}

function cholesky2x2(m) {
    // m must be symmetric positive definite
    const a = m[0][0];
    const b = m[0][1]; // equals m[1][0]
    const c = m[1][1];
    const l00 = Math.sqrt(Math.max(a, 1e-12));
    const l10 = b / l00;
    const l11 = Math.sqrt(Math.max(c - l10 * l10, 1e-12));
    return [
        [l00, 0],
        [l10, l11]
    ];
}

function randn() {
    // Box-Muller
    let u = 0, v = 0;
    while (u === 0) u = Math.random();
    while (v === 0) v = Math.random();
    return Math.sqrt(-2.0 * Math.log(u)) * Math.cos(2.0 * Math.PI * v);
}

	import {measureAsync, sleep} from './utils.js';


	/**
	* RequestManager routes inference requests to on-device or cloud services based on a routing strategy and configurations.
	* The manager does orchestrate the inference requests, collects statistics, evaluates the results and returns the final statistic.
	*
	* We provide different routing strategies:
	* - always_cloud: all requests go to cloud
	* - always_device: all requests go to device
	* - probabilistic: each request goes to cloud with a defined probability
	* - roundrobin: requests alternate between cloud and device
	* - hero: routes to the server with the shortest expected queue time
	*/
	export class RequestManager {
	constructor({
	deviceService,
	cloudService,
	evaluator,
	logger = null,
	routeStrategy = 'roundrobin',
	cloudProb = 0.5,
	devicePerfModel = {slope: 0, intercept: 0},
	cloudPerfModel = {slope: 0, intercept: 0}
	} = {}) {

	/**
	* On-device inference service
	*/
	this.device = deviceService;

	/**
	* Cloud inference service
	*/
	this.cloud = cloudService;

	/**
	* Evaluator instance for evaluating inference results
	*/
	this.evaluator = evaluator;

	/**
	* Optional logger callback function
	* @type {null}
	*/
	this.logger = logger;

	/**
	* Routing strategy (always_cloud, always_device, probabilistic, roundrobin)
	* @type {string}
	*/
	this.routeStrategy = routeStrategy;

	/**
	* Probability of routing to cloud when using 'probabilistic' strategy
	* @type {number}
	*/
	this.cloudProb = cloudProb;

	/**
	* Performance model for the device {slope, intercept}
	* @type {{slope: number, intercept: number}}
	*/
	this.devicePerfModel = devicePerfModel;

	/**
	* Performance model for the cloud {slope, intercept}
	* @type {{slope: number, intercept: number}}
	*/
	this.cloudPerfModel = cloudPerfModel;

	// Initialize Thompson samplers using offline fits as priors
	const deviceMu0 = [this.devicePerfModel.intercept \|\| 0, this.devicePerfModel.slope \|\| 0];
	const cloudMu0 = [this.cloudPerfModel.intercept \|\| 0, this.cloudPerfModel.slope \|\| 0];

	// Hyperparameters: priorLambda and sigma2 (tune as needed)
	const priorLambda = 1e-3; // prior regularization (fast adaptation: 1e-6 … 1e-3, balanced: 1e-2 … 1.0, slow adaptation (anchor to offline fit): 10 … 1e3)
	const observationSigma2 = 1e4; // noise variance (ms^2) => sd ~100ms (low-noise (sd ≈ 20–50 ms): 400 … 2500, medium-noise (sd ≈ 50–100 ms): 2500 … 10000, high-noise (sd ≈ 100–200 ms): 10000 … 40000)

	this.tsDevice = new LinearThompsonSampler(deviceMu0, priorLambda, observationSigma2);
	this.tsCloud = new LinearThompsonSampler(cloudMu0, priorLambda, observationSigma2);

	/**
	* Internal round robin counter (even = cloud, odd = device)
	* @type {number}
	* @private
	*/
	this._rrCounter = 0;

	/**
	* Statistics about routing and evaluations of this job run
	* @type {{cloud: number, evaluations: *[], count: number, device: number, totalLatencyMs: number}}
	*/
	this.stats = {count: 0, cloud: 0, device: 0, totalLatencyMs: 0, results: []};

	/**
	* Cloud job queue
	* @type {*[]}
	*/
	this.cloud_queue = [];

	/**
	* Device job queue
	* @type {*[]}
	*/
	this.device_queue = [];

	// start processing jobs from the queues
	this.runOnDeviceJobsFromQueue();
	this.runCloudJobsFromQueue();
	}

	/**
	* Push a job to the appropriate queue based on routing strategy.
	*
	* @param job - The job to be processed
	*/
	pushJob(job) {
	// get routing strategy and inference service
	const route = this._choose(job);
	console.log(`Device Queue Length: ${this.device_queue.length}, \nCloud Queue Length: ${this.cloud_queue.length}`);

	if (route === 'cloud') {
	this.cloud_queue.push(job);
	} else {
	this.device_queue.push(job);
	}
	}


	/**
	* Update routing configuration
	*
	* @param routeStrategy - New routing strategy
	* @param cloudProb - New cloud probability for 'probabilistic' strategy
	* @param devicePerfModel
	* @param cloudPerfModel
	*/
	updateRouting({routeStrategy, cloudProb, devicePerfModel, cloudPerfModel}) {
	if (routeStrategy) this.routeStrategy = routeStrategy;
	if (cloudProb !== undefined) this.cloudProb = cloudProb;
	if (devicePerfModel) this.devicePerfModel = devicePerfModel;
	if (cloudPerfModel) this.cloudPerfModel = cloudPerfModel;
	}


	/**
	* Handle device jobs by routing it to the appropriate service, as long as there are jobs in the queue.
	*
	* @returns {Promise<void>}
	*/
	async runOnDeviceJobsFromQueue() {
	while (true) {
	if (this.device_queue.length > 0) {
	const job = this._getNextJobFromQueue(this.device_queue, 'fifo');
	const service = this.device;
	const route = 'device';

	// run the job and await until compteted
	await this._runJob(job, route, service);
	}

	// sleep for 10ms to not run into memory leak
	await sleep(10);
	}
	}

	/**
	* Handle cloud jobs by routing it to the appropriate service, as long as there are jobs in the queue.
	*
	* @returns {Promise<void>}
	*/
	async runCloudJobsFromQueue() {
	while (true) {
	if (this.cloud_queue.length > 0) {
	const job = this._getNextJobFromQueue(this.cloud_queue, 'fifo');
	const service = this.cloud;
	const route = 'cloud';

	// run the job and await until it completes
	await this._runJob(job, route, service);
	}

	// sleep for 10ms to not run into memory leak
	await sleep(10);
	}
	}


	/**
	* Run the given job on the specified service and record statistics.
	*
	* @param job - The job object containing prompt and ground truth
	* @param route - The selected route ('cloud' or 'device')
	* @param service - The inference service to use
	* @returns {Promise<void>}
	* @private
	*/
	async _runJob(job, route, service) {
	let full_prompt = job.prompt; // ensure string input

	// this is a little workaround to disable the thinking mode in qwen models
	if (service.getModelName().toLowerCase().includes("qwen3".toLowerCase())) {
	full_prompt = full_prompt; // + "/no_think";
	console.log("ℹ️ \"/no_think\" was added to the prompt to avoid thinking")
	}

	let response, latencyMs, cleanedResponse; // response is object with .answer and .stats
	try {
	// Mark inference start
	job.timestamps.inferenceStart = Date.now();

	const {res, ms} = await measureAsync(() => service.infer(full_prompt));
	response = res;
	latencyMs = ms;

	// Mark inference end
	job.timestamps.inferenceEnd = Date.now();
	} catch (err) {
	response = `__error__:${err.message}`;
	latencyMs = -1;
	job.timestamps.inferenceEnd = Date.now();
	}

	// Calculate timing metrics
	const queueingTime = job.timestamps.inferenceStart - job.timestamps.jobStart;
	const inferenceTime = job.timestamps.inferenceEnd - job.timestamps.inferenceStart;
	const totalLatency = job.timestamps.inferenceEnd - job.timestamps.jobStart;


	// clean response
	cleanedResponse = this._cleanResponse(response);

	// evaluate result and store results
	const evalRes = this.evaluator.evaluate(cleanedResponse, job.groundTruth, latencyMs);
	this._record(route, latencyMs, evalRes, job, cleanedResponse, {queueingTime, inferenceTime, totalLatency});

	if (this.logger) {
	try {
	this.logger({job, route, latency: latencyMs, evalRes, response: cleanedResponse.answer, queueingTime, inferenceTime, totalLatency});
	} catch (error) {
	console.error("Logger encountered an error:", error);
	}
	}

	// update Thompson sampler with observed inference time (ms)
	try {
	if (latencyMs > 0) {
	const x = [1, job.prompt.length];
	const y = inferenceTime; // ms
	if (route === 'device') {
	this.tsDevice.update(x, y);
	} else {
	this.tsCloud.update(x, y);
	}
	// update public linear models (used by _decideHERO UI)
	this._updateLinearModelsForHERO();
	}
	} catch (err) {
	console.warn("TS update failed:", err);
	}

	// logging on console
	console.log(cleanedResponse)
	console.log("🎯 Models Answer: " + response.answer +
	"; \nCleaned Answer: " + cleanedResponse.answer +
	'; \nGround Truth: ' + job.groundTruth +
	"; \nInference Time: " + inferenceTime.toFixed(2) + "ms" +
	"; \nQueueing Time: " + queueingTime.toFixed(2) + "ms" +
	"; \nTotal Latency: " + totalLatency.toFixed(2) + "ms");
	}

	_getNextJobFromQueue(queue, policy) {
	// currently only FIFO is implemented
	return queue.shift();
	}

	/**
	* Choose a route based on the configured strategy.
	*
	* @returns {string} 'cloud' or 'device'
	* @private
	*/
	_choose(job) {
	switch (this.routeStrategy) {
	case 'always_cloud':
	return 'cloud';
	case 'always_device':
	return 'device';
	case 'probabilistic':
	return Math.random() < this.cloudProb ? 'cloud' : 'device';
	case 'roundrobin':
	this._rrCounter++;
	return this._rrCounter % 2 === 0 ? 'cloud' : 'device';
	case 'hero':
	return this._decideHERO(job);
	default:
	return 'device';
	}
	}

	/**
	* Decide route based on our HERO policy.
	*
	* @param job
	* @returns {string}
	* @private
	*/
	_decideHERO(job) {
	const now = Date.now();
	const input_size = job.prompt.length;

	// Thompson sample thetas (intercept, slope) in ms
	const thetaDevice = this.tsDevice.sampleTheta();
	const thetaCloud = this.tsCloud.sampleTheta();

	// predict inference time for the new job on both servers
	const device_predicted_inference_time = thetaDevice[0] + thetaDevice[1] * input_size;
	const cloud_predicted_inference_time = thetaCloud[0] + thetaCloud[1] * input_size;


	// Get the last job in the queue for both servers, to estimate when they will be free
	const lastDeviceJob = this.device_queue.length > 0 ? this.device_queue[this.device_queue.length - 1] : null;
	const lastCloudJob = this.cloud_queue.length > 0 ? this.cloud_queue[this.cloud_queue.length - 1] : null;

	// Calculate when each server will be free based on the last job in the queue
	const device_free_at = Math.max(now, lastDeviceJob?.hero_predictions?.device?.expectedFinishTime \|\| 0);
	const cloud_free_at = Math.max(now, lastCloudJob?.hero_predictions?.cloud?.expectedFinishTime \|\| 0);

	// Calculate expected finish time for the new job on both servers
	const device_expected_finish_time = device_free_at + device_predicted_inference_time;
	const cloud_expected_finish_time = cloud_free_at + cloud_predicted_inference_time;

	// calculate expected total time for both servers (predicted inference time + time until server is free)
	const device_expected_total_time = device_expected_finish_time - now;
	const cloud_expected_total_time = cloud_expected_finish_time - now;

	// store the predicted values in the job for logging and analysis
	job.hero_predictions = {
	device: {
	predictedInferenceTime: device_predicted_inference_time,
	predictedTotalTime: device_expected_total_time,
	expectedFinishTime: device_expected_finish_time,
	numberOfJobsInQueue: this.device_queue.length \|\| 0
	},
	cloud: {
	predictedInferenceTime: cloud_predicted_inference_time,
	predictedTotalTime: cloud_expected_total_time,
	expectedFinishTime: cloud_expected_finish_time,
	numberOfJobsInQueue: this.cloud_queue.length \|\| 0
	}
	};


	// Choose the server with the earlier expected finish time
	if (device_expected_finish_time <= cloud_expected_finish_time) {
	return 'device';
	} else {
	return 'cloud';
	}
	}

	/**
	* Update public linear model values from sampler posteriors (used by UI & logging)
	*/
	_updateLinearModelsForHERO() {
	const devMean = this.tsDevice.posteriorMean();
	const cloudMean = this.tsCloud.posteriorMean();

	// posteriorMean returns [intercept, slope]
	this.devicePerfModel = { intercept: devMean[0], slope: devMean[1] };
	this.cloudPerfModel = { intercept: cloudMean[0], slope: cloudMean[1] };

	// dispatch update event for the frontend
	if (typeof window !== 'undefined') {
	window.dispatchEvent(new CustomEvent('perfModelsUpdated', {detail: { device: this.devicePerfModel, cloud: this.cloudPerfModel }}));
	}
	}

	/**
	* Record statistics for the given job evaluation.
	* Increases counters for total requests and cloud/device usage.
	* Updates the total latency.
	*
	* @param route - The route taken ('cloud' or 'device')
	* @param latency - Latency in milliseconds
	* @param evalRes - Evaluation result object
	* @param job - The job object
	* @param text - The inference result text
	* @param timingMetrics - Object containing queueingTime, inferenceTime, and totalLatency
	* @private
	*/
	_record(route, latency, evalRes, job, text, timingMetrics) {
	this.stats.count++;
	if (route === 'cloud') this.stats.cloud++; else this.stats.device++;
	if (latency > 0) this.stats.totalLatencyMs += latency;
	this.stats.results.push({
	job: job,
	route,
	latency,
	evalRes,
	text,
	queueingTime: timingMetrics.queueingTime,
	inferenceTime: timingMetrics.inferenceTime,
	totalLatency: timingMetrics.totalLatency,
	timestamps: job.timestamps
	});
	}

	/**
	* Remove reasoning/thinking tokens and sections from a model's response.
	* Supports various formats (XML tags, special tokens, markdown, etc.) used by reasoning models.
	* Returns a cleaned response object with only the final answer for evaluation.
	*
	* @param response - The uncleaned response object (may include reasoning/thinking sections)
	* @return {object\|string} - Cleaned response object or original error string
	* @private
	*/
	_cleanResponse(response) {
	// If response is an error string, return as-is
	if (typeof response === 'string' && response.startsWith('__error__')) {
	return response;
	}

	// Clone the response object to avoid mutating the original
	const cleanedResponse = { ...response };

	if (!cleanedResponse.answer \|\| typeof cleanedResponse.answer !== 'string') {
	return cleanedResponse;
	}

	let cleanedAnswer = cleanedResponse.answer;

	// Define patterns for thinking tokens (common formats)
	const thinkingPatterns = [
	// XML-style tags
	/<think>[\s\S]*?<\/think>/gi,
	/<thinking>[\s\S]*?<\/thinking>/gi,
	/<reasoning>[\s\S]*?<\/reasoning>/gi,
	/<thought>[\s\S]*?<\/thought>/gi,

	// Special tokens
	/<\\|startofthinking\\|>[\s\S]*?<\\|endofthinking\\|>/gi,
	/<\\|reasoning_start\\|>[\s\S]*?<\\|reasoning_end\\|>/gi,

	// Markdown-style
	/\[THINKING\][\s\S]*?\[\/THINKING\]/gi,
	/\[REASONING\][\s\S]*?\[\/REASONING\]/gi,
	/\[THOUGHT\][\s\S]*?\[\/THOUGHT\]/gi,

	// Other common patterns
	/\\Thinking:\\[\s\S]?(?=\\Answer:\\*\|$)/gi,
	/\\Reasoning:\\[\s\S]?(?=\\Answer:\\*\|$)/gi,
	];

	// Apply all patterns to remove thinking sections
	for (const pattern of thinkingPatterns) {
	cleanedAnswer = cleanedAnswer.replace(pattern, '');
	}

	// Clean up extra whitespace
	cleanedAnswer = cleanedAnswer.trim();

	// If we removed everything, keep original (safety check)
	if (cleanedAnswer.length === 0 && cleanedResponse.answer.length > 0) {
	console.warn('⚠️ Thinking token removal resulted in empty answer. Keeping original.');
	return cleanedResponse;
	}

	cleanedResponse.answer = cleanedAnswer;
	return cleanedResponse;
	}
	}




	/**
	* 2-D Bayesian linear regressor for Thompson Sampling.
	* Models y = x^T theta + noise, x = [1, n]
	* Maintains A = Lambda + (1/sigma2) X^T X (2x2), b = Lambda*mu0 + (1/sigma2) X^T y (2x1)
	* Posterior: theta \| data ~ N(mu = A^{-1} b, cov = A^{-1})
	*/
	class LinearThompsonSampler {
	constructor(mu0 = [0, 0], priorLambda = 1e-3, sigma2 = 1e4) {
	// mu0: prior mean [intercept, slope]
	// priorLambda: scalar multiplied with identity (regularization)
	// sigma2: observation noise variance (ms^2)
	this.mu0 = [...mu0];
	this.priorLambda = priorLambda;
	this.sigma2 = sigma2;

	// A (2x2) initialized to priorLambda * I
	this.A = [
	[priorLambda, 0],
	[0, priorLambda]
	];

	// b (2x1) initialized to priorLambda * mu0
	this.b = [
	priorLambda * this.mu0[0],
	priorLambda * this.mu0[1]
	];
	}

	/**
	* Sample theta ~ N(mu, cov)
	* Uses Cholesky decomposition for sampling from multivariate normal.
	*
	* @returns {*[]}
	*/
	sampleTheta() {
	const cov = invert2x2(this.A); // cov = A^{-1}
	const mu = matVecMul(cov, this.b);

	// Cholesky of cov (2x2) for sampling
	const L = cholesky2x2(cov);
	const z0 = randn(), z1 = randn();
	// theta = mu + L * z
	const theta0 = mu[0] + L[0][0] * z0;
	const theta1 = mu[1] + L[1][0] * z0 + L[1][1] * z1;
	return [theta0, theta1];
	}

	/**
	* Update with one observation (x: [1, n], y: observed_time_ms)
	*
	* @param x
	* @param y
	*/
	update(x, y) {
	// A += (1/sigma2) * x x^T
	const factor = 1.0 / this.sigma2;
	this.A[0][0] += factor * x[0] * x[0];
	this.A[0][1] += factor * x[0] * x[1];
	this.A[1][0] += factor * x[1] * x[0];
	this.A[1][1] += factor * x[1] * x[1];

	// b += (1/sigma2) * x * y
	this.b[0] += factor * x[0] * y;
	this.b[1] += factor * x[1] * y;
	}

	/**
	* Return posterior mean [intercept, slope]
	*
	* @returns {*}
	*/
	posteriorMean() {
	const cov = invert2x2(this.A);
	return matVecMul(cov, this.b);
	}
	}

	/* ===== Helper linear algebra (small, 2x2 implementations) ===== */

	function matVecMul(mat, vec) {
	return [
	mat[0][0] * vec[0] + mat[0][1] * vec[1],
	mat[1][0] * vec[0] + mat[1][1] * vec[1]
	];
	}

	function invert2x2(m) {
	// returns inverse of 2x2 matrix m
	const a = m[0][0], b = m[0][1], c = m[1][0], d = m[1][1];
	const det = a * d - b * c;
	const eps = 1e-12;
	const detSafe = Math.abs(det) < eps ? (det >= 0 ? eps : -eps) : det;
	const invDet = 1.0 / detSafe;
	return [
	[ d * invDet, -b * invDet ],
	[ -c * invDet, a * invDet ]
	];
	}

	function cholesky2x2(m) {
	// m must be symmetric positive definite
	const a = m[0][0];
	const b = m[0][1]; // equals m[1][0]
	const c = m[1][1];
	const l00 = Math.sqrt(Math.max(a, 1e-12));
	const l10 = b / l00;
	const l11 = Math.sqrt(Math.max(c - l10 * l10, 1e-12));
	return [
	[l00, 0],
	[l10, l11]
	];
	}

	function randn() {
	// Box-Muller
	let u = 0, v = 0;
	while (u === 0) u = Math.random();
	while (v === 0) v = Math.random();
	return Math.sqrt(-2.0 * Math.log(u)) * Math.cos(2.0 * Math.PI * v);
	}