212 lines
15 KiB
JavaScript
212 lines
15 KiB
JavaScript
// Row332 evidence. The only network-capable phase is the single authorized fetch.
|
|
// offline runs in unshare -rn; a transparent local tap records pi payload shapes
|
|
// without retaining headers, then pipes bytes unchanged to the pinned proxy.
|
|
import { mkdirSync, readFileSync, readlinkSync, writeFileSync } from 'node:fs';
|
|
import { dirname, join, resolve } from 'node:path';
|
|
import { fileURLToPath } from 'node:url';
|
|
import { getPriority, setPriority, networkInterfaces } from 'node:os';
|
|
import { spawn } from 'node:child_process';
|
|
import { once } from 'node:events';
|
|
import http from 'node:http';
|
|
import { fetchPinned } from '../../../../packages/model-proxy/src/fetch.mjs';
|
|
import { PIN } from '../../../../packages/model-proxy/src/common.mjs';
|
|
import { initialConfig, verifyInstall } from '../../../../packages/model-proxy/src/settings.mjs';
|
|
const root = process.env.P3_EVIDENCE_ROOT;
|
|
if (!root || !root.startsWith('/mnt/storage/scratch/') || resolve(root) !== root) throw Error('owned evidence root required');
|
|
const enteredNice = getPriority(0);
|
|
setPriority(0, 19);
|
|
const repo = resolve(dirname(fileURLToPath(import.meta.url)), '../../../..');
|
|
const mode = process.argv[2];
|
|
const fakeKeys = ['fx332-client', 'fx332-upstream', 'fx332-manage'];
|
|
function redact(value) {
|
|
let text = String(value);
|
|
for (const key of fakeKeys) text = text.split(key).join('[redacted]');
|
|
return text;
|
|
}
|
|
const emit = (value) => console.log(redact(JSON.stringify(value)));
|
|
if (mode === 'fetch') {
|
|
const dataRoot = join(root, 'data');
|
|
mkdirSync(dataRoot, { mode: 0o700 });
|
|
const installed = await fetchPinned({ dataRoot, pin: PIN });
|
|
writeFileSync(join(root, 'installed.json'), JSON.stringify(installed) + '\n', { mode: 0o600, flag: 'wx' });
|
|
emit({ phase: 'fetch', version: PIN.version, binarySha256: PIN.binary.sha256,
|
|
panelVersion: PIN.panel.version, panelSha256: PIN.panel.sha256, enteredNice, nice: getPriority(0), verified: true });
|
|
} else if (mode === 'offline' || mode === 'glm-tool') {
|
|
const withTool = mode === 'glm-tool';
|
|
if (readlinkSync('/proc/self/ns/net') === process.env.P3_HOST_NET_NS) throw Error('network namespace was not isolated');
|
|
if (Object.keys(networkInterfaces()).some(name => name !== 'lo')) throw Error('loopback-only namespace required');
|
|
// Re-hash the already fetched assets; this function performs no network I/O.
|
|
const installed = verifyInstall(join(root, 'data'), PIN);
|
|
const attemptRoot = join(root, withTool ? 'attempt3-tool' : 'attempt2');
|
|
mkdirSync(attemptRoot, { mode: 0o700 });
|
|
setPriority(0, 19);
|
|
emit({ phase: 'isolation', interfaces: Object.keys(networkInterfaces()), enteredNice, nice: getPriority(0), version: PIN.version,
|
|
binarySha256: PIN.binary.sha256, panelSha256: PIN.panel.sha256, reverified: true });
|
|
const requests = [], responses = [], upstreamRequests = [], children = new Set();
|
|
const managed = (args, options = {}) => {
|
|
setPriority(0, 19);
|
|
const child = spawn('/usr/bin/timeout', ['-k', '10', ...args], { env: {
|
|
PATH: '/usr/bin:/bin', HOME: join(root, 'home'), TMPDIR: process.env.TMPDIR,
|
|
LANG: 'C.UTF-8', ...options.env,
|
|
}, cwd: root, stdio: ['ignore', 'pipe', 'pipe'] });
|
|
children.add(child);
|
|
setPriority(child.pid, 19);
|
|
child.once('close', () => children.delete(child));
|
|
return child;
|
|
};
|
|
const summary = body => ({ keys: Object.keys(body).sort(), model: body.model, stream: body.stream,
|
|
roles: Array.isArray(body.messages) ? body.messages.map(message => message.role) : undefined,
|
|
inputTypes: Array.isArray(body.input) ? body.input.map(item => item.type) : typeof body.input,
|
|
...(Object.hasOwn(body, 'reasoning') ? { reasoning: body.reasoning } : {}),
|
|
...(Object.hasOwn(body, 'thinking') ? { thinking: body.thinking } : {}),
|
|
...(Object.hasOwn(body, 'reasoning_effort') ? { reasoning_effort: body.reasoning_effort } : {}),
|
|
...(Object.hasOwn(body, 'enable_thinking') ? { enable_thinking: body.enable_thinking } : {}),
|
|
...(Object.hasOwn(body, 'tool_stream') ? { tool_stream: body.tool_stream } : {}),
|
|
...(Array.isArray(body.tools) ? { toolNames: body.tools.map(tool => tool.function?.name) } : {}),
|
|
...(Object.hasOwn(body, 'stream_options') ? { stream_options: body.stream_options } : {}),
|
|
...(Object.hasOwn(body, 'max_tokens') ? { max_tokens: body.max_tokens } : {}),
|
|
...(Object.hasOwn(body, 'max_completion_tokens') ? { max_completion_tokens: body.max_completion_tokens } : {}) });
|
|
const payload = async req => { let text = ''; for await (const part of req) text += part; return JSON.parse(text || '{}'); };
|
|
const listen = async server => { await new Promise((ok, no) => { server.once('error', no); server.listen(0, '127.0.0.1', ok); }); return server.address().port; };
|
|
const marker = 'P332_OFFLINE_OK';
|
|
const upstream = http.createServer(async (req, res) => {
|
|
const body = await payload(req);
|
|
upstreamRequests.push({ method: req.method, path: req.url, ...summary(body) });
|
|
const item = { id: 'msg_fixture', type: 'message', role: 'assistant', status: 'completed', content: [{ type: 'output_text', text: marker, annotations: [] }] };
|
|
if (req.url.endsWith('/chat/completions')) {
|
|
res.writeHead(200, { 'Content-Type': 'text/event-stream' });
|
|
const chunk = (delta, finish_reason = null) => res.write(`data: ${JSON.stringify({ id: 'chat_fixture', object: 'chat.completion.chunk', created: 1, model: body.model, choices: [{ index: 0, delta, finish_reason }] })}\n\n`);
|
|
chunk({ role: 'assistant', content: marker }); chunk({}, 'stop');
|
|
res.write(`data: ${JSON.stringify({ id: 'chat_fixture', object: 'chat.completion.chunk', created: 1, model: body.model, choices: [], usage: { prompt_tokens: 1, completion_tokens: 1, total_tokens: 2 } })}\n\n`);
|
|
res.end('data: [DONE]\n\n');
|
|
} else if (req.url.endsWith('/responses')) {
|
|
res.writeHead(200, { 'Content-Type': 'text/event-stream' });
|
|
const event = (type, more) => res.write(`event: ${type}\ndata: ${JSON.stringify({ type, ...more })}\n\n`);
|
|
const response = { id: 'resp_fixture', object: 'response', model: body.model, status: 'in_progress', output: [] };
|
|
event('response.created', { response });
|
|
event('response.output_item.added', { output_index: 0, item: { ...item, status: 'in_progress', content: [] } });
|
|
event('response.content_part.added', { item_id: item.id, output_index: 0, content_index: 0, part: { type: 'output_text', text: '', annotations: [] } });
|
|
event('response.output_text.delta', { item_id: item.id, output_index: 0, content_index: 0, delta: marker });
|
|
event('response.output_text.done', { item_id: item.id, output_index: 0, content_index: 0, text: marker });
|
|
event('response.output_item.done', { output_index: 0, item });
|
|
event('response.completed', { response: { ...response, status: 'completed', output: [item], usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 } } });
|
|
res.end();
|
|
} else { res.writeHead(404); res.end('{"error":"stand-in path refused"}'); }
|
|
});
|
|
let proxy, tap;
|
|
let proxyText = '';
|
|
const corrections = {};
|
|
const restoreNice = pid => {
|
|
try {
|
|
const priority = getPriority(pid);
|
|
if (priority !== 19) { corrections[priority] = (corrections[priority] ?? 0) + 1; setPriority(pid, 19); }
|
|
const descendants = readFileSync(`/proc/${pid}/task/${pid}/children`, 'utf8').trim().split(/\s+/).filter(Boolean);
|
|
for (const child of descendants) restoreNice(Number(child));
|
|
} catch {}
|
|
};
|
|
const keepNice = setInterval(() => restoreNice(process.pid), 100);
|
|
keepNice.unref();
|
|
try {
|
|
const upstreamPort = await listen(upstream);
|
|
const reservation = http.createServer(); const proxyPort = await listen(reservation);
|
|
await new Promise(ok => reservation.close(ok));
|
|
if (proxyPort === 8317 || upstreamPort === 8317) throw Error('reserved host-proxy port excluded');
|
|
const dataRoot = join(attemptRoot, 'offline');
|
|
mkdirSync(dataRoot, { mode: 0o700 });
|
|
mkdirSync(join(dataRoot, 'proxy'), { mode: 0o700 });
|
|
mkdirSync(join(dataRoot, 'proxy', 'auth'), { mode: 0o700 });
|
|
const config = initialConfig({ dataRoot, port: proxyPort, key: fakeKeys[2], sentinel: fakeKeys[0] });
|
|
config.management['disable-control-panel'] = true;
|
|
config.observability.logs['logging-to-file'] = false;
|
|
// Bundled panel1.26.7: Ip reads api-keys.openai-compatibility;
|
|
// Fp/Np require group.keys; xv/hv serialize keys:[{"api-key":...}].
|
|
config['api-keys'] = { 'openai-compatibility': [{ name: 'p332-standin', 'base-url': `http://127.0.0.1:${upstreamPort}/v1`,
|
|
keys: [{ 'api-key': fakeKeys[1] }], models: [{ name: 'gpt-fixture' }, { name: 'glm-fixture' }] }] };
|
|
const configFile = join(dataRoot, 'config.yaml');
|
|
writeFileSync(configFile, JSON.stringify(config, null, 2) + '\n', { mode: 0o600, flag: 'wx' });
|
|
proxy = managed(['200', installed.binary, '-config', configFile, '-local-model'], { env: { MANAGEMENT_STATIC_PATH: dirname(installed.panel) } });
|
|
proxy.stdout.on('data', bytes => { proxyText += bytes; });
|
|
proxy.stderr.on('data', bytes => { proxyText += bytes; });
|
|
let catalog, ids = [];
|
|
const readinessDeadline = Date.now() + 30000;
|
|
for (let poll = 1; Date.now() < readinessDeadline; poll++) {
|
|
if (proxy.exitCode !== null) throw Error('proxy exited before readiness');
|
|
try {
|
|
const result = await fetch(`http://127.0.0.1:${proxyPort}/v1/models`, { headers: { Authorization: `Bearer ${fakeKeys[0]}` }, signal: AbortSignal.timeout(1000) });
|
|
catalog = result.ok ? await result.json() : null;
|
|
ids = (catalog?.data ?? []).map(model => model.id);
|
|
emit({ phase: 'readiness', poll, status: result.status, modelCount: ids.length });
|
|
if (['gpt-fixture', 'glm-fixture'].every(id => ids.includes(id))) break;
|
|
} catch { emit({ phase: 'readiness', poll, status: 'not-ready', modelCount: 0 }); }
|
|
await new Promise(ok => setTimeout(ok, 1000));
|
|
}
|
|
if (!catalog) throw Error('proxy readiness failed');
|
|
emit({ phase: 'catalog', modelIds: ids });
|
|
if (!['gpt-fixture', 'glm-fixture'].every(id => ids.includes(id))) throw Error('stand-in routes absent from proxy catalog');
|
|
tap = http.createServer(async (req, res) => {
|
|
let raw = ''; for await (const part of req) raw += part;
|
|
const body = JSON.parse(raw || '{}');
|
|
const index = requests.push({ method: req.method, path: req.url, ...summary(body) }) - 1;
|
|
const outgoing = http.request({ hostname: '127.0.0.1', port: proxyPort, method: req.method, path: req.url,
|
|
headers: { ...req.headers, host: `127.0.0.1:${proxyPort}` } }, incoming => {
|
|
const chunks = [];
|
|
res.writeHead(incoming.statusCode, incoming.headers);
|
|
incoming.on('data', bytes => { chunks.push(bytes); res.write(bytes); });
|
|
incoming.on('end', () => {
|
|
const text = Buffer.concat(chunks).toString();
|
|
const events = text.split('\n').filter(line => line.startsWith('event: ')).map(line => line.slice(7));
|
|
responses.push({ request: index, status: incoming.statusCode, events, marker: text.includes(marker),
|
|
...(incoming.statusCode >= 400 ? { errorBody: redact(text).slice(0, 800) } : {}) });
|
|
res.end();
|
|
});
|
|
});
|
|
outgoing.on('error', () => { res.writeHead(502); res.end(); });
|
|
// Preserve exact body bytes and never log a header value.
|
|
outgoing.end(raw);
|
|
});
|
|
const tapPort = await listen(tap);
|
|
if (tapPort === 8317) throw Error('reserved host-proxy port excluded');
|
|
const scenarios = [
|
|
['gpt-responses', 'openai-responses', 'gpt-fixture', null],
|
|
['gpt-completions', 'openai-completions', 'gpt-fixture', null],
|
|
['glm-completions', 'openai-completions', 'glm-fixture', { thinkingFormat: 'zai', zaiToolStream: true }],
|
|
];
|
|
for (const [name, api, model, compat] of (withTool ? scenarios.filter(([name]) => name === 'glm-completions') : scenarios)) {
|
|
const turnHome = join(attemptRoot, name, 'home');
|
|
const agentDir = join(turnHome, '.pi', 'agent');
|
|
mkdirSync(agentDir, { recursive: true, mode: 0o700 });
|
|
const models = { providers: { 'p332-probe': { baseUrl: `http://127.0.0.1:${tapPort}/v1`, api,
|
|
apiKey: '$ANTHROPIC_API_KEY', ...(compat ? { compat } : {}), models: [{ id: model, reasoning: true }] } } };
|
|
writeFileSync(join(agentDir, 'models.json'), JSON.stringify(models) + '\n', { mode: 0o600, flag: 'wx' });
|
|
writeFileSync(join(agentDir, 'settings.json'), JSON.stringify({ retry: { enabled: false, maxRetries: 0, provider: { maxRetries: 0 } } }) + '\n', { mode: 0o600, flag: 'wx' });
|
|
const reqStart = requests.length, upstreamStart = upstreamRequests.length, responseStart = responses.length;
|
|
const pi = managed(['50', '/usr/bin/node', join(repo, 'node_modules/@earendil-works/pi-coding-agent/dist/cli.js'),
|
|
'--no-extensions', '--no-skills', '--no-prompt-templates', '--no-themes', '--no-context-files', '--no-session',
|
|
...(withTool ? ['--tools', 'read'] : ['--no-tools']),
|
|
'--provider', 'p332-probe', '--model', model, '--thinking', 'low', '-p', 'Reply P332_OFFLINE_OK only.'],
|
|
{ env: { HOME: turnHome, PI_CODING_AGENT_DIR: agentDir, ANTHROPIC_API_KEY: fakeKeys[0] } });
|
|
let stdout = '', stderr = '';
|
|
pi.stdout.on('data', bytes => { stdout += bytes; });
|
|
pi.stderr.on('data', bytes => { stderr += bytes; });
|
|
const [exit, signal] = await once(pi, 'close');
|
|
emit({ phase: 'turn', name: withTool ? `${name}-tool` : name, api, exit, signal, marker: stdout.includes(marker),
|
|
piRequests: requests.slice(reqStart), proxyResponses: responses.slice(responseStart), upstreamRequests: upstreamRequests.slice(upstreamStart),
|
|
...(exit !== 0 ? { stderr: redact(stderr).slice(0, 1200), stdout: redact(stdout).slice(0, 1200) } : {}), nice: getPriority(0) });
|
|
}
|
|
emit({ phase: 'conclusion', limitation: 'Stand-in openai-compatibility credential only; no Codex OAuth credential or real-auth turn. This script never selects the production API.' });
|
|
emit({ phase: 'nice-corrections', restoredTo: 19, observedBeforeRestore: corrections });
|
|
} catch (error) {
|
|
emit({ phase: 'error', message: error.message, proxyDiagnostic: redact(proxyText).slice(-3000) });
|
|
process.exitCode = 1;
|
|
} finally {
|
|
clearInterval(keepNice);
|
|
for (const child of [...children]) child.kill('SIGTERM');
|
|
for (const child of [...children]) {
|
|
await Promise.race([once(child, 'close'), new Promise(ok => setTimeout(ok, 1200))]);
|
|
if (child.exitCode === null) child.kill('SIGKILL');
|
|
}
|
|
for (const server of [tap, upstream].filter(Boolean)) { server.closeAllConnections(); await new Promise(ok => server.close(ok)); }
|
|
emit({ phase: 'cleanup', remainingManagedChildren: children.size });
|
|
}
|
|
} else { throw Error('fetch, offline or glm-tool phase required'); }
|