arf.ts

import * as http from 'http';
import * as fs from 'fs';
import * as os from 'os';
import * as path from 'path';
import { object, MAX_MESSAGE_BYTES } from './protocol';

export interface ArfSession {
    pid: number;
    socket_path: string;
    cwd?: string;
    r_version?: string;
    started_at?: string;
    session_type?: string;
}

/** arf speaks HTTP JSON-RPC, not sess's JSON Lines framing. */
export function arfRequest(endpoint: string, method: string, params: Record<string, unknown> = {},
    timeout = 30000, signal?: AbortSignal): Promise<unknown> {
    return new Promise((resolve, reject) => {
        const body = JSON.stringify({ jsonrpc: '2.0', id: 1, method, params });
        const request = http.request({ socketPath: endpoint, path: '/', method: 'POST', agent: false, signal,
            headers: { 'Content-Type': 'application/json', 'Content-Length': Buffer.byteLength(body), Connection: 'close' } }, response => {
            const chunks: Buffer[] = [];
            let size = 0;
            response.on('data', (chunk: Buffer) => {
                size += chunk.length;
                if (size > MAX_MESSAGE_BYTES) { request.destroy(new Error('arf response exceeds size limit')); }
                else { chunks.push(chunk); }
            });
            response.on('error', reject);
            response.on('end', () => {
                try {
                    const result = object(JSON.parse(Buffer.concat(chunks).toString('utf8')));
                    if (result.error) { reject(new Error(JSON.stringify(result.error))); }
                    else { resolve(result.result); }
                } catch (error) { reject(error); }
            });
        });
        if (timeout > 0) {
            request.setTimeout(timeout, () => request.destroy(new Error('arf request timed out; evaluation may still be running')));
        }
        request.on('error', reject);
        request.end(body);
    });
}

export function discoverArf(): ArfSession[] {
    const root = process.platform === 'darwin' ? path.join(os.homedir(), 'Library', 'Caches')
        : process.env.XDG_CACHE_HOME ?? path.join(os.homedir(), '.cache');
    const directory = path.join(root, 'arf', 'sessions');
    if (!fs.existsSync(directory)) { return []; }
    const sessions: ArfSession[] = [];
    for (const name of fs.readdirSync(directory)) {
        if (!/^\d+\.json$/.test(name)) { continue; }
        try {
            const file = path.join(directory, name);
            const stat = fs.lstatSync(file);
            if (stat.isSymbolicLink() || (process.getuid && stat.uid !== process.getuid())) { continue; }
            const value = object(JSON.parse(fs.readFileSync(file, 'utf8')));
            if (typeof value.pid !== 'number' || typeof value.socket_path !== 'string') { continue; }
            process.kill(value.pid, 0);
            sessions.push(value as unknown as ArfSession);
        } catch { /* A stale discovery file is not a running session. */ }
    }
    return sessions;
}

/** Reading frontend metadata does not evaluate R or request execution permission. */
export async function probeArfSession(session: ArfSession): Promise<ArfSession | undefined> {
    try {
        await arfRequest(session.socket_path, 'session', {}, 1000);
        return session;
    } catch { return undefined; }
}