openvibers/openvibe.run Commits
Add B3 control leases and session activity (#212)
Changes
nnew file mode 100644@@@ -0,0 +1,3 @@++-- phase: expand++-- B3 control lease: durable current controller state, validated and enforced by Run.++ALTER TABLE run_sessions ADD COLUMN IF NOT EXISTS control_lease jsonb;nnew file mode 100644@@@ -0,0 +1,15 @@++-- phase: expand++-- Owner-readable session command timeline. Content is deliberately limited to metadata, never command values.++CREATE TABLE IF NOT EXISTS run_session_actions (++ id bigserial PRIMARY KEY,++ session_id text NOT NULL REFERENCES run_sessions(id) ON DELETE CASCADE,++ command_id text NOT NULL,++ kind text NOT NULL,++ action text,++ actor_kind text NOT NULL,++ actor_subject text NOT NULL,++ result text NOT NULL,++ code text,++ created_at bigint NOT NULL++);++CREATE INDEX IF NOT EXISTS run_session_actions_newest ON run_session_actions (session_id, id DESC);nnew file mode 100644@@@ -0,0 +1,3 @@++-- phase: expand++-- Record the named start preset chosen for each browser desk.++ALTER TABLE run_sessions ADD COLUMN IF NOT EXISTS template text NOT NULL DEFAULT 'browser';nnew file mode 100644@@@ -0,0 +1,8 @@++-- phase: expand++-- People explicitly named as session controllers may use the session control endpoints.++CREATE TABLE IF NOT EXISTS run_session_control_guests (++ session_id text NOT NULL REFERENCES run_sessions(id) ON DELETE CASCADE,++ subject text NOT NULL,++ created_at bigint NOT NULL,++ PRIMARY KEY (session_id, subject)++);@@@ -15,7 +15,7 @@"express-rate-limit": "^7.5.0","helmet": "^8.1.0","iovalkey": "^0.4.0",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.338.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.346.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.100.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.155.0","pg": "^8.23.0"@@@ -1147,9 +1147,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.338.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.338.0",-- "integrity": "sha512-75FD6IxMsXoud3NFXKDs0/D0rhJtF2D+WL8M92ArIyabkUnM/sZvw6E6L97EqwQ3O7M/kB+uwsNSxGudbil+ng==",++ "version": "0.346.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.346.0",++ "integrity": "sha512-ncmROx6i7VQa6l1FitWZhPArFEYskrIbUWTdMHKty7c5m6OIYkjh8GRm9HxzfwH9vD89isTzQiX16yfU5NYHhw==","license": "AGPL-3.0-only","dependencies": {"ajv": "^8.17.1",@@@ -26,7 +26,7 @@"express-rate-limit": "^7.5.0","helmet": "^8.1.0","iovalkey": "^0.4.0",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.338.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.346.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.100.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.155.0","pg": "^8.23.0"@@@ -6,6 +6,7 @@** - their open browser sessions (run_sessions, server/sessions/), keyed `owner = user:usr_…`. The export carries each* session's settings and timing, never its ticket or stream secret; deletion removes them.++ * - the sessions where they were explicitly granted controls (run_session_control_guests); deletion removes those grants.* - the jobs they submitted (run_jobs.requester = `user:usr_…`). The export carries the job, its result and its* timings; the placement counts learned about their jobs (placement_outcomes) leave with them too; the row itself stays, because it is the project's record of the work and of what it cost, but the* requester becomes `user:deleted`, so the job no longer leads to a person.@@@ -19,6 +20,7 @@ const TABLES = [table: 'run_sessions', subject: 'owner', value: (usr) => `user:${usr}`, file: 'sessions.json', order: 'created_at',columns: ['id', 'via', 'kind', 'state', 'width', 'height', 'ttl_s', 'idle_s', 'machine_asked', 'requirements', 'machine', 'created_at', 'started_at', 'last_used_at', 'ended_at', 'end_reason', 'end_detail', 'metered_minutes', 'updated_at'],},++ { table: 'run_session_control_guests', subject: 'subject', file: 'shared-session-controls.json', order: 'created_at', columns: ['session_id', 'created_at'] },// Placement A1's learned counts: plain weights per option key and step kind, never text. Exported and erased with// the person.{ table: 'placement_outcomes', subject: 'person_id', file: 'placement-outcomes.json', order: 'updated_at' },@@@ -35,21 +35,21 @@ function sessionsRouter({ sessions }) {function who(req, res, cap) {const p = req.principal || { kind: 'anonymous' };if (p.kind === 'anonymous') { problem(res, req, 401, 'token.required', `${cap} needs your OpenVibe sign-in, or a first-party service acting for you`); return null; }-- if (p.kind === 'user') return { owner: `user:${p.subject}`, via: null };++ if (p.kind === 'user') return { owner: `user:${p.subject}`, via: null, actor: { kind: 'person', subject: p.subject } };const sub = String(p.sub || '');if (!sub.startsWith('svc:')) { problem(res, req, 403, 'capability.denied', 'a session is opened by its person, or by a first-party service acting for them'); return null; }const decision = capabilities.check(p.claims, cap);if (!decision.allowed) { problem(res, req, 403, decision.code || 'capability.denied', decision.reason || `${cap} not granted`); return null; }const subject = String(req.get('X-OV-Subject') || '').trim();if (!USER_SUBJECT.test(subject) || !ids.isSubjectId('user', subject)) { problem(res, req, 422, 'run.session.subject_required', 'a service opens a session for a person: X-OV-Subject usr_…'); return null; }-- return { owner: `user:${subject}`, via: `service:${sub.slice(4)}` };++ return { owner: `user:${subject}`, via: `service:${sub.slice(4)}`, actor: { kind: 'ai', subject: sub } };}-- async function mine(req, res, cap) {++ async function mine(req, res, cap, allowGuest = false) {const w = who(req, res, cap);if (!w) return null;const id = String(req.params.id || '');const row = SESSION_RE.test(id) ? await sessions.get(id) : null;-- if (!row || row.owner !== w.owner) { problem(res, req, 404, 'run.session.not_found', 'no such session of yours'); return null; }++ if (!row || (row.owner !== w.owner && !(allowGuest && w.actor.kind === 'person' && await sessions.isGuest(row, w.actor.subject)))) { problem(res, req, 404, 'run.session.not_found', 'no such session of yours'); return null; }return row;}/**@@@ -80,7 +80,7 @@ function sessionsRouter({ sessions }) {r.post('/sessions', wrap(async (req, res) => {const w = who(req, res, 'run.session.create');if (!w) return;-- const { row, created } = await sessions.create(w.owner, w.via, req.body && typeof req.body === 'object' ? req.body : {}, { pool: req.query.pool, strategy: req.query.strategy });++ const { row, created } = await sessions.create(w.owner, w.via, req.body && typeof req.body === 'object' ? req.body : {}, { pool: req.query.pool, strategy: req.query.strategy, template: req.query.template });res.status(created ? 201 : 200).json(sessions.present(row));}));r.get('/sessions', wrap(async (req, res) => {@@@ -96,9 +96,41 @@ function sessionsRouter({ sessions }) {if (row) res.json(req.query.explain === '1' ? { ...sessions.present(row), compute: row.compute_decision || null } : sessions.present(row));}));r.post('/sessions/:id/commands', wrap(async (req, res) => {++ const row = await mine(req, res, 'run.session.control', true);++ if (!row) return;++ const w = who(req, res, 'run.session.control');++ if (!w) return;++ res.json(await sessions.act(row, req.body && typeof req.body === 'object' ? req.body : {}, w.actor));++ }));++ r.post('/sessions/:id/control/request', wrap(async (req, res) => {++ const row = await mine(req, res, 'run.session.control', true);++ if (!row) return;++ const w = who(req, res, 'run.session.control');++ if (!w) return;++ res.json({ lease: await sessions.requestControl(row, w.actor) });++ }));++ r.get('/sessions/:id/control', wrap(async (req, res) => {++ const row = await mine(req, res, 'run.session.read', true);++ if (row) res.json({ lease: await sessions.controlLease(row) });++ }));++ r.get('/sessions/:id/timeline', wrap(async (req, res) => {++ const row = await mine(req, res, 'run.session.read');++ if (row) res.json(await sessions.timeline(row, req.query.limit, req.query.before));++ }));++ r.get('/sessions/:id/snapshot', wrap(async (req, res) => {++ const row = await mine(req, res, 'run.session.read');++ if (row) res.json(sessions.snapshot(row));++ }));++ r.post('/sessions/:id/control/grant', wrap(async (req, res) => {++ const row = await mine(req, res, 'run.session.control');++ if (!row) return;++ const target = req.body && req.body.holder;++ res.json({ lease: await sessions.grantControl(row, target) });++ }));++ r.post('/sessions/:id/control/take', wrap(async (req, res) => {const row = await mine(req, res, 'run.session.control');if (!row) return;-- res.json(await sessions.act(row, req.body && typeof req.body === 'object' ? req.body : {}));++ res.json({ lease: await sessions.takeControl(row) });}));r.post('/sessions/:id/end', wrap(async (req, res) => {const row = await mine(req, res, 'run.session.control');@@@ -24,7 +24,7 @@ const { createServiceOutbox } = require('openvibe-sdk/events');const TABLE = 'run_events_outbox'; // migrations/0001_jobs.sqlconst ACTOR = { type: 'service', id: 'run' };const EVENT_TYPES = ['run.job.queued', 'run.job.started', 'run.job.succeeded', 'run.job.failed', 'run.job.cancelled', 'run.job.expired',-- 'run.session.started', 'run.session.ended',++ 'run.session.started', 'run.session.ended', 'run.session.control_lease_changed', 'run.session.control_requested','agent.step.started', 'agent.step.updated', 'agent.step.ended']; // server/sessions/service.js (contracts 0.223.0)function createRunOutbox({ db, config, fetchImpl, now, log = console }) {@@@ -92,6 +92,9 @@ function createSessions({ config, db, desks, outbox, now = () => Date.now(), logif (compute.pool && (typeof compute.pool !== 'string' || !/^[a-z0-9-]{1,40}$/.test(compute.pool))) throw new SessionError(422, 'run.compute.pool', 'invalid compute pool');const pool = compute.pool || null;const strategy = compute.strategy || null;++ const template = compute.template || 'browser';++ if (!['browser', 'kept-browser', 'throwaway'].includes(template)) throw new SessionError(422, 'run.session.template', 'template must be browser, kept-browser or throwaway');++ if (template === 'kept-browser') throw new SessionError(501, 'run.session.template_unavailable', 'Kept browser profiles are not available on this machine.');if (strategy && !require('openvibe-sdk/compute').STRATEGIES.includes(strategy)) throw new SessionError(422, 'run.compute.strategy', 'unknown compute strategy');if (strategy && !pool) throw new SessionError(422, 'run.compute.pool', 'a strategy requires a compute pool');if (pool) { if (!broker) throw new SessionError(503, 'run.compute.off'); await broker.poolConfig(pool); }@@@ -107,11 +110,14 @@ function createSessions({ config, db, desks, outbox, now = () => Date.now(), log}if ((await store.openCount(db, owner)) >= maxOpen) throw new SessionError(429, 'run.session.limit', `at most ${maxOpen} sessions at once: end one first`);const at = now();++ const id = store.newSessionId();const row = {-- id: store.newSessionId(), owner, via: via || null, kind: 'browser', state: 'placing',++ id, owner, via: via || null, kind: 'browser', state: 'placing',width: body.width || 1280, height: body.height || 800, ttl_s: body.ttl_s || 3600, idle_s: body.idle_s || 900,-- compute_pool: pool, compute_strategy: strategy, compute_decision: null, machine_asked: body.machine || null, requirements: body.requirements || null, idempotency_key: body.idempotency_key || null,-- request_digest: digest, reasons: [], task_id: body.task_id || null, curtain: false, created_at: at, updated_at: at,++ compute_pool: pool, compute_strategy: strategy, compute_decision: null, template, machine_asked: body.machine || null, requirements: body.requirements || null, idempotency_key: body.idempotency_key || null,++ request_digest: digest, reasons: [], task_id: body.task_id || null, curtain: false,++ control_lease: { thing: `run.session:${id}`, holder: { kind: 'person', subject: bare(owner) }, since_ms: at, expires_ms: at + 60_000, watchers: [], seq: 1 },++ created_at: at, updated_at: at,};const out = await store.insert(db, row);if (out.created) setImmediate(() => { tick().catch((e) => log.warn(`[Run] session tick: ${e.message}`)); });@@@ -135,25 +141,130 @@ function createSessions({ config, db, desks, outbox, now = () => Date.now(), log});}++ const HOLDER_MS = 60_000;++ const ownerHolder = (row) => ({ kind: 'person', subject: bare(row.owner) });++ const sameHolder = (a, b) => Boolean(a && b && a.kind === b.kind && a.subject === b.subject);++ function actorHolder(actor, row) {++ if (!actor) return ownerHolder(row);++ if (actor.kind === 'person') return { kind: 'person', subject: String(actor.subject || bare(row.owner)) };++ const kind = actor.kind === 'ai' ? 'ai' : null;++ const subject = String(actor.subject || '').trim().slice(0, 200);++ if (!kind || !subject) throw new SessionError(403, 'run.session.control_holder', 'the caller is not a named controller');++ return { kind, subject };++ }++ async function saveLease(row, lease, reason) {++ contracts.assertValid('common.control-lease@1', lease);++ return db.tx(async (t) => {++ const fresh = await store.get(t, row.id);++ if (!fresh || fresh.state === 'ended') throw new SessionError(409, 'run.session.not_running', 'the session has ended');++ const saved = await store.update(t, row.id, { control_lease: lease, updated_at: now() });++ const prepared = outbox.events.prepare({++ event_type: 'run.session.control_lease_changed', version: 1, source: 'run', actor: ACTOR,++ timestamp: new Date(now()).toISOString(), subject: { type: 'session', id: row.id },++ visibility: 'internal', priority: 'low', payload: { session_id: row.id, reason, lease },++ }, { now: now() });++ contracts.assertValid('events.event-envelope@1', prepared);++ await t.maybe(`INSERT INTO ${EVENTS_TABLE} (event_id, envelope, created_at) VALUES ($1, $2, $3) ON CONFLICT (event_id) DO NOTHING RETURNING id`,++ [`run:${row.id}:control-lease:${lease.seq}`, JSON.stringify(prepared), now()]);++ return saved;++ });++ }++ async function controlLease(row) {++ let lease = row.control_lease || { thing: `run.session:${row.id}`, holder: ownerHolder(row), since_ms: Number(row.created_at), expires_ms: Number(row.created_at) + HOLDER_MS, watchers: [], seq: 1 };++ contracts.assertValid('common.control-lease@1', lease);++ if (lease.expires_ms != null && lease.expires_ms <= now()) {++ lease = { ...lease, holder: { kind: 'nobody', subject: null }, since_ms: now(), expires_ms: null, seq: lease.seq + 1 };++ row = await saveLease(row, lease, 'expired');++ }++ return row.control_lease || lease;++ }++ async function requestControl(row, actor) {++ const current = await controlLease(row);++ const target = actorHolder(actor, row);++ if (sameHolder(current.holder, target)) return current;++ if (current.holder.kind !== 'nobody') {++ const prepared = outbox.events.prepare({ event_type: 'run.session.control_requested', version: 1, source: 'run', actor: ACTOR,++ timestamp: new Date(now()).toISOString(), subject: { type: 'session', id: row.id }, visibility: 'internal', priority: 'low',++ payload: { session_id: row.id, requester: target } }, { now: now() });++ await db.maybe(`INSERT INTO ${EVENTS_TABLE} (event_id, envelope, created_at) VALUES ($1, $2, $3) ON CONFLICT (event_id) DO NOTHING RETURNING id`,++ [`run:${row.id}:control-request:${target.kind}:${target.subject}:${Math.floor(now() / 1000)}`, JSON.stringify(prepared), now()]);++ return current;++ }++ const next = { ...current, holder: target, since_ms: now(), expires_ms: now() + HOLDER_MS, seq: current.seq + 1 };++ const saved = await saveLease(row, next, 'requested');++ return saved.control_lease;++ }++ async function grantControl(row, target, reason = 'granted') {++ const current = await controlLease(row);++ if (!target || !['person', 'ai', 'nobody'].includes(target.kind)) throw new SessionError(422, 'run.session.control_target', 'target kind is person, ai or nobody');++ const subject = target.kind === 'nobody' ? null : String(target.subject || '').trim().slice(0, 200);++ if (target.kind !== 'nobody' && !subject) throw new SessionError(422, 'run.session.control_target', 'a person or AI controller needs a subject');++ if (target.kind === 'person' && !require('openvibe-contracts').ids.isSubjectId('user', subject)) throw new SessionError(422, 'run.session.control_target', 'a person controller needs a Network usr_ subject');++ const holder = { kind: target.kind, subject };++ if (sameHolder(holder, current.holder)) return current;++ const next = { ...current, holder, since_ms: now(), expires_ms: holder.kind === 'nobody' ? null : now() + HOLDER_MS, seq: current.seq + 1 };++ const saved = await saveLease(row, next, reason);++ if (holder.kind === 'person' && holder.subject !== bare(row.owner)) {++ await db.maybe('INSERT INTO run_session_control_guests (session_id, subject, created_at) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING RETURNING session_id', [row.id, holder.subject, now()]);++ }++ return saved.control_lease;++ }++ async function takeControl(row) { return grantControl(row, ownerHolder(row), 'owner_take_back'); }++const RESULTS = new Set(['ack', 'nack', 'refused', 'expired']);-- async function act(row, cmd) {++ async function act(row, cmd, actor = null) {const v = contracts.validate('run.session-control-request@1', cmd);if (!v.valid) throw new SessionError(422, 'run.session.invalid', `invalid command: ${(v.errors && v.errors[0] && v.errors[0].message) || 'bad value'}`.slice(0, 300));-- if (row.state !== 'running') throw new SessionError(409, 'run.session.not_running', row.state === 'ended' ? 'the session has ended' : 'the session\'s desk is not running yet');const id = cmd.id || `c-${now().toString(36)}`;++ if (row.state !== 'running') throw new SessionError(409, 'run.session.not_running', row.state === 'ended' ? 'the session has ended' : 'the session\'s desk is not running yet');++ const lease = await controlLease(row);++ const caller = actorHolder(actor, row);++ if (!sameHolder(lease.holder, caller)) {++ await logAction(row, id, cmd, caller, 'refused', 'run.session.control_held');++ throw new SessionError(403, 'run.session.control_held', 'only the current control holder may send commands');++ }const r = await desks.act(row.machine, row.desk, bare(row.owner), { id, kind: cmd.kind, value: cmd.value });if (r === null) {const ended = await finish(row, 'host_lost', 'its machine no longer holds the desk', { release: false });return { session_id: row.id, id, result: 'refused', code: 'run.session.host_lost', reason: 'the session\'s machine went away', state: ended ? ended.state : 'ended' };}-- await store.update(db, row.id, { last_used_at: now(), updated_at: now() });++ const renewed = { ...lease, expires_ms: now() + HOLDER_MS };++ if (renewed.expires_ms !== lease.expires_ms) await store.update(db, row.id, { control_lease: renewed, last_used_at: now(), updated_at: now() });++ else await store.update(db, row.id, { last_used_at: now(), updated_at: now() });const out = { session_id: row.id, id: String(r.id || id).slice(0, 64), result: RESULTS.has(r.result) ? r.result : 'expired', state: 'running' };if (r.code != null) out.code = String(r.code).slice(0, 80);if (r.reason != null) out.reason = String(r.reason).slice(0, 1000);if (r.data && typeof r.data === 'object' && out.result === 'ack') out.data = r.data;++ await logAction(row, out.id, cmd, caller, out.result, out.code || null);return out;}++ async function logAction(row, id, cmd, actor, result, code) {++ const kind = String(cmd.kind || 'unknown').slice(0, 40);++ const action = kind === 'browser' && cmd.value && typeof cmd.value.action === 'string' ? cmd.value.action.slice(0, 48) : null;++ await db.maybe(`INSERT INTO run_session_actions (session_id, command_id, kind, action, actor_kind, actor_subject, result, code, created_at)++ VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9) RETURNING id`, [row.id, String(id).slice(0, 64), kind, action, actor.kind,++ String(actor.subject || '').slice(0, 200), String(result).slice(0, 24), code ? String(code).slice(0, 80) : null, now()]);++ }++ async function timeline(row, limit = 50, before = null) {++ const n = Math.min(100, Math.max(1, Number(limit) || 50));++ const args = [row.id];++ const cursor = Number(before);++ const where = before && Number.isInteger(cursor) ? ` AND id < $${args.push(cursor)}` : '';++ args.push(n + 1);++ const rows = await db.many(`SELECT id, command_id, kind, action, actor_kind, actor_subject, result, code, created_at++ FROM run_session_actions WHERE session_id = $1${where} ORDER BY id DESC LIMIT $${args.length}`, args);++ const more = rows.length > n;++ const actions = rows.slice(0, n).map((a) => ({ id: Number(a.id), command_id: a.command_id, kind: a.kind, action: a.action,++ actor: { kind: a.actor_kind, subject: a.actor_subject }, result: a.result, code: a.code, at: new Date(Number(a.created_at)).toISOString() }));++ return { actions, next_cursor: more ? String(actions.at(-1).id) : null };++ }++ async function isGuest(row, subject) {++ if (!row || !subject || String(subject) === bare(row.owner)) return false;++ return Boolean(await db.maybe('SELECT 1 FROM run_session_control_guests WHERE session_id = $1 AND subject = $2', [row.id, String(subject)]));++ }++ function snapshot() { return { available: false, message: 'Snapshots are not available on this machine.' }; }++/*** The session's newest frame, as a small JPEG (plan 2026-11-07: the live still Surf's card shows and the Dock* links to). It reuses the desk's own capture — the `screen` command on the session's lease, the same one Actor's@@@ -274,6 +385,7 @@ function createSessions({ config, db, desks, outbox, now = () => Date.now(), logcontinue;}// running++ await controlLease(row);if (now() - row.created_at >= row.ttl_s * 1000) { await finish(row, 'ttl', `its ${row.ttl_s} s ran out`); continue; }if (now() - (row.last_used_at || row.started_at) >= row.idle_s * 1000) { await finish(row, 'idle', `no command for ${row.idle_s} s`); continue; }if (now() - (row.lease_checked_at || 0) >= leaseCheckMs) {@@@ -300,7 +412,7 @@ function createSessions({ config, db, desks, outbox, now = () => Date.now(), log}return {-- create, act, finish, still, setCurtain, tick, SessionError,++ create, act, finish, still, setCurtain, requestControl, grantControl, takeControl, controlLease, timeline, isGuest, snapshot, tick, SessionError,get: (id) => store.get(db, id),list: (owner, opts) => store.listOf(db, owner, opts),present: (row) => store.present(row, now()),@@@ -14,8 +14,8 @@ const { ids } = require('openvibe-contracts');const COLUMNS = ['id', 'owner', 'via', 'kind', 'state', 'width', 'height', 'ttl_s', 'idle_s', 'machine_asked', 'requirements', 'idempotency_key','compute_pool', 'compute_strategy', 'compute_decision', 'request_digest', 'machine', 'desk', 'trust', 'offer_id', 'reasons', 'created_at', 'started_at', 'last_used_at', 'ended_at', 'end_reason',-- 'end_detail', 'metered_minutes', 'lease_checked_at', 'task_id', 'curtain', 'updated_at'];--const JSON_COLUMNS = new Set(['requirements', 'reasons', 'compute_decision']);++ 'end_detail', 'metered_minutes', 'lease_checked_at', 'task_id', 'curtain', 'control_lease', 'template', 'updated_at'];++const JSON_COLUMNS = new Set(['requirements', 'reasons', 'compute_decision', 'control_lease']);const iso = (ms) => (ms == null ? null : new Date(Number(ms)).toISOString());const json = (v, dflt) => { if (v == null) return dflt; if (typeof v === 'object') return v; try { return JSON.parse(v); } catch { return dflt; } };@@@ -78,6 +78,17 @@ function fakeDesks() {assert.ok(started.some((p) => p.id === ses.id && p.state === 'running'), 'run.session.started');});++ await check('browser and throwaway presets are accepted; kept profiles and snapshots say plainly when unavailable', async () => {++ const unavailable = await t.call('POST', '/api/v1/sessions?template=kept-browser', { ...as(bea), body: { kind: 'browser' } });++ assert.strictEqual(unavailable.status, 501);++ assert.match(unavailable.json.detail, /not available on this machine/);++ const throwaway = await t.call('POST', '/api/v1/sessions?template=throwaway', { ...as(bea), body: { kind: 'browser' } });++ assert.strictEqual(throwaway.status, 201, throwaway.text);++ await sessions.finish(await sessions.get(throwaway.json.id), 'ended', 'test complete');++ const snapshot = await t.call('GET', `/api/v1/sessions/${ses.id}/snapshot`, as(ada));++ assert.deepStrictEqual(snapshot.json, { available: false, message: 'Snapshots are not available on this machine.' });++ });++await check('commands reach its desk; anyone else\'s call is 404', async () => {const r = await t.call('POST', `/api/v1/sessions/${ses.id}/commands`, { ...as(ada), body: { id: 'c1', kind: 'browser', value: { action: 'open', url: 'https://example.org/' } } });assert.strictEqual(r.status, 200, r.text);@@@ -94,6 +105,40 @@ function fakeDesks() {valid(list, 'run.session-list-result@1');assert.deepStrictEqual(list.sessions.map((s) => s.id), [ses.id]);assert.deepStrictEqual((await t.call('GET', '/api/v1/sessions', as(bea))).json.sessions, []);++ const timeline = await t.call('GET', `/api/v1/sessions/${ses.id}/timeline`, as(ada));++ assert.strictEqual(timeline.status, 200, timeline.text);++ assert.strictEqual(timeline.json.actions[0].command_id, 'c1');++ assert.strictEqual(timeline.json.actions[0].action, 'open');++ assert.ok(!JSON.stringify(timeline.json).includes('https://example.org'), 'timeline omits command values and URLs');++ assert.strictEqual((await t.call('GET', `/api/v1/sessions/${ses.id}/timeline`, as(bea))).status, 404);++ });++++ await check('the B3 control lease hands off to an agent, refuses the former holder, and owner take-back wins', async () => {++ const getLease = async () => (await t.call('GET', `/api/v1/sessions/${ses.id}/control`, as(ada))).json.lease;++ let lease = getLease ? await getLease() : null;++ valid(lease, 'common.control-lease@1');++ assert.deepStrictEqual(lease.holder, { kind: 'person', subject: ada });++ const agent = { token: t.network.signService({ sub: 'svc:actor', cap: ['run.session.control'] }), headers: { 'X-OV-Subject': ada } };++ assert.strictEqual((await t.call('POST', `/api/v1/sessions/${ses.id}/commands`, { ...agent, body: { kind: 'screen', value: {} } })).status, 403, 'an agent without the lease cannot control the desk');++ const grant = await t.call('POST', `/api/v1/sessions/${ses.id}/control/grant`, { ...as(ada), body: { holder: { kind: 'ai', subject: 'svc:actor' } } });++ valid(grant.json.lease, 'common.control-lease@1');++ const held = await t.call('POST', `/api/v1/sessions/${ses.id}/commands`, { ...agent, body: { kind: 'screen', value: {} } });++ assert.strictEqual(held.status, 200, held.text);++ assert.strictEqual((await t.call('POST', `/api/v1/sessions/${ses.id}/commands`, { ...as(ada), body: { kind: 'screen', value: {} } })).status, 403, 'owner must take back before sending input');++ lease = (await t.call('POST', `/api/v1/sessions/${ses.id}/control/take`, as(ada))).json.lease;++ assert.deepStrictEqual(lease.holder, { kind: 'person', subject: ada });++ assert.ok(lease.seq > grant.json.lease.seq);++ const guest = ids.newId('user');++ await t.call('POST', `/api/v1/sessions/${ses.id}/control/grant`, { ...as(ada), body: { holder: { kind: 'person', subject: guest } } });++ const guestCall = await t.call('POST', `/api/v1/sessions/${ses.id}/commands`, { ...as(guest), body: { kind: 'screen', value: {} } });++ assert.strictEqual(guestCall.status, 200, guestCall.text);++ assert.strictEqual((await t.call('POST', `/api/v1/sessions/${ses.id}/commands`, { ...as(ada), body: { kind: 'screen', value: {} } })).status, 403);++ lease = (await t.call('POST', `/api/v1/sessions/${ses.id}/control/take`, as(ada))).json.lease;++ t.clock.offset += 61_000;++ lease = (await t.call('GET', `/api/v1/sessions/${ses.id}/control`, as(ada))).json.lease;++ assert.deepStrictEqual(lease.holder, { kind: 'nobody', subject: null }, 'a quiet holder expires lazily');++ const events = (await t.outboxRows('run.session.control_lease_changed')).map((e) => e.payload || e);++ assert.ok(events.some((e) => e.lease && e.lease.seq === lease.seq), 'lease transitions are durable events');});await check('a first-party service opens one for a person, with the capability; an app never does', async () => {
new file mode 100644
new file mode 100644
@@ -0,0 +1,3 @@
@@ -0,0 +1,3 @@
+-- phase: expand
+-- B3 control lease: durable current controller state, validated and enforced by Run.
+ALTER TABLE run_sessions ADD COLUMN IF NOT EXISTS control_lease jsonb;
new file mode 100644
new file mode 100644
@@ -0,0 +1,15 @@
@@ -0,0 +1,15 @@
+-- phase: expand
+-- Owner-readable session command timeline. Content is deliberately limited to metadata, never command values.
+CREATE TABLE IF NOT EXISTS run_session_actions (
+ id bigserial PRIMARY KEY,
+ session_id text NOT NULL REFERENCES run_sessions(id) ON DELETE CASCADE,
+ command_id text NOT NULL,
+ kind text NOT NULL,
+ action text,
+ actor_kind text NOT NULL,
+ actor_subject text NOT NULL,
+ result text NOT NULL,
+ code text,
+ created_at bigint NOT NULL
+);
+CREATE INDEX IF NOT EXISTS run_session_actions_newest ON run_session_actions (session_id, id DESC);
new file mode 100644
new file mode 100644
@@ -0,0 +1,3 @@
@@ -0,0 +1,3 @@
+-- phase: expand
+-- Record the named start preset chosen for each browser desk.
+ALTER TABLE run_sessions ADD COLUMN IF NOT EXISTS template text NOT NULL DEFAULT 'browser';
new file mode 100644
new file mode 100644
@@ -0,0 +1,8 @@
@@ -0,0 +1,8 @@
+-- phase: expand
+-- People explicitly named as session controllers may use the session control endpoints.
+CREATE TABLE IF NOT EXISTS run_session_control_guests (
+ session_id text NOT NULL REFERENCES run_sessions(id) ON DELETE CASCADE,
+ subject text NOT NULL,
+ created_at bigint NOT NULL,
+ PRIMARY KEY (session_id, subject)
+);
@@ -15,7 +15,7 @@
@@ -15,7 +15,7 @@
"express-rate-limit": "^7.5.0",
"express-rate-limit": "^7.5.0",
"helmet": "^8.1.0",
"helmet": "^8.1.0",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.338.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.346.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.100.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.100.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.155.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.155.0",
"pg": "^8.23.0"
"pg": "^8.23.0"
@@ -1147,9 +1147,9 @@
@@ -1147,9 +1147,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.338.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.338.0",
- "integrity": "sha512-75FD6IxMsXoud3NFXKDs0/D0rhJtF2D+WL8M92ArIyabkUnM/sZvw6E6L97EqwQ3O7M/kB+uwsNSxGudbil+ng==",
+ "version": "0.346.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.346.0",
+ "integrity": "sha512-ncmROx6i7VQa6l1FitWZhPArFEYskrIbUWTdMHKty7c5m6OIYkjh8GRm9HxzfwH9vD89isTzQiX16yfU5NYHhw==",
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"dependencies": {
"dependencies": {
"ajv": "^8.17.1",
"ajv": "^8.17.1",
@@ -26,7 +26,7 @@
@@ -26,7 +26,7 @@
"express-rate-limit": "^7.5.0",
"express-rate-limit": "^7.5.0",
"helmet": "^8.1.0",
"helmet": "^8.1.0",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.338.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.346.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.100.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.100.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.155.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.155.0",
"pg": "^8.23.0"
"pg": "^8.23.0"
@@ -6,6 +6,7 @@
@@ -6,6 +6,7 @@
*
*
* - their open browser sessions (run_sessions, server/sessions/), keyed `owner = user:usr_…`. The export carries each
* - their open browser sessions (run_sessions, server/sessions/), keyed `owner = user:usr_…`. The export carries each
* session's settings and timing, never its ticket or stream secret; deletion removes them.
* session's settings and timing, never its ticket or stream secret; deletion removes them.
+ * - the sessions where they were explicitly granted controls (run_session_control_guests); deletion removes those grants.
* - the jobs they submitted (run_jobs.requester = `user:usr_…`). The export carries the job, its result and its
* - the jobs they submitted (run_jobs.requester = `user:usr_…`). The export carries the job, its result and its
* timings; the placement counts learned about their jobs (placement_outcomes) leave with them too; the row itself stays, because it is the project's record of the work and of what it cost, but the
* timings; the placement counts learned about their jobs (placement_outcomes) leave with them too; the row itself stays, because it is the project's record of the work and of what it cost, but the
* requester becomes `user:deleted`, so the job no longer leads to a person.
* requester becomes `user:deleted`, so the job no longer leads to a person.
@@ -19,6 +20,7 @@ const TABLES = [
@@ -19,6 +20,7 @@ const TABLES = [
table: 'run_sessions', subject: 'owner', value: (usr) => `user:${usr}`, file: 'sessions.json', order: 'created_at',
table: 'run_sessions', subject: 'owner', value: (usr) => `user:${usr}`, file: 'sessions.json', order: 'created_at',
columns: ['id', 'via', 'kind', 'state', 'width', 'height', 'ttl_s', 'idle_s', 'machine_asked', 'requirements', 'machine', 'created_at', 'started_at', 'last_used_at', 'ended_at', 'end_reason', 'end_detail', 'metered_minutes', 'updated_at'],
columns: ['id', 'via', 'kind', 'state', 'width', 'height', 'ttl_s', 'idle_s', 'machine_asked', 'requirements', 'machine', 'created_at', 'started_at', 'last_used_at', 'ended_at', 'end_reason', 'end_detail', 'metered_minutes', 'updated_at'],
},
},
+ { table: 'run_session_control_guests', subject: 'subject', file: 'shared-session-controls.json', order: 'created_at', columns: ['session_id', 'created_at'] },
// Placement A1's learned counts: plain weights per option key and step kind, never text. Exported and erased with
// Placement A1's learned counts: plain weights per option key and step kind, never text. Exported and erased with
// the person.
// the person.
{ table: 'placement_outcomes', subject: 'person_id', file: 'placement-outcomes.json', order: 'updated_at' },
{ table: 'placement_outcomes', subject: 'person_id', file: 'placement-outcomes.json', order: 'updated_at' },
@@ -35,21 +35,21 @@ function sessionsRouter({ sessions }) {
@@ -35,21 +35,21 @@ function sessionsRouter({ sessions }) {
function who(req, res, cap) {
function who(req, res, cap) {
const p = req.principal || { kind: 'anonymous' };
const p = req.principal || { kind: 'anonymous' };
if (p.kind === 'anonymous') { problem(res, req, 401, 'token.required', `${cap} needs your OpenVibe sign-in, or a first-party service acting for you`); return null; }
if (p.kind === 'anonymous') { problem(res, req, 401, 'token.required', `${cap} needs your OpenVibe sign-in, or a first-party service acting for you`); return null; }
- if (p.kind === 'user') return { owner: `user:${p.subject}`, via: null };
+ if (p.kind === 'user') return { owner: `user:${p.subject}`, via: null, actor: { kind: 'person', subject: p.subject } };
const sub = String(p.sub || '');
const sub = String(p.sub || '');
if (!sub.startsWith('svc:')) { problem(res, req, 403, 'capability.denied', 'a session is opened by its person, or by a first-party service acting for them'); return null; }
if (!sub.startsWith('svc:')) { problem(res, req, 403, 'capability.denied', 'a session is opened by its person, or by a first-party service acting for them'); return null; }
const decision = capabilities.check(p.claims, cap);
const decision = capabilities.check(p.claims, cap);
if (!decision.allowed) { problem(res, req, 403, decision.code || 'capability.denied', decision.reason || `${cap} not granted`); return null; }
if (!decision.allowed) { problem(res, req, 403, decision.code || 'capability.denied', decision.reason || `${cap} not granted`); return null; }
const subject = String(req.get('X-OV-Subject') || '').trim();
const subject = String(req.get('X-OV-Subject') || '').trim();
if (!USER_SUBJECT.test(subject) || !ids.isSubjectId('user', subject)) { problem(res, req, 422, 'run.session.subject_required', 'a service opens a session for a person: X-OV-Subject usr_…'); return null; }
if (!USER_SUBJECT.test(subject) || !ids.isSubjectId('user', subject)) { problem(res, req, 422, 'run.session.subject_required', 'a service opens a session for a person: X-OV-Subject usr_…'); return null; }
- return { owner: `user:${subject}`, via: `service:${sub.slice(4)}` };
+ return { owner: `user:${subject}`, via: `service:${sub.slice(4)}`, actor: { kind: 'ai', subject: sub } };
}
}
- async function mine(req, res, cap) {
+ async function mine(req, res, cap, allowGuest = false) {
const w = who(req, res, cap);
const w = who(req, res, cap);
if (!w) return null;
if (!w) return null;
const id = String(req.params.id || '');
const id = String(req.params.id || '');
const row = SESSION_RE.test(id) ? await sessions.get(id) : null;
const row = SESSION_RE.test(id) ? await sessions.get(id) : null;
- if (!row || row.owner !== w.owner) { problem(res, req, 404, 'run.session.not_found', 'no such session of yours'); return null; }
+ if (!row || (row.owner !== w.owner && !(allowGuest && w.actor.kind === 'person' && await sessions.isGuest(row, w.actor.subject)))) { problem(res, req, 404, 'run.session.not_found', 'no such session of yours'); return null; }
return row;
return row;
}
}
/**
/**
@@ -80,7 +80,7 @@ function sessionsRouter({ sessions }) {
@@ -80,7 +80,7 @@ function sessionsRouter({ sessions }) {
r.post('/sessions', wrap(async (req, res) => {
r.post('/sessions', wrap(async (req, res) => {
const w = who(req, res, 'run.session.create');
const w = who(req, res, 'run.session.create');
if (!w) return;
if (!w) return;
- const { row, created } = await sessions.create(w.owner, w.via, req.body && typeof req.body === 'object' ? req.body : {}, { pool: req.query.pool, strategy: req.query.strategy });
+ const { row, created } = await sessions.create(w.owner, w.via, req.body && typeof req.body === 'object' ? req.body : {}, { pool: req.query.pool, strategy: req.query.strategy, template: req.query.template });
res.status(created ? 201 : 200).json(sessions.present(row));
res.status(created ? 201 : 200).json(sessions.present(row));
}));
}));
r.get('/sessions', wrap(async (req, res) => {
r.get('/sessions', wrap(async (req, res) => {
@@ -96,9 +96,41 @@ function sessionsRouter({ sessions }) {
@@ -96,9 +96,41 @@ function sessionsRouter({ sessions }) {
if (row) res.json(req.query.explain === '1' ? { ...sessions.present(row), compute: row.compute_decision || null } : sessions.present(row));
if (row) res.json(req.query.explain === '1' ? { ...sessions.present(row), compute: row.compute_decision || null } : sessions.present(row));
}));
}));
r.post('/sessions/:id/commands', wrap(async (req, res) => {
r.post('/sessions/:id/commands', wrap(async (req, res) => {
+ const row = await mine(req, res, 'run.session.control', true);
+ if (!row) return;
+ const w = who(req, res, 'run.session.control');
+ if (!w) return;
+ res.json(await sessions.act(row, req.body && typeof req.body === 'object' ? req.body : {}, w.actor));
+ }));
+ r.post('/sessions/:id/control/request', wrap(async (req, res) => {
+ const row = await mine(req, res, 'run.session.control', true);
+ if (!row) return;
+ const w = who(req, res, 'run.session.control');
+ if (!w) return;
+ res.json({ lease: await sessions.requestControl(row, w.actor) });
+ }));
+ r.get('/sessions/:id/control', wrap(async (req, res) => {
+ const row = await mine(req, res, 'run.session.read', true);
+ if (row) res.json({ lease: await sessions.controlLease(row) });
+ }));
+ r.get('/sessions/:id/timeline', wrap(async (req, res) => {
+ const row = await mine(req, res, 'run.session.read');
+ if (row) res.json(await sessions.timeline(row, req.query.limit, req.query.before));
+ }));
+ r.get('/sessions/:id/snapshot', wrap(async (req, res) => {
+ const row = await mine(req, res, 'run.session.read');
+ if (row) res.json(sessions.snapshot(row));
+ }));
+ r.post('/sessions/:id/control/grant', wrap(async (req, res) => {
+ const row = await mine(req, res, 'run.session.control');
+ if (!row) return;
+ const target = req.body && req.body.holder;
+ res.json({ lease: await sessions.grantControl(row, target) });
+ }));
+ r.post('/sessions/:id/control/take', wrap(async (req, res) => {
const row = await mine(req, res, 'run.session.control');
const row = await mine(req, res, 'run.session.control');
if (!row) return;
if (!row) return;
- res.json(await sessions.act(row, req.body && typeof req.body === 'object' ? req.body : {}));
+ res.json({ lease: await sessions.takeControl(row) });
}));
}));
r.post('/sessions/:id/end', wrap(async (req, res) => {
r.post('/sessions/:id/end', wrap(async (req, res) => {
const row = await mine(req, res, 'run.session.control');
const row = await mine(req, res, 'run.session.control');
@@ -24,7 +24,7 @@ const { createServiceOutbox } = require('openvibe-sdk/events');
@@ -24,7 +24,7 @@ const { createServiceOutbox } = require('openvibe-sdk/events');
const TABLE = 'run_events_outbox'; // migrations/0001_jobs.sql
const TABLE = 'run_events_outbox'; // migrations/0001_jobs.sql
const ACTOR = { type: 'service', id: 'run' };
const ACTOR = { type: 'service', id: 'run' };
const EVENT_TYPES = ['run.job.queued', 'run.job.started', 'run.job.succeeded', 'run.job.failed', 'run.job.cancelled', 'run.job.expired',
const EVENT_TYPES = ['run.job.queued', 'run.job.started', 'run.job.succeeded', 'run.job.failed', 'run.job.cancelled', 'run.job.expired',
- 'run.session.started', 'run.session.ended',
+ 'run.session.started', 'run.session.ended', 'run.session.control_lease_changed', 'run.session.control_requested',
'agent.step.started', 'agent.step.updated', 'agent.step.ended']; // server/sessions/service.js (contracts 0.223.0)
'agent.step.started', 'agent.step.updated', 'agent.step.ended']; // server/sessions/service.js (contracts 0.223.0)
function createRunOutbox({ db, config, fetchImpl, now, log = console }) {
function createRunOutbox({ db, config, fetchImpl, now, log = console }) {
@@ -92,6 +92,9 @@ function createSessions({ config, db, desks, outbox, now = () => Date.now(), log
@@ -92,6 +92,9 @@ function createSessions({ config, db, desks, outbox, now = () => Date.now(), log
if (compute.pool && (typeof compute.pool !== 'string' || !/^[a-z0-9-]{1,40}$/.test(compute.pool))) throw new SessionError(422, 'run.compute.pool', 'invalid compute pool');
if (compute.pool && (typeof compute.pool !== 'string' || !/^[a-z0-9-]{1,40}$/.test(compute.pool))) throw new SessionError(422, 'run.compute.pool', 'invalid compute pool');
const pool = compute.pool || null;
const pool = compute.pool || null;
const strategy = compute.strategy || null;
const strategy = compute.strategy || null;
+ const template = compute.template || 'browser';
+ if (!['browser', 'kept-browser', 'throwaway'].includes(template)) throw new SessionError(422, 'run.session.template', 'template must be browser, kept-browser or throwaway');
+ if (template === 'kept-browser') throw new SessionError(501, 'run.session.template_unavailable', 'Kept browser profiles are not available on this machine.');
if (strategy && !require('openvibe-sdk/compute').STRATEGIES.includes(strategy)) throw new SessionError(422, 'run.compute.strategy', 'unknown compute strategy');
if (strategy && !require('openvibe-sdk/compute').STRATEGIES.includes(strategy)) throw new SessionError(422, 'run.compute.strategy', 'unknown compute strategy');
if (strategy && !pool) throw new SessionError(422, 'run.compute.pool', 'a strategy requires a compute pool');
if (strategy && !pool) throw new SessionError(422, 'run.compute.pool', 'a strategy requires a compute pool');
if (pool) { if (!broker) throw new SessionError(503, 'run.compute.off'); await broker.poolConfig(pool); }
if (pool) { if (!broker) throw new SessionError(503, 'run.compute.off'); await broker.poolConfig(pool); }
@@ -107,11 +110,14 @@ function createSessions({ config, db, desks, outbox, now = () => Date.now(), log
@@ -107,11 +110,14 @@ function createSessions({ config, db, desks, outbox, now = () => Date.now(), log
}
}
if ((await store.openCount(db, owner)) >= maxOpen) throw new SessionError(429, 'run.session.limit', `at most ${maxOpen} sessions at once: end one first`);
if ((await store.openCount(db, owner)) >= maxOpen) throw new SessionError(429, 'run.session.limit', `at most ${maxOpen} sessions at once: end one first`);
const at = now();
const at = now();
+ const id = store.newSessionId();
const row = {
const row = {
- id: store.newSessionId(), owner, via: via || null, kind: 'browser', state: 'placing',
+ id, owner, via: via || null, kind: 'browser', state: 'placing',
width: body.width || 1280, height: body.height || 800, ttl_s: body.ttl_s || 3600, idle_s: body.idle_s || 900,
width: body.width || 1280, height: body.height || 800, ttl_s: body.ttl_s || 3600, idle_s: body.idle_s || 900,
- compute_pool: pool, compute_strategy: strategy, compute_decision: null, machine_asked: body.machine || null, requirements: body.requirements || null, idempotency_key: body.idempotency_key || null,
- request_digest: digest, reasons: [], task_id: body.task_id || null, curtain: false, created_at: at, updated_at: at,
+ compute_pool: pool, compute_strategy: strategy, compute_decision: null, template, machine_asked: body.machine || null, requirements: body.requirements || null, idempotency_key: body.idempotency_key || null,
+ request_digest: digest, reasons: [], task_id: body.task_id || null, curtain: false,
+ control_lease: { thing: `run.session:${id}`, holder: { kind: 'person', subject: bare(owner) }, since_ms: at, expires_ms: at + 60_000, watchers: [], seq: 1 },
+ created_at: at, updated_at: at,
};
};
const out = await store.insert(db, row);
const out = await store.insert(db, row);
if (out.created) setImmediate(() => { tick().catch((e) => log.warn(`[Run] session tick: ${e.message}`)); });
if (out.created) setImmediate(() => { tick().catch((e) => log.warn(`[Run] session tick: ${e.message}`)); });
@@ -135,25 +141,130 @@ function createSessions({ config, db, desks, outbox, now = () => Date.now(), log
@@ -135,25 +141,130 @@ function createSessions({ config, db, desks, outbox, now = () => Date.now(), log
});
});
}
}
+ const HOLDER_MS = 60_000;
+ const ownerHolder = (row) => ({ kind: 'person', subject: bare(row.owner) });
+ const sameHolder = (a, b) => Boolean(a && b && a.kind === b.kind && a.subject === b.subject);
+ function actorHolder(actor, row) {
+ if (!actor) return ownerHolder(row);
+ if (actor.kind === 'person') return { kind: 'person', subject: String(actor.subject || bare(row.owner)) };
+ const kind = actor.kind === 'ai' ? 'ai' : null;
+ const subject = String(actor.subject || '').trim().slice(0, 200);
+ if (!kind || !subject) throw new SessionError(403, 'run.session.control_holder', 'the caller is not a named controller');
+ return { kind, subject };
+ }
+ async function saveLease(row, lease, reason) {
+ contracts.assertValid('common.control-lease@1', lease);
+ return db.tx(async (t) => {
+ const fresh = await store.get(t, row.id);
+ if (!fresh || fresh.state === 'ended') throw new SessionError(409, 'run.session.not_running', 'the session has ended');
+ const saved = await store.update(t, row.id, { control_lease: lease, updated_at: now() });
+ const prepared = outbox.events.prepare({
+ event_type: 'run.session.control_lease_changed', version: 1, source: 'run', actor: ACTOR,
+ timestamp: new Date(now()).toISOString(), subject: { type: 'session', id: row.id },
+ visibility: 'internal', priority: 'low', payload: { session_id: row.id, reason, lease },
+ }, { now: now() });
+ contracts.assertValid('events.event-envelope@1', prepared);
+ await t.maybe(`INSERT INTO ${EVENTS_TABLE} (event_id, envelope, created_at) VALUES ($1, $2, $3) ON CONFLICT (event_id) DO NOTHING RETURNING id`,
+ [`run:${row.id}:control-lease:${lease.seq}`, JSON.stringify(prepared), now()]);
+ return saved;
+ });
+ }
+ async function controlLease(row) {
+ let lease = row.control_lease || { thing: `run.session:${row.id}`, holder: ownerHolder(row), since_ms: Number(row.created_at), expires_ms: Number(row.created_at) + HOLDER_MS, watchers: [], seq: 1 };
+ contracts.assertValid('common.control-lease@1', lease);
+ if (lease.expires_ms != null && lease.expires_ms <= now()) {
+ lease = { ...lease, holder: { kind: 'nobody', subject: null }, since_ms: now(), expires_ms: null, seq: lease.seq + 1 };
+ row = await saveLease(row, lease, 'expired');
+ }
+ return row.control_lease || lease;
+ }
+ async function requestControl(row, actor) {
+ const current = await controlLease(row);
+ const target = actorHolder(actor, row);
+ if (sameHolder(current.holder, target)) return current;
+ if (current.holder.kind !== 'nobody') {
+ const prepared = outbox.events.prepare({ event_type: 'run.session.control_requested', version: 1, source: 'run', actor: ACTOR,
+ timestamp: new Date(now()).toISOString(), subject: { type: 'session', id: row.id }, visibility: 'internal', priority: 'low',
+ payload: { session_id: row.id, requester: target } }, { now: now() });
+ await db.maybe(`INSERT INTO ${EVENTS_TABLE} (event_id, envelope, created_at) VALUES ($1, $2, $3) ON CONFLICT (event_id) DO NOTHING RETURNING id`,
+ [`run:${row.id}:control-request:${target.kind}:${target.subject}:${Math.floor(now() / 1000)}`, JSON.stringify(prepared), now()]);
+ return current;
+ }
+ const next = { ...current, holder: target, since_ms: now(), expires_ms: now() + HOLDER_MS, seq: current.seq + 1 };
+ const saved = await saveLease(row, next, 'requested');
+ return saved.control_lease;
+ }
+ async function grantControl(row, target, reason = 'granted') {
+ const current = await controlLease(row);
+ if (!target || !['person', 'ai', 'nobody'].includes(target.kind)) throw new SessionError(422, 'run.session.control_target', 'target kind is person, ai or nobody');
+ const subject = target.kind === 'nobody' ? null : String(target.subject || '').trim().slice(0, 200);
+ if (target.kind !== 'nobody' && !subject) throw new SessionError(422, 'run.session.control_target', 'a person or AI controller needs a subject');
+ if (target.kind === 'person' && !require('openvibe-contracts').ids.isSubjectId('user', subject)) throw new SessionError(422, 'run.session.control_target', 'a person controller needs a Network usr_ subject');
+ const holder = { kind: target.kind, subject };
+ if (sameHolder(holder, current.holder)) return current;
+ const next = { ...current, holder, since_ms: now(), expires_ms: holder.kind === 'nobody' ? null : now() + HOLDER_MS, seq: current.seq + 1 };
+ const saved = await saveLease(row, next, reason);
+ if (holder.kind === 'person' && holder.subject !== bare(row.owner)) {
+ await db.maybe('INSERT INTO run_session_control_guests (session_id, subject, created_at) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING RETURNING session_id', [row.id, holder.subject, now()]);
+ }
+ return saved.control_lease;
+ }
+ async function takeControl(row) { return grantControl(row, ownerHolder(row), 'owner_take_back'); }
+
const RESULTS = new Set(['ack', 'nack', 'refused', 'expired']);
const RESULTS = new Set(['ack', 'nack', 'refused', 'expired']);
- async function act(row, cmd) {
+ async function act(row, cmd, actor = null) {
const v = contracts.validate('run.session-control-request@1', cmd);
const v = contracts.validate('run.session-control-request@1', cmd);
if (!v.valid) throw new SessionError(422, 'run.session.invalid', `invalid command: ${(v.errors && v.errors[0] && v.errors[0].message) || 'bad value'}`.slice(0, 300));
if (!v.valid) throw new SessionError(422, 'run.session.invalid', `invalid command: ${(v.errors && v.errors[0] && v.errors[0].message) || 'bad value'}`.slice(0, 300));
- if (row.state !== 'running') throw new SessionError(409, 'run.session.not_running', row.state === 'ended' ? 'the session has ended' : 'the session\'s desk is not running yet');
const id = cmd.id || `c-${now().toString(36)}`;
const id = cmd.id || `c-${now().toString(36)}`;
+ if (row.state !== 'running') throw new SessionError(409, 'run.session.not_running', row.state === 'ended' ? 'the session has ended' : 'the session\'s desk is not running yet');
+ const lease = await controlLease(row);
+ const caller = actorHolder(actor, row);
+ if (!sameHolder(lease.holder, caller)) {
+ await logAction(row, id, cmd, caller, 'refused', 'run.session.control_held');
+ throw new SessionError(403, 'run.session.control_held', 'only the current control holder may send commands');
+ }
const r = await desks.act(row.machine, row.desk, bare(row.owner), { id, kind: cmd.kind, value: cmd.value });
const r = await desks.act(row.machine, row.desk, bare(row.owner), { id, kind: cmd.kind, value: cmd.value });
if (r === null) {
if (r === null) {
const ended = await finish(row, 'host_lost', 'its machine no longer holds the desk', { release: false });
const ended = await finish(row, 'host_lost', 'its machine no longer holds the desk', { release: false });
return { session_id: row.id, id, result: 'refused', code: 'run.session.host_lost', reason: 'the session\'s machine went away', state: ended ? ended.state : 'ended' };
return { session_id: row.id, id, result: 'refused', code: 'run.session.host_lost', reason: 'the session\'s machine went away', state: ended ? ended.state : 'ended' };
}
}
- await store.update(db, row.id, { last_used_at: now(), updated_at: now() });
+ const renewed = { ...lease, expires_ms: now() + HOLDER_MS };
+ if (renewed.expires_ms !== lease.expires_ms) await store.update(db, row.id, { control_lease: renewed, last_used_at: now(), updated_at: now() });
+ else await store.update(db, row.id, { last_used_at: now(), updated_at: now() });
const out = { session_id: row.id, id: String(r.id || id).slice(0, 64), result: RESULTS.has(r.result) ? r.result : 'expired', state: 'running' };
const out = { session_id: row.id, id: String(r.id || id).slice(0, 64), result: RESULTS.has(r.result) ? r.result : 'expired', state: 'running' };
if (r.code != null) out.code = String(r.code).slice(0, 80);
if (r.code != null) out.code = String(r.code).slice(0, 80);
if (r.reason != null) out.reason = String(r.reason).slice(0, 1000);
if (r.reason != null) out.reason = String(r.reason).slice(0, 1000);
if (r.data && typeof r.data === 'object' && out.result === 'ack') out.data = r.data;
if (r.data && typeof r.data === 'object' && out.result === 'ack') out.data = r.data;
+ await logAction(row, out.id, cmd, caller, out.result, out.code || null);
return out;
return out;
}
}
+ async function logAction(row, id, cmd, actor, result, code) {
+ const kind = String(cmd.kind || 'unknown').slice(0, 40);
+ const action = kind === 'browser' && cmd.value && typeof cmd.value.action === 'string' ? cmd.value.action.slice(0, 48) : null;
+ await db.maybe(`INSERT INTO run_session_actions (session_id, command_id, kind, action, actor_kind, actor_subject, result, code, created_at)
+ VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9) RETURNING id`, [row.id, String(id).slice(0, 64), kind, action, actor.kind,
+ String(actor.subject || '').slice(0, 200), String(result).slice(0, 24), code ? String(code).slice(0, 80) : null, now()]);
+ }
+ async function timeline(row, limit = 50, before = null) {
+ const n = Math.min(100, Math.max(1, Number(limit) || 50));
+ const args = [row.id];
+ const cursor = Number(before);
+ const where = before && Number.isInteger(cursor) ? ` AND id < $${args.push(cursor)}` : '';
+ args.push(n + 1);
+ const rows = await db.many(`SELECT id, command_id, kind, action, actor_kind, actor_subject, result, code, created_at
+ FROM run_session_actions WHERE session_id = $1${where} ORDER BY id DESC LIMIT $${args.length}`, args);
+ const more = rows.length > n;
+ const actions = rows.slice(0, n).map((a) => ({ id: Number(a.id), command_id: a.command_id, kind: a.kind, action: a.action,
+ actor: { kind: a.actor_kind, subject: a.actor_subject }, result: a.result, code: a.code, at: new Date(Number(a.created_at)).toISOString() }));
+ return { actions, next_cursor: more ? String(actions.at(-1).id) : null };
+ }
+ async function isGuest(row, subject) {
+ if (!row || !subject || String(subject) === bare(row.owner)) return false;
+ return Boolean(await db.maybe('SELECT 1 FROM run_session_control_guests WHERE session_id = $1 AND subject = $2', [row.id, String(subject)]));
+ }
+ function snapshot() { return { available: false, message: 'Snapshots are not available on this machine.' }; }
+
/**
/**
* The session's newest frame, as a small JPEG (plan 2026-11-07: the live still Surf's card shows and the Dock
* The session's newest frame, as a small JPEG (plan 2026-11-07: the live still Surf's card shows and the Dock
* links to). It reuses the desk's own capture — the `screen` command on the session's lease, the same one Actor's
* links to). It reuses the desk's own capture — the `screen` command on the session's lease, the same one Actor's
@@ -274,6 +385,7 @@ function createSessions({ config, db, desks, outbox, now = () => Date.now(), log
@@ -274,6 +385,7 @@ function createSessions({ config, db, desks, outbox, now = () => Date.now(), log
continue;
continue;
}
}
// running
// running
+ await controlLease(row);
if (now() - row.created_at >= row.ttl_s * 1000) { await finish(row, 'ttl', `its ${row.ttl_s} s ran out`); continue; }
if (now() - row.created_at >= row.ttl_s * 1000) { await finish(row, 'ttl', `its ${row.ttl_s} s ran out`); continue; }
if (now() - (row.last_used_at || row.started_at) >= row.idle_s * 1000) { await finish(row, 'idle', `no command for ${row.idle_s} s`); continue; }
if (now() - (row.last_used_at || row.started_at) >= row.idle_s * 1000) { await finish(row, 'idle', `no command for ${row.idle_s} s`); continue; }
if (now() - (row.lease_checked_at || 0) >= leaseCheckMs) {
if (now() - (row.lease_checked_at || 0) >= leaseCheckMs) {
@@ -300,7 +412,7 @@ function createSessions({ config, db, desks, outbox, now = () => Date.now(), log
@@ -300,7 +412,7 @@ function createSessions({ config, db, desks, outbox, now = () => Date.now(), log
}
}
return {
return {
- create, act, finish, still, setCurtain, tick, SessionError,
+ create, act, finish, still, setCurtain, requestControl, grantControl, takeControl, controlLease, timeline, isGuest, snapshot, tick, SessionError,
get: (id) => store.get(db, id),
get: (id) => store.get(db, id),
list: (owner, opts) => store.listOf(db, owner, opts),
list: (owner, opts) => store.listOf(db, owner, opts),
present: (row) => store.present(row, now()),
present: (row) => store.present(row, now()),
@@ -14,8 +14,8 @@ const { ids } = require('openvibe-contracts');
@@ -14,8 +14,8 @@ const { ids } = require('openvibe-contracts');
const COLUMNS = ['id', 'owner', 'via', 'kind', 'state', 'width', 'height', 'ttl_s', 'idle_s', 'machine_asked', 'requirements', 'idempotency_key',
const COLUMNS = ['id', 'owner', 'via', 'kind', 'state', 'width', 'height', 'ttl_s', 'idle_s', 'machine_asked', 'requirements', 'idempotency_key',
'compute_pool', 'compute_strategy', 'compute_decision', 'request_digest', 'machine', 'desk', 'trust', 'offer_id', 'reasons', 'created_at', 'started_at', 'last_used_at', 'ended_at', 'end_reason',
'compute_pool', 'compute_strategy', 'compute_decision', 'request_digest', 'machine', 'desk', 'trust', 'offer_id', 'reasons', 'created_at', 'started_at', 'last_used_at', 'ended_at', 'end_reason',
- 'end_detail', 'metered_minutes', 'lease_checked_at', 'task_id', 'curtain', 'updated_at'];
-const JSON_COLUMNS = new Set(['requirements', 'reasons', 'compute_decision']);
+ 'end_detail', 'metered_minutes', 'lease_checked_at', 'task_id', 'curtain', 'control_lease', 'template', 'updated_at'];
+const JSON_COLUMNS = new Set(['requirements', 'reasons', 'compute_decision', 'control_lease']);
const iso = (ms) => (ms == null ? null : new Date(Number(ms)).toISOString());
const iso = (ms) => (ms == null ? null : new Date(Number(ms)).toISOString());
const json = (v, dflt) => { if (v == null) return dflt; if (typeof v === 'object') return v; try { return JSON.parse(v); } catch { return dflt; } };
const json = (v, dflt) => { if (v == null) return dflt; if (typeof v === 'object') return v; try { return JSON.parse(v); } catch { return dflt; } };
@@ -78,6 +78,17 @@ function fakeDesks() {
@@ -78,6 +78,17 @@ function fakeDesks() {
assert.ok(started.some((p) => p.id === ses.id && p.state === 'running'), 'run.session.started');
assert.ok(started.some((p) => p.id === ses.id && p.state === 'running'), 'run.session.started');
});
});
+ await check('browser and throwaway presets are accepted; kept profiles and snapshots say plainly when unavailable', async () => {
+ const unavailable = await t.call('POST', '/api/v1/sessions?template=kept-browser', { ...as(bea), body: { kind: 'browser' } });
+ assert.strictEqual(unavailable.status, 501);
+ assert.match(unavailable.json.detail, /not available on this machine/);
+ const throwaway = await t.call('POST', '/api/v1/sessions?template=throwaway', { ...as(bea), body: { kind: 'browser' } });
+ assert.strictEqual(throwaway.status, 201, throwaway.text);
+ await sessions.finish(await sessions.get(throwaway.json.id), 'ended', 'test complete');
+ const snapshot = await t.call('GET', `/api/v1/sessions/${ses.id}/snapshot`, as(ada));
+ assert.deepStrictEqual(snapshot.json, { available: false, message: 'Snapshots are not available on this machine.' });
+ });
+
await check('commands reach its desk; anyone else\'s call is 404', async () => {
await check('commands reach its desk; anyone else\'s call is 404', async () => {
const r = await t.call('POST', `/api/v1/sessions/${ses.id}/commands`, { ...as(ada), body: { id: 'c1', kind: 'browser', value: { action: 'open', url: 'https://example.org/' } } });
const r = await t.call('POST', `/api/v1/sessions/${ses.id}/commands`, { ...as(ada), body: { id: 'c1', kind: 'browser', value: { action: 'open', url: 'https://example.org/' } } });
assert.strictEqual(r.status, 200, r.text);
assert.strictEqual(r.status, 200, r.text);
@@ -94,6 +105,40 @@ function fakeDesks() {
@@ -94,6 +105,40 @@ function fakeDesks() {
valid(list, 'run.session-list-result@1');
valid(list, 'run.session-list-result@1');
assert.deepStrictEqual(list.sessions.map((s) => s.id), [ses.id]);
assert.deepStrictEqual(list.sessions.map((s) => s.id), [ses.id]);
assert.deepStrictEqual((await t.call('GET', '/api/v1/sessions', as(bea))).json.sessions, []);
assert.deepStrictEqual((await t.call('GET', '/api/v1/sessions', as(bea))).json.sessions, []);
+ const timeline = await t.call('GET', `/api/v1/sessions/${ses.id}/timeline`, as(ada));
+ assert.strictEqual(timeline.status, 200, timeline.text);
+ assert.strictEqual(timeline.json.actions[0].command_id, 'c1');
+ assert.strictEqual(timeline.json.actions[0].action, 'open');
+ assert.ok(!JSON.stringify(timeline.json).includes('https://example.org'), 'timeline omits command values and URLs');
+ assert.strictEqual((await t.call('GET', `/api/v1/sessions/${ses.id}/timeline`, as(bea))).status, 404);
+ });
+
+ await check('the B3 control lease hands off to an agent, refuses the former holder, and owner take-back wins', async () => {
+ const getLease = async () => (await t.call('GET', `/api/v1/sessions/${ses.id}/control`, as(ada))).json.lease;
+ let lease = getLease ? await getLease() : null;
+ valid(lease, 'common.control-lease@1');
+ assert.deepStrictEqual(lease.holder, { kind: 'person', subject: ada });
+ const agent = { token: t.network.signService({ sub: 'svc:actor', cap: ['run.session.control'] }), headers: { 'X-OV-Subject': ada } };
+ assert.strictEqual((await t.call('POST', `/api/v1/sessions/${ses.id}/commands`, { ...agent, body: { kind: 'screen', value: {} } })).status, 403, 'an agent without the lease cannot control the desk');
+ const grant = await t.call('POST', `/api/v1/sessions/${ses.id}/control/grant`, { ...as(ada), body: { holder: { kind: 'ai', subject: 'svc:actor' } } });
+ valid(grant.json.lease, 'common.control-lease@1');
+ const held = await t.call('POST', `/api/v1/sessions/${ses.id}/commands`, { ...agent, body: { kind: 'screen', value: {} } });
+ assert.strictEqual(held.status, 200, held.text);
+ assert.strictEqual((await t.call('POST', `/api/v1/sessions/${ses.id}/commands`, { ...as(ada), body: { kind: 'screen', value: {} } })).status, 403, 'owner must take back before sending input');
+ lease = (await t.call('POST', `/api/v1/sessions/${ses.id}/control/take`, as(ada))).json.lease;
+ assert.deepStrictEqual(lease.holder, { kind: 'person', subject: ada });
+ assert.ok(lease.seq > grant.json.lease.seq);
+ const guest = ids.newId('user');
+ await t.call('POST', `/api/v1/sessions/${ses.id}/control/grant`, { ...as(ada), body: { holder: { kind: 'person', subject: guest } } });
+ const guestCall = await t.call('POST', `/api/v1/sessions/${ses.id}/commands`, { ...as(guest), body: { kind: 'screen', value: {} } });
+ assert.strictEqual(guestCall.status, 200, guestCall.text);
+ assert.strictEqual((await t.call('POST', `/api/v1/sessions/${ses.id}/commands`, { ...as(ada), body: { kind: 'screen', value: {} } })).status, 403);
+ lease = (await t.call('POST', `/api/v1/sessions/${ses.id}/control/take`, as(ada))).json.lease;
+ t.clock.offset += 61_000;
+ lease = (await t.call('GET', `/api/v1/sessions/${ses.id}/control`, as(ada))).json.lease;
+ assert.deepStrictEqual(lease.holder, { kind: 'nobody', subject: null }, 'a quiet holder expires lazily');
+ const events = (await t.outboxRows('run.session.control_lease_changed')).map((e) => e.payload || e);
+ assert.ok(events.some((e) => e.lease && e.lease.seq === lease.seq), 'lease transitions are durable events');
});
});
await check('a first-party service opens one for a person, with the capability; an app never does', async () => {
await check('a first-party service opens one for a person, with the capability; an app never does', async () => {