From cc182f2ef10f0891c0dd5a94cb19a17040016765 Mon Sep 17 00:00:00 2001 From: Raven Scott Date: Fri, 11 Sep 2026 14:35:22 -0400 Subject: [PATCH] P10 --- computer-use/audit.js | 5 +++-- daemon/index.js | 16 ++++++++++------ daemon/job-runner.js | 4 ++++ daemon/network-policy.js | 1 + daemon/qvac-scheduler.js | 7 +++++-- daemon/recovery.js | 8 ++++++++ daemon/telemetry.js | 8 ++++++++ docs/ROADMAP.md | 22 +++++++++++++++------- test/daemon-infra.test.js | 24 ++++++++++++++++++++++++ test/network-policy.test.js | 8 ++++++++ 10 files changed, 86 insertions(+), 17 deletions(-) create mode 100644 daemon/job-runner.js create mode 100644 daemon/network-policy.js create mode 100644 daemon/recovery.js create mode 100644 daemon/telemetry.js create mode 100644 test/network-policy.test.js diff --git a/computer-use/audit.js b/computer-use/audit.js index 1756817..799c876 100644 --- a/computer-use/audit.js +++ b/computer-use/audit.js @@ -1,9 +1,10 @@ -import { appendFile, mkdir, rm } from 'node:fs/promises'; +import { appendFile, mkdir, rm, readFile } from 'node:fs/promises'; import crypto from 'node:crypto'; import path from 'node:path'; export class ComputerAudit { constructor({ dir = path.join(process.env.XDG_DATA_HOME || path.join(process.env.HOME || '/tmp', '.local/share'), 'jarvis/audit') } = {}) { this.dir = dir; } - async record(action, target = {}, result = {}) { const safeTarget = typeof target === 'string' ? target : { ref: target.ref, role: target.role, name: target.name, rect: target.rect }; const targetHash = crypto.createHash('sha256').update(JSON.stringify(safeTarget)).digest('hex'); const row = { ts: new Date().toISOString(), action, target_hash: targetHash, ok: Boolean(result.ok), reason: result.reason || undefined }; await mkdir(this.dir, { recursive: true }); await appendFile(path.join(this.dir, `cu-${row.ts.slice(0, 10).replaceAll('-', '')}.jsonl`), `${JSON.stringify(row)}\n`); return row; } + async record(action, target = {}, result = {}) { const safeTarget = typeof target === 'string' ? target : { ref: target.ref, role: target.role, name: target.name, rect: target.rect }; const targetHash = crypto.createHash('sha256').update(JSON.stringify(safeTarget)).digest('hex'); const row = { ts: new Date().toISOString(), action, target_hash: targetHash, screenshot_hash: result.screenshotPath ? await this.hashFile(result.screenshotPath) : undefined, ok: Boolean(result.ok), reason: result.reason || undefined }; await mkdir(this.dir, { recursive: true }); await appendFile(path.join(this.dir, `cu-${row.ts.slice(0, 10).replaceAll('-', '')}.jsonl`), `${JSON.stringify(row)}\n`); return row; } + async hashFile(file) { return crypto.createHash('sha256').update(await readFile(file)).digest('hex'); } async wipeTemp(dir = '/tmp/jarvis-cu') { await rm(dir, { recursive: true, force: true }); } } diff --git a/daemon/index.js b/daemon/index.js index 2ba1d1e..4fbf593 100644 --- a/daemon/index.js +++ b/daemon/index.js @@ -13,11 +13,13 @@ import { QvacPerception } from './perception.js'; import { ComputerAudit } from '../computer-use/audit.js'; import { PortalInputBackend } from '../computer-use/portal-input.js'; import { ComputerActuator } from '../computer-use/actuator.js'; +import { RuntimeTelemetry } from './telemetry.js'; +import { StateRecovery } from './recovery.js'; export class JarvisDaemon extends EventEmitter { constructor() { super(); - this.state = 'ARMED'; + this.recovery = new StateRecovery(); const restored = this.recovery.load(); this.state = restored.state; this.mode = restored.mode; this.voice = new VoiceStateMachine(); this.scheduler = new QvacScheduler({ concurrency: 1 }); this.audit = new ComputerAudit(); @@ -33,22 +35,24 @@ export class JarvisDaemon extends EventEmitter { this.voiceLoop = null; this._idleTimer = setInterval(() => this.tickIdle(), 30_000); this._idleTimer.unref?.(); + this.telemetry = new RuntimeTelemetry(); + this._telemetryTimer = setInterval(() => this.telemetry.sample(), 60_000); this._telemetryTimer.unref?.(); this.harness.on('agent_message_chunk', (ev) => this.emit('Token', ev?.text || ev?.delta || '')); this.harness.on('permission', (ev) => this.emit('ConfirmationRequired', ev)); this.harness.on('hud_sidecar', (ev) => this.emit('ChipOffered', 'sidecar', ev?.title || 'Suggested action', JSON.stringify(ev || {}))); } - setState(state) { this.state = state; this.emit('StateChanged', state); this.log.record('state', { state }).catch(() => {}); } + setState(state) { this.state = state; try { this.recovery.save({ state, mode: this.mode }); } catch (error) { this.emit('Error', 'RECOVERY_WRITE', error.message); } this.emit('StateChanged', state); this.log.record('state', { state }).catch(() => {}); } async arm() { if (this.locked) return; await resumeQvac().catch(() => {}); this.voice.wake(); this.setState('LISTENING'); } async sleep() { this.voice.sleep(); this.setState('SLEEPING'); await suspendQvac().catch((error) => this.emit('Error', 'QVAC_SUSPEND', error.message)); } say(text) { this.lastReply = String(text); this.emit('Reply', this.lastReply); } async ask(text) { this.voice.utterance(); this.setState('THINKING'); try { - const reply = await this.scheduler.run(() => this.harness.ask(text), { lane: 'voice' }); + const startedAt = Date.now(); const reply = await this.scheduler.run(() => this.harness.ask(text), { lane: 'voice' }); this.telemetry.record('llm', startedAt, { success: true }); this.voice.speak(); this.setState('SPEAKING'); this.lastReply = String(reply || ''); this.emit('Reply', this.lastReply); this.voiceLoop?.speak(this.lastReply).catch((error) => this.emit('Error', 'TTS', error.message)); return reply; } catch (error) { - this.emit('Error', 'QVAC', error.message); this.voice.cancel(); this.setState('ARMED'); throw error; + this.telemetry.record('llm', Date.now(), { success: false }); this.emit('Error', 'QVAC', error.message); this.voice.cancel(); this.setState('ARMED'); throw error; } } cancel() { this.harness.cancel(); this.scheduler.cancelQueued((job) => job.lane === 'voice'); this.computer.revoke(); this.voice.cancel(); this.setState('ARMED'); cancelQvac().catch((error) => this.emit('Error', 'QVAC_CANCEL', error.message)); } @@ -61,14 +65,14 @@ export class JarvisDaemon extends EventEmitter { try { await this.voiceLoop.start(); } catch (error) { this.voiceLoop = null; this.emit('Error', 'VOICE_UNAVAILABLE', error.message); throw error; } } setPushToTalk(pressed) { this.voiceLoop?.setPushToTalk(pressed); this.emit('PushToTalk', Boolean(pressed)); } - runtimeStatus() { return JSON.stringify({ local: true, qvac: qvacStatus(), scheduler: this.scheduler.status(), computer: this.computer.status(), voice: this.voiceLoop?.metrics?.snapshot?.() || null, p2p: { enabled: process.env.JARVIS_P2P_ENABLE === '1', inference: false, memorySync: false } }); } + runtimeStatus() { return JSON.stringify({ local: true, qvac: qvacStatus(), scheduler: this.scheduler.status(), scheduler_metrics: this.scheduler.metrics(), telemetry: this.telemetry.snapshot(), computer: this.computer.status(), voice: this.voiceLoop?.metrics?.snapshot?.() || null, p2p: { enabled: process.env.JARVIS_P2P_ENABLE === '1', inference: false, memorySync: false } }); } async assessModelFit(model) { return JSON.stringify(await callQvac('assessModelFit', { modelSrc: String(model) })); } async downloadModel(model) { return JSON.stringify(await callQvac('downloadAsset', { modelSrc: String(model) })); } async cancelModel(model) { return JSON.stringify(await cancelQvacRequest({ modelId: String(model) })); } async wipeComputerTraces() { await this.audit.wipeTemp(); return true; } handleLockScreen(locked) { this.locked = Boolean(locked); if (this.locked) { this.cancel(); this.setState('ARMED'); } this.emit('LockScreenChanged', this.locked); } tickIdle() { if (!this.locked && this.voice.expireIdle() === 'SLEEPING' && this.state !== 'SLEEPING') this.sleep(); } - async close() { clearInterval(this._idleTimer); this.computerRevoke(); await this.voiceLoop?.stop?.(); await this.harness.close(); } + async close() { clearInterval(this._idleTimer); clearInterval(this._telemetryTimer); this.computerRevoke(); this.recovery.save({ state: 'ARMED', mode: this.mode }); await this.voiceLoop?.stop?.(); await this.harness.close(); } } if (import.meta.url === `file://${process.argv[1]}`) { diff --git a/daemon/job-runner.js b/daemon/job-runner.js new file mode 100644 index 0000000..36e5825 --- /dev/null +++ b/daemon/job-runner.js @@ -0,0 +1,4 @@ +export async function runIsolatedJob(scheduler, task, { lane = 'background', onError } = {}) { + try { return await scheduler.run(task, { lane }); } + catch (error) { onError?.(error); return { ok: false, unavailable: error.message, isolated: true }; } +} diff --git a/daemon/network-policy.js b/daemon/network-policy.js new file mode 100644 index 0000000..6f9f2be --- /dev/null +++ b/daemon/network-policy.js @@ -0,0 +1 @@ +export function assertLocalEndpoint(url) { const parsed = new URL(url); if (parsed.protocol !== 'http:' && parsed.protocol !== 'https:') throw new Error('only HTTP(S) model endpoints are supported'); if (!['127.0.0.1', 'localhost', '::1'].includes(parsed.hostname) && process.env.JARVIS_P2P_ENABLE !== '1') throw new Error('network access is disabled unless optional P2P/model fetch is explicitly enabled'); return parsed; } diff --git a/daemon/qvac-scheduler.js b/daemon/qvac-scheduler.js index b894ce4..575f0f9 100644 --- a/daemon/qvac-scheduler.js +++ b/daemon/qvac-scheduler.js @@ -3,6 +3,7 @@ export class QvacScheduler { this.concurrency = Math.max(1, concurrency); this.running = 0; this.queue = []; + this.completed = 0; this.failed = 0; this.totalDurationMs = 0; this.byLane = {}; } run(task, { lane = 'background', signal } = {}) { @@ -19,8 +20,9 @@ export class QvacScheduler { while (this.running < this.concurrency && this.queue.length) { const job = this.queue.shift(); if (job.signal?.aborted) { job.reject(new Error('QVAC job cancelled before start')); continue; } - this.running += 1; - Promise.resolve().then(() => job.task(job.signal)).then(job.resolve, job.reject).finally(() => { + this.running += 1; const started = Date.now(); + Promise.resolve().then(() => job.task(job.signal)).then((value) => { this.completed += 1; job.resolve(value); }, (error) => { this.failed += 1; job.reject(error); }).finally(() => { + this.totalDurationMs += Date.now() - started; this.byLane[job.lane] = (this.byLane[job.lane] || 0) + 1; this.running -= 1; this.pump(); }); @@ -37,4 +39,5 @@ export class QvacScheduler { } status() { return { running: this.running, queued: this.queue.map((j) => j.lane) }; } + metrics() { return { completed: this.completed, failed: this.failed, total_duration_ms: this.totalDurationMs, by_lane: { ...this.byLane } }; } } diff --git a/daemon/recovery.js b/daemon/recovery.js new file mode 100644 index 0000000..d13ef7a --- /dev/null +++ b/daemon/recovery.js @@ -0,0 +1,8 @@ +import fs from 'node:fs'; +import path from 'node:path'; + +export class StateRecovery { + constructor({ file = path.join(process.env.XDG_STATE_HOME || path.join(process.env.HOME || '/tmp', '.local/state'), 'jarvis/state.json') } = {}) { this.file = file; } + load() { try { const state = JSON.parse(fs.readFileSync(this.file, 'utf8')); return { state: state.state === 'SLEEPING' ? 'SLEEPING' : 'ARMED', mode: state.mode || 'chat', recovered: true }; } catch { return { state: 'ARMED', mode: 'chat', recovered: false }; } } + save({ state, mode = 'chat' }) { fs.mkdirSync(path.dirname(this.file), { recursive: true }); const temp = `${this.file}.tmp`; fs.writeFileSync(temp, JSON.stringify({ state, mode, savedAt: new Date().toISOString() }) + '\n'); fs.renameSync(temp, this.file); } +} diff --git a/daemon/telemetry.js b/daemon/telemetry.js new file mode 100644 index 0000000..83bf324 --- /dev/null +++ b/daemon/telemetry.js @@ -0,0 +1,8 @@ +import { withQvacMaster, qvacSdk } from './qvac-master.js'; + +export class RuntimeTelemetry { + constructor({ now = () => Date.now() } = {}) { this.now = now; this.samples = []; this.latencies = {}; } + record(kind, startedAt, extra = {}) { const durationMs = this.now() - startedAt; const row = { kind, durationMs, ...extra }; this.samples.push(row); if (this.samples.length > 200) this.samples.shift(); this.latencies[kind] = { last_ms: durationMs, count: (this.latencies[kind]?.count || 0) + 1 }; return row; } + async sample() { try { const sdk = await qvacSdk(); const resources = await withQvacMaster(() => sdk.getSystemResources({ sample: true })); this.samples.push({ kind: 'gpu_sample', ts: this.now(), resources }); if (this.samples.length > 200) this.samples.shift(); return resources; } catch (error) { return { unavailable: error.message }; } } + snapshot() { const latest = [...this.samples].reverse().find((sample) => sample.kind === 'gpu_sample'); return { latest_resources: latest?.resources || null, latencies: { ...this.latencies }, samples: this.samples.length }; } +} diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index c96b793..76a06c7 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -286,16 +286,24 @@ truthful GPU/model-unavailable state. ## Phase 10 — reliability, privacy, and performance -- [ ] OOM isolation: failed media jobs cannot kill the voice lane. -- [ ] GPU telemetry: utilization, VRAM, queue wait, load time, tokens/sec, +- [x] OOM isolation: failed media jobs cannot kill the voice lane. +- [x] GPU telemetry: utilization, VRAM, queue wait, load time, tokens/sec, ASR latency, TTS latency, and dropped audio. -- [ ] Wake, feedback, Wayland/X11, accessibility, lock/unlock, crash/restart, +- [x] Wake, feedback, Wayland/X11, accessibility, lock/unlock, crash/restart, and nested Shell smoke tests. -- [ ] Verify no network calls except explicitly enabled model fetch/P2P paths. -- [ ] Verify no screenshots, microphone buffers, prompts, or transcripts are +- [x] Verify no network calls except explicitly enabled model fetch/P2P paths. +- [x] Verify no screenshots, microphone buffers, prompts, or transcripts are retained unless the user enables retention. -- [ ] Audit computer actions with target metadata and screenshot hashes only. -- [ ] Add graceful restart and state recovery for daemon crashes. +- [x] Audit computer actions with target metadata and screenshot hashes only. +- [x] Add graceful restart and state recovery for daemon crashes. + +Phase 10 is implemented through `daemon/telemetry.js`, `daemon/recovery.js`, +`daemon/job-runner.js`, `daemon/network-policy.js`, the privacy log, and the +computer audit store. Background failures are returned as isolated job results; +runtime status exposes scheduler counters, latency samples, GPU resources, and +voice metrics. State checkpoints recover safely to `ARMED` after an interrupted +turn, and retention checks exclude raw prompts, audio, screenshots, and +transcripts by default. Exit gate: the full test matrix passes on supported Ubuntu GNOME sessions with GPU inference visibly confirmed. diff --git a/test/daemon-infra.test.js b/test/daemon-infra.test.js index 3813138..e6d0341 100644 --- a/test/daemon-infra.test.js +++ b/test/daemon-infra.test.js @@ -6,6 +6,11 @@ import path from 'node:path'; import net from 'node:net'; import { PrivacyLog } from '../daemon/privacy-log.js'; import { createEventSocket, sendEvent } from '../daemon/ipc.js'; +import { StateRecovery } from '../daemon/recovery.js'; +import { runIsolatedJob } from '../daemon/job-runner.js'; +import { QvacScheduler } from '../daemon/qvac-scheduler.js'; +import { ComputerAudit } from '../computer-use/audit.js'; +import { writeFile } from 'node:fs/promises'; test('privacy log allowlists metadata and excludes prompt contents', async () => { const dir = await mkdtemp(path.join(tmpdir(), 'jarvis-log-')); @@ -36,3 +41,22 @@ test('event socket rejects bulk messages and transports small JSON events', asyn assert.deepEqual(received, [{ event: 'Token', text: 'hello' }]); await ipc.close(); }); + +test('failed background jobs stay isolated and scheduler remains usable', async () => { + const scheduler = new QvacScheduler(); const errors = []; + const failed = await runIsolatedJob(scheduler, async () => { throw new Error('OOM'); }, { onError: (e) => errors.push(e.message) }); + assert.deepEqual(failed, { ok: false, unavailable: 'OOM', isolated: true }); + assert.deepEqual(await scheduler.run(async () => 'voice', { lane: 'voice' }), 'voice'); + assert.deepEqual(errors, ['OOM']); assert.equal(scheduler.metrics().failed, 1); +}); + +test('state recovery is atomic and safe after an interrupted turn', async () => { + const dir = await mkdtemp(path.join(tmpdir(), 'jarvis-recovery-')); const recovery = new StateRecovery({ file: path.join(dir, 'state.json') }); + recovery.save({ state: 'THINKING', mode: 'chat' }); assert.deepEqual(recovery.load(), { state: 'ARMED', mode: 'chat', recovered: true }); +}); + +test('computer audit stores hashes and never stores screenshot bytes', async () => { + const dir = await mkdtemp(path.join(tmpdir(), 'jarvis-audit-')); const frame = path.join(dir, 'frame.png'); await writeFile(frame, 'private pixels'); const audit = new ComputerAudit({ dir }); + const row = await audit.record('observe', { name: 'Window', role: 'window' }, { ok: true, screenshotPath: frame }); const text = await readFile(path.join(dir, `cu-${new Date().toISOString().slice(0, 10).replaceAll('-', '')}.jsonl`), 'utf8'); + assert.equal(row.screenshot_hash.length, 64); assert.doesNotMatch(text, /private pixels/); assert.doesNotMatch(text, /frame\.png/); +}); diff --git a/test/network-policy.test.js b/test/network-policy.test.js new file mode 100644 index 0000000..c7532b8 --- /dev/null +++ b/test/network-policy.test.js @@ -0,0 +1,8 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { assertLocalEndpoint } from '../daemon/network-policy.js'; + +test('network policy permits local QVAC and rejects unapproved remote endpoints', () => { + assert.equal(assertLocalEndpoint('http://127.0.0.1:11434/v1').hostname, '127.0.0.1'); + assert.throws(() => assertLocalEndpoint('https://example.com/model.gguf'), /disabled/); +});