openvibers/openvibe.chat Commits

A6: Chat publishes the agent's step (agent.step.started|ended) (#375)

OP OpenVibers committed 7 days ago · 19c93c113c · 5 files, +158 −3

A task in a room, or in Vibe's own chat, is one step of the agent's timeline: its
generated title, its state and its page. One outbox row per task change, with the
person who asked as the subject; never the person's words, a message or a result.

Changes

ddiff --git a/server/events/agent-step.js b/server/events/agent-step.js
nnew file mode 100644
iindex 0000000..a94fe4d
---- /dev/null
++++ b/server/events/agent-step.js
@@@ -0,0 +1,73 @@
++'use strict';
++/**
++ * The agent's step (design 2026-11-07): a task in a Chat room (or in Vibe's own chat) is one step of the agent's
++ * timeline. Chat publishes agent.step.started|updated|ended with the person who asked as the subject, so the one
++ * timeline behind the Dock shows what the agent did for them. The payload is a title, an outcome and a link: never
++ * the task's title or details (the person's own words), never a message, a result or a source.
++ */
++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 STATUS_OF = Object.freeze({ queued: 'waiting', running: 'running', succeeded: 'done', failed: 'failed', cancelled: 'stopped' });
++const TITLE_OF_KIND = Object.freeze({
++ answer: 'Answering in your chat',
++ research: 'Researching for your chat',
++ web: 'Looking it up for your chat',
++ code: 'Coding from your chat',
++ computer: 'Working on your computer from your chat',
++ other: 'Working on your chat task',
++});
++
++/** What one task's current state is as an agent step: a generated title, an outcome and a link. */
++function agentStepOf(task, room, { status = null, baseUrl = 'https://openvibe.chat', startedAt = null, endedAt = null, costUsd = null, reason = null } = {}) {
++ const state = STATUS_OF[status] || status || STATUS_OF[task.status] || 'running';
++ const cost = costUsd == null ? task.cost_usd : costUsd;
++ const base = String(baseUrl || 'https://openvibe.chat').replace(/\/+$/, '');
++ const slug = room && room.slug ? String(room.slug) : null;
++ return {
++ step_id: String(task.id),
++ mode: 'talk',
++ product: 'chat',
++ ref: String(task.id),
++ title: TITLE_OF_KIND[task.kind] || TITLE_OF_KIND.other,
++ status: state,
++ started_at: new Date(startedAt == null ? Date.now() : startedAt).toISOString(),
++ ended_at: state === 'waiting' || state === 'running' ? null : new Date(endedAt == null ? Date.now() : endedAt).toISOString(),
++ placement: { where: 'In your chat', why: 'it runs from the conversation you are in, with the answer coming back here' },
++ cost_usd: typeof cost === 'number' && Number.isFinite(cost) && cost >= 0 ? cost : null,
++ link: slug ? `${base}/r/${encodeURIComponent(slug)}/tasks/${Number(task.id)}` : null,
++ reason: state === 'failed' || state === 'stopped' ? (reason == null ? null : String(reason).slice(0, 240)) : null,
++ };
++}
++
++/** The topic a current state is: a step starts, changes or ends. */
++function topicOf(status) {
++ const state = STATUS_OF[status] || status;
++ return state === 'waiting' || state === 'running' ? 'started' : 'ended';
++}
++
++/**
++ * Write one agent step row for the task's current state, in the caller's transaction (the same one as the change).
++ * A task with no person behind it (a service's) has no timeline and publishes nothing.
++ */
++async function emitAgentStep(task, room, { topic = null, status = null, costUsd = null, reason = null, baseUrl = 'https://openvibe.chat', startedAt = null, now = Date.now } = {}) {
++ if (!task || !USER_SUBJECT.test(String(task.requested_subject || ''))) return null;
++ const state = STATUS_OF[status] || status || STATUS_OF[task.status] || 'running';
++ const kind = topic || topicOf(state);
++ const event_type = AGENT_TOPICS[kind];
++ if (!event_type) return null;
++ const outbox = require('./service-outbox');
++ const ms = now();
++ return outbox.emit({
++ event_type,
++ subject: { type: 'user', id: task.requested_subject },
++ actorSubject: task.requested_subject,
++ priority: kind === 'ended' ? 'important' : 'low',
++ payload: agentStepOf(task, room, { status: state, costUsd, reason, baseUrl, startedAt, endedAt: ms }),
++ });
++}
++
++module.exports = { AGENT_TOPICS, agentStepOf, emitAgentStep, topicOf, STATUS_OF, TITLE_OF_KIND, USER_SUBJECT };
ddiff --git a/server/events/service-outbox.js b/server/events/service-outbox.js
iindex b8be3fa..83f1373 100644
---- a/server/events/service-outbox.js
++++ b/server/events/service-outbox.js
@@@ -13,6 +13,7 @@ const USER_SUBJECT = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
const EVENT_TYPES = [
'chat.message.created', 'chat.message.deleted', 'chat.dm.created',
'chat.moderation.action', 'chat.room.message.created', 'chat.room.message.deleted',
++ 'agent.step.started', 'agent.step.updated', 'agent.step.ended',
];
let instance = null;
@@@ -50,4 +51,4 @@ async function emit({ event_type, subject, payload, visibility = 'internal', pri
function reset() { instance = null; }
--module.exports = { configure, current, emit, reset, actorFor };
++module.exports = { configure, current, emit, reset, actorFor, EVENT_TYPES };
ddiff --git a/server/rooms/workspace.js b/server/rooms/workspace.js
iindex 0eeaf6a..1440889 100644
---- a/server/rooms/workspace.js
++++ b/server/rooms/workspace.js
@@@ -83,6 +83,18 @@ async function get(room, user, id) {
return shape(t, room);
}
++/** The agent's step for this task (design 2026-11-07): a title, an outcome and a link, never the person's words. */
++async function announceStep(room, task, { topic = null, status = null, ...rest } = {}) {
++ if (!task) return null;
++ try {
++ return await require('../events/agent-step').emitAgentStep(task, room, { baseUrl: config.web.baseUrl, topic, status, ...rest });
++ } catch (err) {
++ // A step that could not be queued never fails the task it describes; the outbox says why.
++ console.warn('[Workspace] agent step:', err && err.message);
++ return null;
++ }
++}
++
/** Tell every open page of the room that the queue changed (the panel refreshes itself). */
function changed(room, task) {
try { require('../chat/chat-server').broadcastToRoom(room.id, { type: 'room_task', room: room.slug, task: task ? { id: task.id, status: task.status } : null }); } catch { /* sockets are optional */ }
@@@ -105,6 +117,7 @@ async function add(room, user, { title, details = '', kind = 'research', via = '
const id = Number(r.lastInsertRowid);
if (QUICK.has(k)) await startOnActor(room, await row(room, id), deps);
const task = shape(await row(room, id), room);
++ await announceStep(room, await row(room, id), { status: task.status });
changed(room, task);
return task;
}
@@@ -126,6 +139,7 @@ async function cancel(room, user, id, deps = {}) {
if (t.actor_task) await actorCall('POST', `/api/v1/tasks/${encodeURIComponent(t.actor_task)}/cancel`, t.requested_subject, null, deps).catch(() => null);
await setStatus(room, t.id, { status: 'cancelled', finished_at: nowText(), error: `Cancelled by ${user.display_name || user.username || 'someone'}` });
const task = shape(await row(room, t.id), room);
++ await announceStep(room, await row(room, t.id), { status: 'cancelled', reason: task.error });
changed(room, task);
return task;
}
@@@ -162,6 +176,7 @@ async function claim(room, user, id, { runner = null } = {}) {
const r = await db.run("UPDATE room_tasks SET status = 'running', runner = ?, started_at = ov_now(), updated_at = ov_now() WHERE id = ? AND room_id = ? AND status = 'queued'", [who, t.id, room.id]);
if (!r.changes) fail(409, 'work.taken', 'Someone took it a moment ago');
const task = shape(await row(room, t.id), room);
++ await announceStep(room, await row(room, t.id), { status: 'running' });
changed(room, task);
return task;
}
@@@ -181,6 +196,7 @@ async function report(room, user, id, { status, result = '', result_url = null,
if (!url) fail(422, 'work.url', 'result_url is an https address');
}
await setStatus(room, t.id, { status, result: clean(result, LIMITS.result) || null, result_url: url, error: status === 'failed' ? (oneLine(error, LIMITS.error) || 'It did not work out') : null, finished_at: nowText() });
++ await announceStep(room, await row(room, t.id), { status, reason: status === 'failed' ? (oneLine(error, LIMITS.error) || 'It did not work out') : null });
await announce(room, await row(room, t.id), deps);
return shape(await row(room, t.id), room);
}
@@@ -227,8 +243,10 @@ async function startOnActor(room, t, deps = {}) {
const task = await actorCall('POST', '/api/v1/tasks', t.requested_subject, { task: actorText(t), idempotency_key: `chat-room-task-${t.id}` }, deps);
// A task a worker could take stays taken while Actor holds it, so nobody claims it twice.
await setStatus(room, t.id, { actor_task: String(task.id), runner: 'actor', status: task.state === 'running' || !QUICK.has(t.kind) ? 'running' : 'queued', started_at: nowText() });
++ await announceStep(room, await row(room, t.id), { status: task.state === 'running' || !QUICK.has(t.kind) ? 'running' : 'queued' });
} catch (err) {
await setStatus(room, t.id, { status: 'failed', runner: 'actor', error: oneLine(err.message, LIMITS.error), finished_at: nowText() });
++ await announceStep(room, await row(room, t.id), { status: 'failed', reason: oneLine(err.message, LIMITS.error) });
await announce(room, await row(room, t.id), deps).catch(() => {});
}
}
@@@ -241,11 +259,11 @@ async function poll(deps = {}) {
if (!room) continue;
let task;
try { task = await actorCall('GET', `/api/v1/tasks/${encodeURIComponent(t.actor_task)}`, t.requested_subject, null, deps); } catch (err) {
-- if (err.status === 404) { await setStatus(room, t.id, { status: 'failed', error: 'OpenVibe.Actor no longer has this task', finished_at: nowText() }); await announce(room, await row(room, t.id), deps).catch(() => {}); }
++ if (err.status === 404) { await setStatus(room, t.id, { status: 'failed', error: 'OpenVibe.Actor no longer has this task', finished_at: nowText() }); await announceStep(room, await row(room, t.id), { status: 'failed', reason: 'OpenVibe.Actor no longer has this task' }); await announce(room, await row(room, t.id), deps).catch(() => {}); }
continue;
}
const state = task && task.state;
-- if (state === 'running' && t.status !== 'running') { await setStatus(room, t.id, { status: 'running' }); changed(room, { id: Number(t.id), status: 'running' }); }
++ if (state === 'running' && t.status !== 'running') { await setStatus(room, t.id, { status: 'running' }); await announceStep(room, await row(room, t.id), { status: 'running' }); changed(room, { id: Number(t.id), status: 'running' }); }
// The agent handed its person a step on its desk (Desks plan D4): Vibe asks them here, once per hand-off.
if (state === 'running' && task.needs_you && task.needs_you.url) await askPerson(room, t, task.needs_you, deps).catch(() => {});
// The agent works on a desk of its own (Desks plan): in a Vibe chat, Vibe offers to watch it and take it over, once.
@@@ -261,6 +279,7 @@ async function poll(deps = {}) {
status: state, result: state === 'succeeded' ? clean(r.answer, LIMITS.result) || null : null, sources: JSON.stringify(sources), how,
error: state === 'succeeded' ? null : oneLine((task.error && (task.error.detail || task.error.code)) || state, LIMITS.error), cost_usd: cost, finished_at: nowText(),
});
++ await announceStep(room, await row(room, t.id), { status: state, costUsd: cost, reason: state === 'succeeded' ? null : oneLine((task.error && (task.error.detail || task.error.code)) || state, LIMITS.error) });
await announce(room, await row(room, t.id), deps).catch(() => {});
}
}
ddiff --git a/test/agent-step.test.js b/test/agent-step.test.js
nnew file mode 100644
iindex 0000000..390f62d
---- /dev/null
++++ b/test/agent-step.test.js
@@@ -0,0 +1,51 @@
++'use strict';
++/**
++ * The agent's step (design 2026-11-07): a task in a chat room is one step of the agent's timeline — waiting, running,
++ * done, failed or stopped. The payload is a generated title, an outcome and a link: never the task's title or
++ * details (the person's own words), never a message, a result or a source.
++ *
++ * node test/agent-step.test.js
++ */
++const assert = require('assert');
++const { AGENT_TOPICS, agentStepOf, topicOf, STATUS_OF } = require('../server/events/agent-step');
++const { EVENT_TYPES } = require('../server/events/service-outbox');
++
++const ROOM = { slug: 'staff' };
++const TASK = {
++ id: 42, title: 'Add a dark favicon to Tools', details: 'the person\'s own words, which never leave Chat',
++ kind: 'code', status: 'queued', requested_subject: 'usr_01JAB2C3D4E5F6G7H8J9K0MNPS',
++ result: 'the answer the agent wrote', error: null, cost_usd: null,
++};
++
++assert.strictEqual(AGENT_TOPICS.started, 'agent.step.started');
++assert.strictEqual(AGENT_TOPICS.updated, 'agent.step.updated');
++assert.strictEqual(AGENT_TOPICS.ended, 'agent.step.ended');
++assert.ok(EVENT_TYPES.includes('agent.step.started') && EVENT_TYPES.includes('agent.step.ended'), 'the outbox allows the agent topics');
++
++const waiting = agentStepOf(TASK, ROOM, { status: 'waiting', startedAt: 1762000000000, baseUrl: 'https://openvibe.chat' });
++assert.strictEqual(waiting.step_id, '42');
++assert.strictEqual(waiting.mode, 'talk');
++assert.strictEqual(waiting.product, 'chat');
++assert.strictEqual(waiting.status, 'waiting');
++assert.strictEqual(waiting.title, 'Coding from your chat');
++assert.strictEqual(waiting.link, 'https://openvibe.chat/r/staff/tasks/42');
++assert.strictEqual(waiting.ended_at, null);
++assert.ok(waiting.placement && waiting.placement.where && waiting.placement.why);
++const raw = JSON.stringify(waiting);
++assert.ok(!raw.includes('Add a dark favicon to Tools'), 'the task title never travels');
++assert.ok(!raw.includes('the person\'s own words'), 'the details never travel');
++assert.ok(!('title_text' in waiting) && !('details' in waiting) && !('result' in waiting) && !('messages' in waiting), 'no field for a person\'s content');
++
++const done = agentStepOf({ ...TASK, status: 'succeeded', cost_usd: 0.04 }, ROOM, { endedAt: 1762000060000 });
++assert.strictEqual(done.status, 'done');
++assert.strictEqual(done.cost_usd, 0.04);
++assert.ok(done.ended_at, 'an ended step says when it ended');
++assert.strictEqual(agentStepOf({ ...TASK, status: 'failed' }, ROOM, { reason: 'the build failed' }).reason, 'the build failed');
++
++assert.strictEqual(STATUS_OF.cancelled, 'stopped');
++assert.strictEqual(topicOf('queued'), 'started');
++assert.strictEqual(topicOf('running'), 'started');
++assert.strictEqual(topicOf('succeeded'), 'ended');
++assert.strictEqual(topicOf('cancelled'), 'ended');
++console.log('chat agent step: a generated title, an outcome and a link — never the person\'s words');
++console.log('all passed');
ddiff --git a/test/workspace.test.js b/test/workspace.test.js
iindex 18f9b4a..c380f56 100644
---- a/test/workspace.test.js
++++ b/test/workspace.test.js
@@@ -109,6 +109,17 @@ t('code work waits for a worker: claim, report, the room hears it', async () =>
assert.strictEqual(r.body.task.status, 'succeeded');
const msgs = (await api('GET', '/rooms/staff/messages', staff)).body.messages;
assert.match(msgs[msgs.length - 1].message, /^✅ Done: "Add a dark favicon to Tools"[\s\S]*https:\/\/github\.com\/OpenVibers\/OpenVibe\.Tools\/pull\/1/);
++ // The agent's step (design 2026-11-07): claim and report are its two changes, one row each. The payload is a
++ // generated title, an outcome and a link: never the task's title (the person's words) or its result.
++ const steps = (await h.db.getDb().prepare('SELECT envelope FROM service_outbox ORDER BY id').all())
++ .map((r) => typeof r.envelope === 'string' ? JSON.parse(r.envelope) : r.envelope)
++ .filter((e) => String(e.event_type || '').startsWith('agent.step.') && e.payload && e.payload.step_id === String(id));
++ assert.deepStrictEqual(steps.map((e) => e.event_type), ['agent.step.started', 'agent.step.started', 'agent.step.ended']);
++ assert.deepStrictEqual(steps.map((e) => e.payload.status), ['waiting', 'running', 'done']);
++ assert.deepStrictEqual(steps[0].subject, { type: 'user', id: staff.subject_id }, 'the person who asked is the step\'s subject, not whoever claimed it');
++ const raw = JSON.stringify(steps);
++ assert.ok(!raw.includes('Add a dark favicon to Tools'), 'the person\'s words never travel');
++ assert.ok(!raw.includes('Added favicon-dark.svg'), 'the result never travels');
});
t('run on my computer: the asker hands their code task to Actor, which runs it on their own machine', async () => {