openvibers/openvibe.live Commits
A6: Live publishes the agent's step (agent.step.started|ended) (#660)
Changes
nnew file mode 100644@@@ -0,0 +1,70 @@++'use strict';++/**++ * The agent's step (design 2026-11-07): an agent stream is one step of the agent's timeline. Live publishes++ * agent.step.started when the desk starts publishing and agent.step.ended when the stream ends, with the owner's++ * subject, so the one timeline behind the Dock shows it. The payload is a title, an outcome and a link: never the++ * task, a frame, a chat line or a voice line.++ */++const AGENT_TOPICS = Object.freeze({++ started: 'agent.step.started',++ updated: 'agent.step.updated',++ ended: 'agent.step.ended',++});++const USER_SUBJECT = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;++++const iso = (v) => {++ if (v == null || v === '') return null;++ const d = v instanceof Date || typeof v === 'number' ? new Date(v) : new Date(String(v).includes('T') ? v : `${String(v).replace(' ', 'T')}Z`);++ return Number.isNaN(d.getTime()) ? null : d.toISOString();++};++++/** The stream's public page on Live (its own address); the step carries no unlisted link's secret. */++function pageUrl(row, baseUrl = 'https://openvibe.live') {++ const base = String(baseUrl || 'https://openvibe.live').replace(/\/+$/, '');++ return `${base}/@${encodeURIComponent(row.username || '')}/agents/${encodeURIComponent(row.public_id)}`;++}++++/** One agent stream → the agent's step: its title, its state and its page. Never the task or a frame. */++function agentStepOf(row, { status = 'running', startedAt = null, endedAt = null, reason = null, baseUrl = null } = {}) {++ const title = String(row.title || '').trim().slice(0, 120) || 'an agent stream';++ return {++ step_id: row.public_id,++ mode: 'stream',++ product: 'live',++ ref: row.public_id,++ title: `Streaming ${title}`,++ status,++ started_at: iso(startedAt) || iso(row.started_at) || new Date().toISOString(),++ ended_at: status === 'running' ? null : (iso(endedAt) || iso(row.ended_at)),++ placement: { where: 'On your desk', why: 'it runs from your own machine, and only you can end it' },++ cost_usd: null,++ link: pageUrl(row, baseUrl),++ reason: status === 'stopped' || status === 'failed' ? (reason == null ? null : String(reason).slice(0, 240)) : null,++ };++}++++/** Which step a state change is: the stream going live starts it, the stream ending stops it. A pause is not a step. */++function stepChangeOf(row, set, event = null) {++ if (!row || !set) return null;++ if (set.state === 'live' && row.state !== 'live') return { topic: 'started', status: 'running', startedAt: set.started_at || new Date() };++ if (set.state === 'ended' && row.state !== 'ended') return { topic: 'ended', status: 'stopped', endedAt: set.ended_at || new Date(), reason: (event && event.extra && event.extra.reason) || null };++ return null;++}++++/** The agent.step.* envelope for one stream change, or null when the owner has no person subject to show it to. */++function agentStepEnvelope(row, { topic, ...rest } = {}, { source = 'live' } = {}) {++ if (!row || !USER_SUBJECT.test(String(row.owner_subject || ''))) return null;++ const event_type = AGENT_TOPICS[topic];++ if (!event_type) return null;++ return {++ event_type,++ actor: { type: 'user', id: row.owner_subject },++ subject: { type: 'user', id: row.owner_subject },++ visibility: 'internal',++ priority: topic === 'ended' ? 'important' : 'low',++ payload: agentStepOf(row, rest),++ source,++ };++}++++module.exports = { AGENT_TOPICS, agentStepOf, agentStepEnvelope, stepChangeOf, pageUrl, USER_SUBJECT };@@@ -21,6 +21,7 @@ const db = require('../db/database');const contracts = require('openvibe-contracts');const identity = require('../auth/identity-sync');const sharing = require('./sharing');++const { agentStepEnvelope, stepChangeOf } = require('./agent-step');const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;const AGENT_ID_RE = /^ags_[0-9A-HJKMNP-TV-Z]{26}$/;@@@ -286,15 +287,21 @@ async function update(row, { set = {}, event = null }) {priority: event.kind === 'started' || event.kind === 'ended' ? 'important' : 'normal',payload: { stream_id: row.public_id, kind: event.kind, at: new Date().toISOString(), ...(event.extra || {}) },} : null;++ // The agent's step (design 2026-11-07): the stream starts as a running step and ends as a stopped one. A pause,++ // a curtain or a mute is not a step; the stream is the one step, and a mode change is the next stream's.++ const stepChange = stepChangeOf(row, set, event);++ const stepEnv = stepChange ? agentStepEnvelope(row, { ...stepChange, baseUrl: config.baseUrl }) : null;let outbox;++ let stepOutbox;await db.getDb().tx(async () => {await db.run(`UPDATE agent_streams SET ${names.map((n) => `${n} = ?`).join(', ')}, updated_at = now() WHERE managed_stream_id = ?`,[...names.map((n) => set[n]), row.managed_stream_id]);if (env) outbox = await require('../events/stream-events').enqueue(env);++ if (stepEnv) stepOutbox = await require('../events/stream-events').enqueue(stepEnv);});-- if (outbox) require('../events/stream-events').kick();++ if (outbox || stepOutbox) require('../events/stream-events').kick();return await getByManagedStreamId(row.managed_stream_id);}nnew file mode 100644@@@ -0,0 +1,63 @@++'use strict';++/**++ * The agent's step (design 2026-11-07): an agent stream is one step of the agent's timeline — running while the desk++ * publishes, stopped when it ends. The payload is a title, an outcome and a link: never the task, a frame, a chat++ * line or a voice line.++ *++ * node test/agent-step.test.js++ */++const assert = require('assert');++const { AGENT_TOPICS, agentStepOf, agentStepEnvelope, stepChangeOf, pageUrl } = require('../server/agent-streams/agent-step');++++const ROW = {++ public_id: 'ags_01JAB2C3D4E5F6G7H8J9K0MNPS',++ username: 'ada',++ title: 'Fixing the checkout tests',++ owner_subject: 'usr_01JAB2C3D4E5F6G7H8J9K0MNPS',++ state: 'idle',++ started_at: null,++ task: { prompt: 'the desk task, which never leaves Live' },++ latest_step_title: 'running the tests',++};++++assert.strictEqual(AGENT_TOPICS.started, 'agent.step.started');++assert.strictEqual(AGENT_TOPICS.ended, 'agent.step.ended');++++const running = agentStepOf(ROW, { status: 'running', startedAt: 1762000000000, baseUrl: 'https://openvibe.live' });++assert.strictEqual(running.step_id, ROW.public_id);++assert.strictEqual(running.mode, 'stream');++assert.strictEqual(running.product, 'live');++assert.strictEqual(running.status, 'running');++assert.strictEqual(running.title, 'Streaming Fixing the checkout tests');++assert.strictEqual(running.link, 'https://openvibe.live/@ada/agents/ags_01JAB2C3D4E5F6G7H8J9K0MNPS');++assert.strictEqual(running.ended_at, null);++assert.ok(running.placement && running.placement.where && running.placement.why);++const raw = JSON.stringify(running);++assert.ok(!raw.includes('the desk task'), 'the task never travels');++assert.ok(!raw.includes('running the tests'), 'the desk steps never travel');++assert.ok(!('task' in running) && !('frames' in running) && !('chat' in running) && !('voice' in running), 'no field for the content');++++const ended = agentStepOf(ROW, { status: 'stopped', startedAt: 1762000000000, endedAt: 1762000060000, reason: null });++assert.strictEqual(ended.status, 'stopped');++assert.ok(ended.ended_at, 'an ended stream says when it ended');++assert.strictEqual(agentStepOf(ROW, { status: 'stopped', reason: 'you ended it' }).reason, 'you ended it');++++const start = stepChangeOf(ROW, { state: 'live' });++assert.strictEqual(start.topic, 'started');++assert.strictEqual(start.status, 'running');++assert.ok(start.startedAt, 'the step says when the stream started');++assert.strictEqual(stepChangeOf({ ...ROW, state: 'live' }, { state: 'live' }), null, 'a pause of a live stream is not a new step');++const stop = stepChangeOf({ ...ROW, state: 'live' }, { state: 'ended' }, { kind: 'ended', extra: { reason: 'ended' } });++assert.strictEqual(stop.topic, 'ended');++assert.strictEqual(stop.status, 'stopped');++assert.strictEqual(stop.reason, 'ended');++assert.ok(stop.endedAt, 'the step says when the stream ended');++++const env = agentStepEnvelope(ROW, { topic: 'started', status: 'running', startedAt: 1762000000000 });++assert.strictEqual(env.event_type, 'agent.step.started');++assert.deepStrictEqual(env.subject, { type: 'user', id: ROW.owner_subject });++assert.strictEqual(env.visibility, 'internal');++assert.strictEqual(agentStepEnvelope({ ...ROW, owner_subject: null }, { topic: 'started', status: 'running' }), null, 'no subject, no step');++assert.strictEqual(pageUrl(ROW), 'https://openvibe.live/@ada/agents/' + ROW.public_id);++console.log('live agent step: one stream, one running step — never the task, a frame or a chat line');++console.log('all passed');
new file mode 100644
new file mode 100644
@@ -0,0 +1,70 @@
@@ -0,0 +1,70 @@
+'use strict';
+/**
+ * The agent's step (design 2026-11-07): an agent stream is one step of the agent's timeline. Live publishes
+ * agent.step.started when the desk starts publishing and agent.step.ended when the stream ends, with the owner's
+ * subject, so the one timeline behind the Dock shows it. The payload is a title, an outcome and a link: never the
+ * task, a frame, a chat line or a voice line.
+ */
+const AGENT_TOPICS = Object.freeze({
+ started: 'agent.step.started',
+ updated: 'agent.step.updated',
+ ended: 'agent.step.ended',
+});
+const USER_SUBJECT = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
+
+const iso = (v) => {
+ if (v == null || v === '') return null;
+ const d = v instanceof Date || typeof v === 'number' ? new Date(v) : new Date(String(v).includes('T') ? v : `${String(v).replace(' ', 'T')}Z`);
+ return Number.isNaN(d.getTime()) ? null : d.toISOString();
+};
+
+/** The stream's public page on Live (its own address); the step carries no unlisted link's secret. */
+function pageUrl(row, baseUrl = 'https://openvibe.live') {
+ const base = String(baseUrl || 'https://openvibe.live').replace(/\/+$/, '');
+ return `${base}/@${encodeURIComponent(row.username || '')}/agents/${encodeURIComponent(row.public_id)}`;
+}
+
+/** One agent stream → the agent's step: its title, its state and its page. Never the task or a frame. */
+function agentStepOf(row, { status = 'running', startedAt = null, endedAt = null, reason = null, baseUrl = null } = {}) {
+ const title = String(row.title || '').trim().slice(0, 120) || 'an agent stream';
+ return {
+ step_id: row.public_id,
+ mode: 'stream',
+ product: 'live',
+ ref: row.public_id,
+ title: `Streaming ${title}`,
+ status,
+ started_at: iso(startedAt) || iso(row.started_at) || new Date().toISOString(),
+ ended_at: status === 'running' ? null : (iso(endedAt) || iso(row.ended_at)),
+ placement: { where: 'On your desk', why: 'it runs from your own machine, and only you can end it' },
+ cost_usd: null,
+ link: pageUrl(row, baseUrl),
+ reason: status === 'stopped' || status === 'failed' ? (reason == null ? null : String(reason).slice(0, 240)) : null,
+ };
+}
+
+/** Which step a state change is: the stream going live starts it, the stream ending stops it. A pause is not a step. */
+function stepChangeOf(row, set, event = null) {
+ if (!row || !set) return null;
+ if (set.state === 'live' && row.state !== 'live') return { topic: 'started', status: 'running', startedAt: set.started_at || new Date() };
+ if (set.state === 'ended' && row.state !== 'ended') return { topic: 'ended', status: 'stopped', endedAt: set.ended_at || new Date(), reason: (event && event.extra && event.extra.reason) || null };
+ return null;
+}
+
+/** The agent.step.* envelope for one stream change, or null when the owner has no person subject to show it to. */
+function agentStepEnvelope(row, { topic, ...rest } = {}, { source = 'live' } = {}) {
+ if (!row || !USER_SUBJECT.test(String(row.owner_subject || ''))) return null;
+ const event_type = AGENT_TOPICS[topic];
+ if (!event_type) return null;
+ return {
+ event_type,
+ actor: { type: 'user', id: row.owner_subject },
+ subject: { type: 'user', id: row.owner_subject },
+ visibility: 'internal',
+ priority: topic === 'ended' ? 'important' : 'low',
+ payload: agentStepOf(row, rest),
+ source,
+ };
+}
+
+module.exports = { AGENT_TOPICS, agentStepOf, agentStepEnvelope, stepChangeOf, pageUrl, USER_SUBJECT };
@@ -21,6 +21,7 @@ const db = require('../db/database');
@@ -21,6 +21,7 @@ const db = require('../db/database');
const contracts = require('openvibe-contracts');
const contracts = require('openvibe-contracts');
const identity = require('../auth/identity-sync');
const identity = require('../auth/identity-sync');
const sharing = require('./sharing');
const sharing = require('./sharing');
+const { agentStepEnvelope, stepChangeOf } = require('./agent-step');
const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
const AGENT_ID_RE = /^ags_[0-9A-HJKMNP-TV-Z]{26}$/;
const AGENT_ID_RE = /^ags_[0-9A-HJKMNP-TV-Z]{26}$/;
@@ -286,15 +287,21 @@ async function update(row, { set = {}, event = null }) {
@@ -286,15 +287,21 @@ async function update(row, { set = {}, event = null }) {
priority: event.kind === 'started' || event.kind === 'ended' ? 'important' : 'normal',
priority: event.kind === 'started' || event.kind === 'ended' ? 'important' : 'normal',
payload: { stream_id: row.public_id, kind: event.kind, at: new Date().toISOString(), ...(event.extra || {}) },
payload: { stream_id: row.public_id, kind: event.kind, at: new Date().toISOString(), ...(event.extra || {}) },
} : null;
} : null;
+ // The agent's step (design 2026-11-07): the stream starts as a running step and ends as a stopped one. A pause,
+ // a curtain or a mute is not a step; the stream is the one step, and a mode change is the next stream's.
+ const stepChange = stepChangeOf(row, set, event);
+ const stepEnv = stepChange ? agentStepEnvelope(row, { ...stepChange, baseUrl: config.baseUrl }) : null;
let outbox;
let outbox;
+ let stepOutbox;
await db.getDb().tx(async () => {
await db.getDb().tx(async () => {
await db.run(
await db.run(
`UPDATE agent_streams SET ${names.map((n) => `${n} = ?`).join(', ')}, updated_at = now() WHERE managed_stream_id = ?`,
`UPDATE agent_streams SET ${names.map((n) => `${n} = ?`).join(', ')}, updated_at = now() WHERE managed_stream_id = ?`,
[...names.map((n) => set[n]), row.managed_stream_id]
[...names.map((n) => set[n]), row.managed_stream_id]
);
);
if (env) outbox = await require('../events/stream-events').enqueue(env);
if (env) outbox = await require('../events/stream-events').enqueue(env);
+ if (stepEnv) stepOutbox = await require('../events/stream-events').enqueue(stepEnv);
});
});
- if (outbox) require('../events/stream-events').kick();
+ if (outbox || stepOutbox) require('../events/stream-events').kick();
return await getByManagedStreamId(row.managed_stream_id);
return await getByManagedStreamId(row.managed_stream_id);
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,63 @@
@@ -0,0 +1,63 @@
+'use strict';
+/**
+ * The agent's step (design 2026-11-07): an agent stream is one step of the agent's timeline — running while the desk
+ * publishes, stopped when it ends. The payload is a title, an outcome and a link: never the task, a frame, a chat
+ * line or a voice line.
+ *
+ * node test/agent-step.test.js
+ */
+const assert = require('assert');
+const { AGENT_TOPICS, agentStepOf, agentStepEnvelope, stepChangeOf, pageUrl } = require('../server/agent-streams/agent-step');
+
+const ROW = {
+ public_id: 'ags_01JAB2C3D4E5F6G7H8J9K0MNPS',
+ username: 'ada',
+ title: 'Fixing the checkout tests',
+ owner_subject: 'usr_01JAB2C3D4E5F6G7H8J9K0MNPS',
+ state: 'idle',
+ started_at: null,
+ task: { prompt: 'the desk task, which never leaves Live' },
+ latest_step_title: 'running the tests',
+};
+
+assert.strictEqual(AGENT_TOPICS.started, 'agent.step.started');
+assert.strictEqual(AGENT_TOPICS.ended, 'agent.step.ended');
+
+const running = agentStepOf(ROW, { status: 'running', startedAt: 1762000000000, baseUrl: 'https://openvibe.live' });
+assert.strictEqual(running.step_id, ROW.public_id);
+assert.strictEqual(running.mode, 'stream');
+assert.strictEqual(running.product, 'live');
+assert.strictEqual(running.status, 'running');
+assert.strictEqual(running.title, 'Streaming Fixing the checkout tests');
+assert.strictEqual(running.link, 'https://openvibe.live/@ada/agents/ags_01JAB2C3D4E5F6G7H8J9K0MNPS');
+assert.strictEqual(running.ended_at, null);
+assert.ok(running.placement && running.placement.where && running.placement.why);
+const raw = JSON.stringify(running);
+assert.ok(!raw.includes('the desk task'), 'the task never travels');
+assert.ok(!raw.includes('running the tests'), 'the desk steps never travel');
+assert.ok(!('task' in running) && !('frames' in running) && !('chat' in running) && !('voice' in running), 'no field for the content');
+
+const ended = agentStepOf(ROW, { status: 'stopped', startedAt: 1762000000000, endedAt: 1762000060000, reason: null });
+assert.strictEqual(ended.status, 'stopped');
+assert.ok(ended.ended_at, 'an ended stream says when it ended');
+assert.strictEqual(agentStepOf(ROW, { status: 'stopped', reason: 'you ended it' }).reason, 'you ended it');
+
+const start = stepChangeOf(ROW, { state: 'live' });
+assert.strictEqual(start.topic, 'started');
+assert.strictEqual(start.status, 'running');
+assert.ok(start.startedAt, 'the step says when the stream started');
+assert.strictEqual(stepChangeOf({ ...ROW, state: 'live' }, { state: 'live' }), null, 'a pause of a live stream is not a new step');
+const stop = stepChangeOf({ ...ROW, state: 'live' }, { state: 'ended' }, { kind: 'ended', extra: { reason: 'ended' } });
+assert.strictEqual(stop.topic, 'ended');
+assert.strictEqual(stop.status, 'stopped');
+assert.strictEqual(stop.reason, 'ended');
+assert.ok(stop.endedAt, 'the step says when the stream ended');
+
+const env = agentStepEnvelope(ROW, { topic: 'started', status: 'running', startedAt: 1762000000000 });
+assert.strictEqual(env.event_type, 'agent.step.started');
+assert.deepStrictEqual(env.subject, { type: 'user', id: ROW.owner_subject });
+assert.strictEqual(env.visibility, 'internal');
+assert.strictEqual(agentStepEnvelope({ ...ROW, owner_subject: null }, { topic: 'started', status: 'running' }), null, 'no subject, no step');
+assert.strictEqual(pageUrl(ROW), 'https://openvibe.live/@ada/agents/' + ROW.public_id);
+console.log('live agent step: one stream, one running step — never the task, a frame or a chat line');
+console.log('all passed');