openvibers/openvibe.live Commits
Agent Streams S3: the stream part of the session report (#631)
Changes
nnew file mode 100644@@@ -0,0 +1,33 @@++'use strict';++++/**++ * Agent Streams S3: the stream part of a session report.++ *++ * When an agent stream ends, Live tells OpenVibe.Report about it. The report the tool wrote for the same task (the++ * same source and task ref) gets the stream added to it; when no tool finished the session - the stream was the++ * session - Report writes a stream-only report instead.++ *++ * The payload is the stream's own: its public id (ags_<ULID>), the VOD it left behind, the task it was for and the++ * person it was theirs.++ */++function buildStreamReport(row, next, worked) {++ const task = (row && row.task && typeof row.task === 'object') ? row.task : {};++ return {++ source: task.source || 'desk',++ task_ref: task.ref || null,++ ags_id: (row && row.public_id) || null,++ vod_id: (row && row.vod_id) || null,++ owner: (row && row.owner_subject) || null,++ task: {++ title: task.title || (row && row.title) || 'An agent session',++ ref: task.ref || (row && row.public_id) || '',++ url: task.url || null,++ },++ started_at: (row && row.started_at) || null,++ ended_at: (next && next.ended_at) || null,++ worked_seconds: Number(worked) || 0,++ visibility: (row && row.visibility) || 'private',++ };++}++++module.exports = { buildStreamReport };@@@ -285,6 +285,13 @@ async function markPublishing(row) {}/** The stream ended. Worked seconds are frozen, the session (if any) and every publish ticket go. */++let reportClient = null;++/** Live's client for OpenVibe.Report (the session report of an agent stream). */++function reportClientFor() {++ if (!reportClient) reportClient = require('../report').createReportClient({ config, log: console });++ return reportClient;++}++async function end(row, { reason = 'ended' } = {}) {if (row.state === 'ended') return row;const session = await liveSession(row.managed_stream_id);@@@ -305,6 +312,14 @@ async function end(row, { reason = 'ended' } = {}) {} catch (err) { console.warn(`[AgentStream] ending session ${session.id} failed: ${err.message}`); }}try { require('../streaming/curtains').lower(session ? session.id : null, { notify: false }); } catch { /* */ }++ // Agent Streams S3: the stream part of the session report. The report the tool wrote for this task gets the++ // stream added to it; when no tool finished the session, a stream-only report is written instead. A report that++ // cannot be written never changes the stream: it ended either way.++ try {++ await reportClientFor().sendStream(require('./report').buildStreamReport(row, next, worked));++ } catch (err) {++ console.warn('[AgentStream] the session report of ' + row.public_id + ' failed: ' + String((err && err.message) || err).slice(0, 200));++ }return next;}nnew file mode 100644@@@ -0,0 +1,62 @@++'use strict';++++/**++ * The session report of an agent stream (Agent Streams S3).++ *++ * When an agent stream ends, Live tells OpenVibe.Report about it. The report the tool wrote for the same task gets++ * the stream added to it (the stream part); when no tool finished the session - the stream was the session - a++ * stream-only report is written instead.++ *++ * The token is Live's own, for the audience openvibe.report, with the capability report.session.write (Network's++ * principals seed).++ */++const TOKEN_TTL_MS = 10 * 60_000;++++function createReportClient({ config = {}, fetchImpl = globalThis.fetch, log = console, now = () => Date.now() } = {}) {++ const base = String(config.reportInternalUrl || process.env.OV_REPORT_INTERNAL_URL || 'http://127.0.0.1:5080').replace(/\/+$/, '');++ const networkUrl = String(config.networkInternalUrl || process.env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/+$/, '');++ let cached = null;++++ async function token() {++ if (cached && cached.until > now()) return cached.token;++ const clientId = (config.oauth && config.oauth.clientId) || process.env.OV_OAUTH_CLIENT_ID || 'live';++ const clientSecret = (config.oauth && config.oauth.clientSecret) || process.env.OV_OAUTH_CLIENT_SECRET || '';++ if (!clientId || !clientSecret) throw new Error('Live has no Network client secret, so it cannot write a session report');++ const body = new URLSearchParams({++ grant_type: 'client_credentials', client_id: clientId, client_secret: clientSecret,++ audience: 'openvibe.report', scope: 'report.session.write',++ });++ const res = await fetchImpl(`${networkUrl}/oauth/token`, {++ method: 'POST',++ headers: { 'Content-Type': 'application/x-www-form-urlencoded', Accept: 'application/json' },++ body: body.toString(),++ signal: AbortSignal.timeout(15_000),++ });++ const doc = await res.json().catch(() => null);++ if (!res.ok || !doc || !doc.access_token) throw new Error(`OpenVibe.Network answered ${res.status}`);++ cached = { token: String(doc.access_token), until: now() + TOKEN_TTL_MS };++ return cached.token;++ }++++ /** Tell Report that a stream ended. */++ async function sendStream(payload) {++ const t = await token();++ const res = await fetchImpl(`${base}/internal/session-reports/stream`, {++ method: 'POST',++ headers: { 'Content-Type': 'application/json', Accept: 'application/json', Authorization: `Bearer ${t}` },++ body: JSON.stringify(payload),++ signal: AbortSignal.timeout(15_000),++ });++ const doc = await res.json().catch(() => null);++ if (!res.ok) {++ const err = new Error((doc && (doc.detail || doc.error)) || `OpenVibe.Report answered ${res.status}`);++ err.status = res.status;++ throw err;++ }++ return doc;++ }++++ return { sendStream, token, baseUrl: base };++}++++module.exports = { createReportClient };@@@ -336,6 +336,7 @@ async function check(name, fn) {const REVIEWED = {'server/net/egress.js': 'the guard itself','server/drill.js': 'the restore-drill guard (refuses connections)',++ 'server/report.js': 'OpenVibe.Report (OV_REPORT_INTERNAL_URL, configured) for the report of an agent stream, and the Network token endpoint (OV_NETWORK_INTERNAL_URL, configured) for the Live service token; the paths are fixed, no URL from a person','server/config.js': 'Network registry (configured)','server/media-client.js': 'OpenVibe.Media (configured)','server/media-proxy/live-thumbs.js': 'OpenVibe.Media (configured)',nnew file mode 100644@@@ -0,0 +1,64 @@++'use strict';++/**++ * Agent Streams S3: Live adds the stream part to the session report.++ *++ * node test/session-report.test.js++ */++const assert = require('assert');++const { buildStreamReport } = require('../server/agent-streams/report');++const { createReportClient } = require('../server/report');++++const OWNER = 'usr_01JAN2C3D4E5F6G7H8J9K0MNPS';++++(async () => {++ // The stream part of the report.++ const payload = buildStreamReport({++ public_id: 'ags_01JAN2C3D4E5F6G7H8J9K0MNPS',++ vod_id: 'vod_01JAN2C3D4E5F6G7H8J9K0MNPS',++ owner_subject: OWNER,++ title: 'The agent at work',++ visibility: 'public',++ task: { source: 'codes', ref: 'tsk_01JAN2C3D4E5F6G7H8J9K0MNPS', title: 'Ship the checkout page', url: 'https://openvibe.codes/t/tsk_1' },++ started_at: '2026-11-06T18:00:00.000Z',++ }, { ended_at: '2026-11-06T18:42:00.000Z' }, 2520);++ assert.strictEqual(payload.ags_id, 'ags_01JAN2C3D4E5F6G7H8J9K0MNPS', 'the stream names itself');++ assert.strictEqual(payload.vod_id, 'vod_01JAN2C3D4E5F6G7H8J9K0MNPS', 'and the VOD it left behind');++ assert.strictEqual(payload.source, 'codes', 'the source is the one that ran the session');++ assert.strictEqual(payload.task_ref, 'tsk_01JAN2C3D4E5F6G7H8J9K0MNPS', 'the task ref matches the tool\'s report');++ assert.strictEqual(payload.owner, OWNER, 'the report is the person\'s');++ assert.strictEqual(payload.worked_seconds, 2520, '42 minutes of work');++ assert.strictEqual(payload.visibility, 'public', 'the stream\'s own visibility');++++ // A stream of nobody\'s task still names its stream.++ const alone = buildStreamReport({ public_id: 'ags_01JAN2C3D4E5F6G7H8J9K0MNPS' }, null, 0);++ assert.strictEqual(alone.source, 'desk');++ assert.strictEqual(alone.task_ref, null, 'no tool finished it: a stream-only report');++ assert.strictEqual(alone.task.ref, 'ags_01JAN2C3D4E5F6G7H8J9K0MNPS');++++ // The client: Live\'s own token for Report\'s audience, and the stream in the body.++ const calls = [];++ const fetchImpl = async (url, opts = {}) => {++ calls.push({ url: String(url), opts });++ if (String(url).includes('/oauth/token')) {++ return { ok: true, status: 200, json: async () => ({ access_token: 'the-service-token' }) };++ }++ return { ok: true, status: 200, json: async () => ({ id: 'rps_01JAN2C3D4E5F6G7H8J9K0MNPS', url: 'https://openvibe.report/s/rps_1', created: false }) };++ };++ const client = createReportClient({++ config: { networkInternalUrl: 'http://127.0.0.1:4000', reportInternalUrl: 'http://127.0.0.1:5080', oauth: { clientId: 'live', clientSecret: 'the-secret' } },++ fetchImpl,++ });++ await client.sendStream(payload);++ const token = calls.find((c) => c.url.includes('/oauth/token'));++ assert.ok(token, 'Live asks Network for a token');++ assert.ok(String(token.opts.body).includes('audience=openvibe.report'), 'for Report\'s audience');++ assert.ok(String(token.opts.body).includes('scope=report.session.write'), 'with the report capability');++ const post = calls.find((c) => c.url.includes('/internal/session-reports/stream'));++ assert.ok(post, 'Live posts the stream to Report');++ assert.strictEqual(post.opts.method, 'POST');++ assert.strictEqual(post.opts.headers.Authorization, 'Bearer the-service-token');++ const sent = JSON.parse(post.opts.body);++ assert.strictEqual(sent.ags_id, 'ags_01JAN2C3D4E5F6G7H8J9K0MNPS', 'the stream part goes with it');++++ console.log(' ✓ Live adds the stream part to the session report');++})().catch((e) => { console.error(e); process.exit(1); });
new file mode 100644
new file mode 100644
@@ -0,0 +1,33 @@
@@ -0,0 +1,33 @@
+'use strict';
+
+/**
+ * Agent Streams S3: the stream part of a session report.
+ *
+ * When an agent stream ends, Live tells OpenVibe.Report about it. The report the tool wrote for the same task (the
+ * same source and task ref) gets the stream added to it; when no tool finished the session - the stream was the
+ * session - Report writes a stream-only report instead.
+ *
+ * The payload is the stream's own: its public id (ags_<ULID>), the VOD it left behind, the task it was for and the
+ * person it was theirs.
+ */
+function buildStreamReport(row, next, worked) {
+ const task = (row && row.task && typeof row.task === 'object') ? row.task : {};
+ return {
+ source: task.source || 'desk',
+ task_ref: task.ref || null,
+ ags_id: (row && row.public_id) || null,
+ vod_id: (row && row.vod_id) || null,
+ owner: (row && row.owner_subject) || null,
+ task: {
+ title: task.title || (row && row.title) || 'An agent session',
+ ref: task.ref || (row && row.public_id) || '',
+ url: task.url || null,
+ },
+ started_at: (row && row.started_at) || null,
+ ended_at: (next && next.ended_at) || null,
+ worked_seconds: Number(worked) || 0,
+ visibility: (row && row.visibility) || 'private',
+ };
+}
+
+module.exports = { buildStreamReport };
@@ -285,6 +285,13 @@ async function markPublishing(row) {
@@ -285,6 +285,13 @@ async function markPublishing(row) {
}
}
/** The stream ended. Worked seconds are frozen, the session (if any) and every publish ticket go. */
/** The stream ended. Worked seconds are frozen, the session (if any) and every publish ticket go. */
+let reportClient = null;
+/** Live's client for OpenVibe.Report (the session report of an agent stream). */
+function reportClientFor() {
+ if (!reportClient) reportClient = require('../report').createReportClient({ config, log: console });
+ return reportClient;
+}
+
async function end(row, { reason = 'ended' } = {}) {
async function end(row, { reason = 'ended' } = {}) {
if (row.state === 'ended') return row;
if (row.state === 'ended') return row;
const session = await liveSession(row.managed_stream_id);
const session = await liveSession(row.managed_stream_id);
@@ -305,6 +312,14 @@ async function end(row, { reason = 'ended' } = {}) {
@@ -305,6 +312,14 @@ async function end(row, { reason = 'ended' } = {}) {
} catch (err) { console.warn(`[AgentStream] ending session ${session.id} failed: ${err.message}`); }
} catch (err) { console.warn(`[AgentStream] ending session ${session.id} failed: ${err.message}`); }
}
}
try { require('../streaming/curtains').lower(session ? session.id : null, { notify: false }); } catch { /* */ }
try { require('../streaming/curtains').lower(session ? session.id : null, { notify: false }); } catch { /* */ }
+ // Agent Streams S3: the stream part of the session report. The report the tool wrote for this task gets the
+ // stream added to it; when no tool finished the session, a stream-only report is written instead. A report that
+ // cannot be written never changes the stream: it ended either way.
+ try {
+ await reportClientFor().sendStream(require('./report').buildStreamReport(row, next, worked));
+ } catch (err) {
+ console.warn('[AgentStream] the session report of ' + row.public_id + ' failed: ' + String((err && err.message) || err).slice(0, 200));
+ }
return next;
return next;
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,62 @@
@@ -0,0 +1,62 @@
+'use strict';
+
+/**
+ * The session report of an agent stream (Agent Streams S3).
+ *
+ * When an agent stream ends, Live tells OpenVibe.Report about it. The report the tool wrote for the same task gets
+ * the stream added to it (the stream part); when no tool finished the session - the stream was the session - a
+ * stream-only report is written instead.
+ *
+ * The token is Live's own, for the audience openvibe.report, with the capability report.session.write (Network's
+ * principals seed).
+ */
+const TOKEN_TTL_MS = 10 * 60_000;
+
+function createReportClient({ config = {}, fetchImpl = globalThis.fetch, log = console, now = () => Date.now() } = {}) {
+ const base = String(config.reportInternalUrl || process.env.OV_REPORT_INTERNAL_URL || 'http://127.0.0.1:5080').replace(/\/+$/, '');
+ const networkUrl = String(config.networkInternalUrl || process.env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/+$/, '');
+ let cached = null;
+
+ async function token() {
+ if (cached && cached.until > now()) return cached.token;
+ const clientId = (config.oauth && config.oauth.clientId) || process.env.OV_OAUTH_CLIENT_ID || 'live';
+ const clientSecret = (config.oauth && config.oauth.clientSecret) || process.env.OV_OAUTH_CLIENT_SECRET || '';
+ if (!clientId || !clientSecret) throw new Error('Live has no Network client secret, so it cannot write a session report');
+ const body = new URLSearchParams({
+ grant_type: 'client_credentials', client_id: clientId, client_secret: clientSecret,
+ audience: 'openvibe.report', scope: 'report.session.write',
+ });
+ const res = await fetchImpl(`${networkUrl}/oauth/token`, {
+ method: 'POST',
+ headers: { 'Content-Type': 'application/x-www-form-urlencoded', Accept: 'application/json' },
+ body: body.toString(),
+ signal: AbortSignal.timeout(15_000),
+ });
+ const doc = await res.json().catch(() => null);
+ if (!res.ok || !doc || !doc.access_token) throw new Error(`OpenVibe.Network answered ${res.status}`);
+ cached = { token: String(doc.access_token), until: now() + TOKEN_TTL_MS };
+ return cached.token;
+ }
+
+ /** Tell Report that a stream ended. */
+ async function sendStream(payload) {
+ const t = await token();
+ const res = await fetchImpl(`${base}/internal/session-reports/stream`, {
+ method: 'POST',
+ headers: { 'Content-Type': 'application/json', Accept: 'application/json', Authorization: `Bearer ${t}` },
+ body: JSON.stringify(payload),
+ signal: AbortSignal.timeout(15_000),
+ });
+ const doc = await res.json().catch(() => null);
+ if (!res.ok) {
+ const err = new Error((doc && (doc.detail || doc.error)) || `OpenVibe.Report answered ${res.status}`);
+ err.status = res.status;
+ throw err;
+ }
+ return doc;
+ }
+
+ return { sendStream, token, baseUrl: base };
+}
+
+module.exports = { createReportClient };
@@ -336,6 +336,7 @@ async function check(name, fn) {
@@ -336,6 +336,7 @@ async function check(name, fn) {
const REVIEWED = {
const REVIEWED = {
'server/net/egress.js': 'the guard itself',
'server/net/egress.js': 'the guard itself',
'server/drill.js': 'the restore-drill guard (refuses connections)',
'server/drill.js': 'the restore-drill guard (refuses connections)',
+ 'server/report.js': 'OpenVibe.Report (OV_REPORT_INTERNAL_URL, configured) for the report of an agent stream, and the Network token endpoint (OV_NETWORK_INTERNAL_URL, configured) for the Live service token; the paths are fixed, no URL from a person',
'server/config.js': 'Network registry (configured)',
'server/config.js': 'Network registry (configured)',
'server/media-client.js': 'OpenVibe.Media (configured)',
'server/media-client.js': 'OpenVibe.Media (configured)',
'server/media-proxy/live-thumbs.js': 'OpenVibe.Media (configured)',
'server/media-proxy/live-thumbs.js': 'OpenVibe.Media (configured)',
new file mode 100644
new file mode 100644
@@ -0,0 +1,64 @@
@@ -0,0 +1,64 @@
+'use strict';
+/**
+ * Agent Streams S3: Live adds the stream part to the session report.
+ *
+ * node test/session-report.test.js
+ */
+const assert = require('assert');
+const { buildStreamReport } = require('../server/agent-streams/report');
+const { createReportClient } = require('../server/report');
+
+const OWNER = 'usr_01JAN2C3D4E5F6G7H8J9K0MNPS';
+
+(async () => {
+ // The stream part of the report.
+ const payload = buildStreamReport({
+ public_id: 'ags_01JAN2C3D4E5F6G7H8J9K0MNPS',
+ vod_id: 'vod_01JAN2C3D4E5F6G7H8J9K0MNPS',
+ owner_subject: OWNER,
+ title: 'The agent at work',
+ visibility: 'public',
+ task: { source: 'codes', ref: 'tsk_01JAN2C3D4E5F6G7H8J9K0MNPS', title: 'Ship the checkout page', url: 'https://openvibe.codes/t/tsk_1' },
+ started_at: '2026-11-06T18:00:00.000Z',
+ }, { ended_at: '2026-11-06T18:42:00.000Z' }, 2520);
+ assert.strictEqual(payload.ags_id, 'ags_01JAN2C3D4E5F6G7H8J9K0MNPS', 'the stream names itself');
+ assert.strictEqual(payload.vod_id, 'vod_01JAN2C3D4E5F6G7H8J9K0MNPS', 'and the VOD it left behind');
+ assert.strictEqual(payload.source, 'codes', 'the source is the one that ran the session');
+ assert.strictEqual(payload.task_ref, 'tsk_01JAN2C3D4E5F6G7H8J9K0MNPS', 'the task ref matches the tool\'s report');
+ assert.strictEqual(payload.owner, OWNER, 'the report is the person\'s');
+ assert.strictEqual(payload.worked_seconds, 2520, '42 minutes of work');
+ assert.strictEqual(payload.visibility, 'public', 'the stream\'s own visibility');
+
+ // A stream of nobody\'s task still names its stream.
+ const alone = buildStreamReport({ public_id: 'ags_01JAN2C3D4E5F6G7H8J9K0MNPS' }, null, 0);
+ assert.strictEqual(alone.source, 'desk');
+ assert.strictEqual(alone.task_ref, null, 'no tool finished it: a stream-only report');
+ assert.strictEqual(alone.task.ref, 'ags_01JAN2C3D4E5F6G7H8J9K0MNPS');
+
+ // The client: Live\'s own token for Report\'s audience, and the stream in the body.
+ const calls = [];
+ const fetchImpl = async (url, opts = {}) => {
+ calls.push({ url: String(url), opts });
+ if (String(url).includes('/oauth/token')) {
+ return { ok: true, status: 200, json: async () => ({ access_token: 'the-service-token' }) };
+ }
+ return { ok: true, status: 200, json: async () => ({ id: 'rps_01JAN2C3D4E5F6G7H8J9K0MNPS', url: 'https://openvibe.report/s/rps_1', created: false }) };
+ };
+ const client = createReportClient({
+ config: { networkInternalUrl: 'http://127.0.0.1:4000', reportInternalUrl: 'http://127.0.0.1:5080', oauth: { clientId: 'live', clientSecret: 'the-secret' } },
+ fetchImpl,
+ });
+ await client.sendStream(payload);
+ const token = calls.find((c) => c.url.includes('/oauth/token'));
+ assert.ok(token, 'Live asks Network for a token');
+ assert.ok(String(token.opts.body).includes('audience=openvibe.report'), 'for Report\'s audience');
+ assert.ok(String(token.opts.body).includes('scope=report.session.write'), 'with the report capability');
+ const post = calls.find((c) => c.url.includes('/internal/session-reports/stream'));
+ assert.ok(post, 'Live posts the stream to Report');
+ assert.strictEqual(post.opts.method, 'POST');
+ assert.strictEqual(post.opts.headers.Authorization, 'Bearer the-service-token');
+ const sent = JSON.parse(post.opts.body);
+ assert.strictEqual(sent.ags_id, 'ags_01JAN2C3D4E5F6G7H8J9K0MNPS', 'the stream part goes with it');
+
+ console.log(' ✓ Live adds the stream part to the session report');
+})().catch((e) => { console.error(e); process.exit(1); });