openvibers/openvibe.chat Commits
Stop while Vibe writes: what it wrote so far stays as its answer, marked as stopped, and the run is cancelled at AI (POST /api/chat/vibe/:slug/stop; a Stop button in the writing bubble) (#186)
Changes
@@@ -98,6 +98,11 @@ body.oc-app-body { overflow: hidden; }.vb-scroll > .oc-notice { max-width: var(--vb-col); margin: 8px auto; }.vb-thread { list-style: none; margin: 0 auto; padding: 8px 0 0; max-width: var(--vb-col); display: grid; grid-template-columns: minmax(0, 1fr); gap: 22px; }.vb-turn { display: flex; gap: 14px; min-width: 0; }++/* The writing bubble's Stop (vibe.js showWait): at the bubble's end, quiet like the actions under answers. */++.vb-stop { align-self: flex-start; margin-left: auto; padding: 4px 10px; border-radius: 8px; border: 1px solid var(--oc-line); background: transparent; color: var(--oc-muted); font: inherit; font-size: .8rem; cursor: pointer; }++.vb-stop:hover { color: var(--oc-text); background: var(--oc-panel-2); }++.vb-stop:focus-visible { outline: 2px solid var(--oc-accent); outline-offset: 1px; }++.vb-stop[disabled] { opacity: .5; cursor: default; }.vb-me { justify-content: flex-end; }.vb-bubble {max-width: min(85%, 620px); padding: 10px 16px; border-radius: 20px 20px 6px 20px;@@@ -200,6 +200,17 @@det.appendChild(ul); body.appendChild(det);}function clearWait() { var w = document.getElementById('vb-wait'); if (w) w.remove(); if (waitTimer) { clearTimeout(waitTimer); waitTimer = null; } }++ // Stop: what Vibe wrote so far stays as its answer (POST /api/chat/vibe/:slug/stop); shown once the chat exists (a new++ // chat's first bubble is drawn before it does, so the first words add it).++ function addStop(li) {++ if (!room || !li || li.querySelector('.vb-stop')) return;++ var stop = el('button', 'vb-act vb-stop', 'Stop'); stop.type = 'button';++ stop.addEventListener('click', function () {++ stop.disabled = true;++ fetch('/api/chat/vibe/' + encodeURIComponent(room) + '/stop', { method: 'POST', credentials: 'same-origin' }).catch(function () { stop.disabled = false; });++ });++ li.appendChild(stop);++ }function showWait() {if (document.getElementById('vb-wait')) return;var li = el('li', 'vb-turn vb-ai vb-wait'); li.id = 'vb-wait';@@@ -207,6 +218,7 @@var body = el('div', 'vb-body'); var dots = el('span', 'vb-dots'); dots.setAttribute('aria-label', 'Vibe is writing');dots.appendChild(el('i')); dots.appendChild(el('i')); dots.appendChild(el('i'));body.appendChild(dots); li.appendChild(body);++ addStop(li);thread().appendChild(li);// An answer takes up to a minute with tools; after two the dots give way to a way forward.waitTimer = setTimeout(function () { clearWait(); notice('Vibe did not answer this time. Send your message again, or try again in a minute.'); }, 120000);@@@ -216,6 +228,7 @@function streamText(text) {showWait();var w = document.getElementById('vb-wait'); if (!w) return;++ addStop(w);var stick = nearBottom();var body = w.querySelector('.vb-body');var md = body.querySelector('.vb-md');@@@ -85,13 +85,17 @@ async function stream(workflow, input, opts = {}, onText = () => {}) {try {if (!_tokens) ai();const token = await _tokens.getToken();++ const timeout = AbortSignal.timeout(config.ai.waitMs + 10000);const res = await (opts.fetchImpl || globalThis.fetch)(`${config.ai.internalUrl}/api/v1/runs?stream=1&wait=${Math.max(0, Math.floor(config.ai.waitMs))}`, {method: 'POST',headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json', Accept: 'text/event-stream', ...(opts.idempotencyKey ? { 'Idempotency-Key': opts.idempotencyKey } : {}) },body: JSON.stringify({ workflow, input, ...(opts.credential ? { credential: opts.credential } : {}), ...(opts.attribution !== undefined ? { attribution: opts.attribution } : {}) }),-- signal: AbortSignal.timeout(config.ai.waitMs + 10000),++ signal: opts.signal ? AbortSignal.any([opts.signal, timeout]) : timeout,});if (!res.ok) { console.warn(`[AI] ${workflow}: AI answered ${res.status}`); return null; }++ // Stopped by the person (opts.signal): the run is cancelled at AI, so it spends nothing more.++ const runId = res.headers.get('x-openvibe-run-id');++ if (opts.signal && runId) opts.signal.addEventListener('abort', () => { try { const { runs } = ai(); if (runs && runs.cancel) runs.cancel(runId).catch(() => {}); } catch { /* AI away */ } }, { once: true });let run = null;if (!/event-stream/.test(String(res.headers.get('content-type') || ''))) {const body = await res.json();@@@ -119,7 +123,7 @@ async function stream(workflow, input, opts = {}, onText = () => {}) {if (!run || run.status !== 'succeeded' || run.synthetic) return null;return run.output || null;} catch (err) {-- console.warn(`[AI] ${workflow}: ${err && err.message}`);++ if (!(opts.signal && opts.signal.aborted)) console.warn(`[AI] ${workflow}: ${err && err.message}`);return null;}}@@@ -45,6 +45,7 @@ const PER_PERSON_HOUR_VIBE = 60;const MAX_TOOL_ROUNDS = 3;const MAX_CALLS_PER_ROUND = 4;const inFlight = new Set(); // room ids with an answer being written++const stoppers = new Map(); // a Vibe chat's room id → the AbortController of the answer being written (stop())const personHour = new Map(); // `${room}:${user}` → [timestamps]const handleOf = (name) => String(name || '').toLowerCase().normalize('NFKD').replace(/[^\p{L}\p{N}]+/gu, '').slice(0, 24);@@@ -283,7 +284,7 @@ function userPrompt({ member, context, transcript, askedBy, text }) {* The answer through chat.member.turn with tools: up to MAX_TOOL_ROUNDS rounds of tool calls run here, then a turn* with tool_choice none. → the reply text, or null when a turn failed (the caller falls back to chat.member.reply).*/--async function answerWithTools({ room, member, context, transcript, askedBy, text, opts, subject = null, images = [], user = null, messageId = null, memory = null, called = new Set(), onText = null, steps = null }) {++async function answerWithTools({ room, member, context, transcript, askedBy, text, opts, subject = null, images = [], user = null, messageId = null, memory = null, called = new Set(), onText = null, steps = null, signal = null }) {const tools = aiTools.definitions(member, room, { memory: memory !== null });const vibe = Number(room.assistant) === 1;const prompt = userPrompt({ member, context, transcript, askedBy, text });@@@ -293,9 +294,10 @@ async function answerWithTools({ room, member, context, transcript, askedBy, tex{ role: 'user', content: images.length ? [{ type: 'text', text: `${prompt}\n\n(${images.length === 1 ? 'The picture they sent is' : 'The pictures they sent are'} attached.)` }, ...images.map((url) => ({ type: 'image_url', image_url: { url } }))] : prompt },];for (let round = 0; round <= MAX_TOOL_ROUNDS; round++) {++ if (signal && signal.aborted) return null; // stopped (stop()): the caller keeps what was writtenconst last = round === MAX_TOOL_ROUNDS;const turn = { messages, tools, tool_choice: last ? 'none' : 'auto', max_tokens: vibe ? 1500 : 600, temperature: 0.7 };-- const turnOpts = { ...opts, idempotencyKey: `${opts.idempotencyKey}-t${round}` };++ const turnOpts = { ...opts, idempotencyKey: `${opts.idempotencyKey}-t${round}`, ...(signal ? { signal } : {}) };// Vibe's answer streams (each round's text so far, to the chat as it is written); other members answer whole.const out = onText ? await aiClient.stream('chat.member.turn', turn, turnOpts, onText) : await aiClient.structured('chat.member.turn', turn, turnOpts);const msg = out && Array.isArray(out.choices) && out.choices[0] && out.choices[0].message;@@@ -369,9 +371,17 @@ async function afterPost(room, message, user, deps = {}) {lastDelta = now;broadcast({ type: 'room_ai_delta', room: room.slug, name: member.name, text: String(pending).slice(0, 8000) });} : null;++ // Stop (stop(), the chat's Stop button): the answer so far is kept, marked as stopped, and the run is cancelled.++ const controller = vibe ? new AbortController() : null;++ if (controller) stoppers.set(room.id, controller);if (config.ai.memberTools) {const images = await picturesFor(message, history, user).catch(() => []);-- try { text = clean(await answerWithTools({ room, member, context, transcript, askedBy, text: String(message.message).slice(0, vibe ? 12000 : 2000), opts, subject: user.subject_id || null, images, user, messageId: message.id, memory, called, onText, steps }) || '', vibe ? 8000 : 2000); } catch { text = ''; }++ try { text = clean(await answerWithTools({ room, member, context, transcript, askedBy, text: String(message.message).slice(0, vibe ? 12000 : 2000), opts, subject: user.subject_id || null, images, user, messageId: message.id, memory, called, onText, steps, signal: controller && controller.signal }) || '', vibe ? 8000 : 2000); } catch { text = ''; }++ }++ if (controller && controller.signal.aborted) {++ const sofar = clean(pending || '', 7900);++ if (sofar && !own) await require('./vibe').countReply(user);++ return await say(sofar ? `${sofar}\n\n*Stopped.*` : 'Stopped before I wrote anything.', steps && steps.steps.length ? steps : null);}// A plain "remember that …" is kept even when the model only promised it (small models often do).if (memory !== null && text && !called.has('remember')) {@@@ -395,9 +405,18 @@ async function afterPost(room, message, user, deps = {}) {return null;} finally {inFlight.delete(room.id);++ stoppers.delete(room.id);}}++/** Stop the answer being written in a Vibe chat → true when one was. */++function stop(roomId) {++ const c = stoppers.get(Number(roomId)) || stoppers.get(roomId);++ if (!c || c.signal.aborted) return false;++ c.abort();++ return true;++}++/** What the room shows a member thinks with: "OpenVibe", or the own key's provider, model and whose key it is. */const modelOf = (m) => (m.engine === 'own' ? (m.engine_label || 'someone\'s own key') : 'OpenVibe');@@@ -407,6 +426,6 @@ const toWire = (m) => ({engine: m.engine || 'openvibe', model: modelOf(m), created_at: m.created_at, updated_at: m.updated_at,});--function _reset() { inFlight.clear(); personHour.clear(); }++function _reset() { inFlight.clear(); personHour.clear(); stoppers.clear(); }--module.exports = { MAX_PER_ROOM, LIMITS, AiMemberError, enabled, list, byId, add, update, remove, dropKey, relabelKey, modelOf, mentioned, handleOf, afterPost, transcriptOf, vibePrompt, toWire, _reset };++module.exports = { MAX_PER_ROOM, LIMITS, AiMemberError, enabled, list, byId, add, update, remove, dropKey, relabelKey, modelOf, mentioned, handleOf, afterPost, stop, transcriptOf, vibePrompt, toWire, _reset };@@@ -14,6 +14,7 @@* DELETE /:slug 204: the chat and its messages are deleted* POST /:slug/retry 202 { removed }: Vibe's last answer goes and it answers the person's last message anew (409* when the last two messages are not that exchange)++ * POST /:slug/stop 202 { stopped }: the answer being written stops; what was written stays, marked as stopped* GET /memory { on, notes: [{ id, text, source, created_at }], max }: what Vibe remembers about the person* (server/rooms/vibe-memory.js)* POST /memory { text } → 201 { note } (a note of their own), 422 when it is empty, too long, secret-looking,@@@ -136,6 +137,14 @@ router.post('/:slug/retry', requireAuth, personOnly, startLimit, async (req, res} catch (err) { send(res, err); }});++router.post('/:slug/stop', requireAuth, personOnly, manageLimit, async (req, res) => {++ try {++ const room = await vibe.ownChat(req.params.slug, req.user);++ if (!room) return res.status(404).json({ error: 'No such chat', code: 'rooms.not_found' });++ res.status(202).json({ stopped: require('./ai-members').stop(room.id) });++ } catch (err) { send(res, err); }++});++router.patch('/:slug', requireAuth, personOnly, manageLimit, async (req, res) => {try {const room = await vibe.ownChat(req.params.slug, req.user);nnew file mode 100644@@@ -0,0 +1,76 @@++'use strict';++/**++ * Stop while Vibe writes (server/rooms/ai-members.js stop, vibe-routes.js POST /:slug/stop, server/ai/client.js stream):++ * the answer so far stays as Vibe's message, marked as stopped, and the run is cancelled at AI; with nothing written,++ * the message says so; a chat with nothing being written answers stopped: false; another person's chat is not there.++ */++const assert = require('assert');++const http = require('http');++const { ids } = require('openvibe-contracts');++const { boot, suite } = require('./helpers');++++const t = suite('vibe stop');++let h, ann, bob, ai, aiClient;++const cancels = [];++let wordsBeforeFirst = 0; // how long the fake AI "thinks" before its first word, in 150 ms steps++const pause = (ms) => new Promise((r) => setTimeout(r, ms));++const vibeApi = (method, path, u, body) => h.http(method, `/api/chat/vibe${path}`, { token: u.token, body });++async function lastAi(slug) {++ for (let i = 0; i < 120; i++) {++ const ai2 = ((await h.http('GET', `/api/chat/rooms/${slug}/messages`, { token: ann.token })).body.messages || []).filter((m) => m.ai_member_id);++ if (ai2.length) return ai2[ai2.length - 1];++ await pause(25);++ }++ throw new Error('no answer');++}++++t('boot', async () => {++ ai = http.createServer((req, res) => {++ req.resume();++ req.on('end', async () => {++ const cancel = /^\/api\/v1\/runs\/(run_\w+)\/cancel$/.exec(req.url);++ if (cancel) { cancels.push(cancel[1]); res.writeHead(200, { 'content-type': 'application/json' }); return res.end(JSON.stringify({ run: { id: cancel[1], status: 'cancelled' } })); }++ res.writeHead(200, { 'content-type': 'text/event-stream', 'x-openvibe-run-id': 'run_slow1' });++ let closed = false;++ res.on('close', () => { closed = true; });++ for (let i = 0; i < wordsBeforeFirst && !closed; i++) await pause(150);++ for (let i = 0; i < 40 && !closed; i++) { res.write(`event: delta\ndata: ${JSON.stringify({ text: `word${i} ` })}\n\n`); await pause(150); }++ if (!closed) { res.write(`event: run\ndata: ${JSON.stringify({ run: { id: 'run_slow1', status: 'succeeded', output: { choices: [{ message: { content: 'all words' } }] } } })}\n\n`); res.end(); }++ });++ });++ await new Promise((r) => ai.listen(0, '127.0.0.1', r));++ h = await boot({ env: { CHAT_WEB_URL: 'https://openvibe.chat' } });++ require('../server/config').ai.internalUrl = `http://127.0.0.1:${ai.address().port}`;++ aiClient = require('../server/ai/client');++ aiClient._reset();++ aiClient._setTokens({ getToken: async () => 'chat-ai-token' });++ ann = h.addUser('ann', { subject: ids.newId('user') });++ bob = h.addUser('bob', { subject: ids.newId('user') });++ for (const u of [ann, bob]) await h.ctx.upsertUser(h.live.users.get(u.id));++});++++t('Stop keeps what was written, marked as stopped, and cancels the run at AI', async () => {++ const r = await vibeApi('POST', '', ann, { message: 'Write me a long list of words' });++ assert.strictEqual(r.status, 201, r.text);++ await pause(700); // a few words in++ const s = await vibeApi('POST', `/${r.body.room.slug}/stop`, ann);++ assert.deepStrictEqual([s.status, s.body.stopped], [202, true]);++ const m = await lastAi(r.body.room.slug);++ assert.match(m.message, /^word0 word1 [\s\S]*\n\n\*Stopped\.\*$/, m.message);++ assert.ok(!m.message.includes('word39'), 'not the whole answer');++ for (let i = 0; i < 40 && !cancels.length; i++) await pause(25);++ assert.deepStrictEqual(cancels, ['run_slow1'], 'the run is cancelled at AI');++ assert.strictEqual(require('../server/rooms/ai-members').stop(999999), false, 'a room with nothing being written');++});++++t('stopped before a word: the message says so; nothing being written answers stopped: false; others cannot', async () => {++ wordsBeforeFirst = 20;++ const r = await vibeApi('POST', '', ann, { message: 'Think hard first' });++ await pause(300);++ assert.strictEqual((await vibeApi('POST', `/${r.body.room.slug}/stop`, bob)).status, 404, 'another person\'s chat');++ assert.strictEqual((await vibeApi('POST', `/${r.body.room.slug}/stop`, ann)).body.stopped, true);++ assert.strictEqual((await lastAi(r.body.room.slug)).message, 'Stopped before I wrote anything.');++ assert.strictEqual((await vibeApi('POST', `/${r.body.room.slug}/stop`, ann)).body.stopped, false, 'nothing is being written now');++});++++t.run(async () => { aiClient._reset(); if (h && h.close) await h.close(); await new Promise((r) => ai.close(r)); });
@@ -98,6 +98,11 @@ body.oc-app-body { overflow: hidden; }
@@ -98,6 +98,11 @@ body.oc-app-body { overflow: hidden; }
.vb-scroll > .oc-notice { max-width: var(--vb-col); margin: 8px auto; }
.vb-scroll > .oc-notice { max-width: var(--vb-col); margin: 8px auto; }
.vb-thread { list-style: none; margin: 0 auto; padding: 8px 0 0; max-width: var(--vb-col); display: grid; grid-template-columns: minmax(0, 1fr); gap: 22px; }
.vb-thread { list-style: none; margin: 0 auto; padding: 8px 0 0; max-width: var(--vb-col); display: grid; grid-template-columns: minmax(0, 1fr); gap: 22px; }
.vb-turn { display: flex; gap: 14px; min-width: 0; }
.vb-turn { display: flex; gap: 14px; min-width: 0; }
+/* The writing bubble's Stop (vibe.js showWait): at the bubble's end, quiet like the actions under answers. */
+.vb-stop { align-self: flex-start; margin-left: auto; padding: 4px 10px; border-radius: 8px; border: 1px solid var(--oc-line); background: transparent; color: var(--oc-muted); font: inherit; font-size: .8rem; cursor: pointer; }
+.vb-stop:hover { color: var(--oc-text); background: var(--oc-panel-2); }
+.vb-stop:focus-visible { outline: 2px solid var(--oc-accent); outline-offset: 1px; }
+.vb-stop[disabled] { opacity: .5; cursor: default; }
.vb-me { justify-content: flex-end; }
.vb-me { justify-content: flex-end; }
.vb-bubble {
.vb-bubble {
max-width: min(85%, 620px); padding: 10px 16px; border-radius: 20px 20px 6px 20px;
max-width: min(85%, 620px); padding: 10px 16px; border-radius: 20px 20px 6px 20px;
@@ -200,6 +200,17 @@
@@ -200,6 +200,17 @@
det.appendChild(ul); body.appendChild(det);
det.appendChild(ul); body.appendChild(det);
}
}
function clearWait() { var w = document.getElementById('vb-wait'); if (w) w.remove(); if (waitTimer) { clearTimeout(waitTimer); waitTimer = null; } }
function clearWait() { var w = document.getElementById('vb-wait'); if (w) w.remove(); if (waitTimer) { clearTimeout(waitTimer); waitTimer = null; } }
+ // Stop: what Vibe wrote so far stays as its answer (POST /api/chat/vibe/:slug/stop); shown once the chat exists (a new
+ // chat's first bubble is drawn before it does, so the first words add it).
+ function addStop(li) {
+ if (!room || !li || li.querySelector('.vb-stop')) return;
+ var stop = el('button', 'vb-act vb-stop', 'Stop'); stop.type = 'button';
+ stop.addEventListener('click', function () {
+ stop.disabled = true;
+ fetch('/api/chat/vibe/' + encodeURIComponent(room) + '/stop', { method: 'POST', credentials: 'same-origin' }).catch(function () { stop.disabled = false; });
+ });
+ li.appendChild(stop);
+ }
function showWait() {
function showWait() {
if (document.getElementById('vb-wait')) return;
if (document.getElementById('vb-wait')) return;
var li = el('li', 'vb-turn vb-ai vb-wait'); li.id = 'vb-wait';
var li = el('li', 'vb-turn vb-ai vb-wait'); li.id = 'vb-wait';
@@ -207,6 +218,7 @@
@@ -207,6 +218,7 @@
var body = el('div', 'vb-body'); var dots = el('span', 'vb-dots'); dots.setAttribute('aria-label', 'Vibe is writing');
var body = el('div', 'vb-body'); var dots = el('span', 'vb-dots'); dots.setAttribute('aria-label', 'Vibe is writing');
dots.appendChild(el('i')); dots.appendChild(el('i')); dots.appendChild(el('i'));
dots.appendChild(el('i')); dots.appendChild(el('i')); dots.appendChild(el('i'));
body.appendChild(dots); li.appendChild(body);
body.appendChild(dots); li.appendChild(body);
+ addStop(li);
thread().appendChild(li);
thread().appendChild(li);
// An answer takes up to a minute with tools; after two the dots give way to a way forward.
// An answer takes up to a minute with tools; after two the dots give way to a way forward.
waitTimer = setTimeout(function () { clearWait(); notice('Vibe did not answer this time. Send your message again, or try again in a minute.'); }, 120000);
waitTimer = setTimeout(function () { clearWait(); notice('Vibe did not answer this time. Send your message again, or try again in a minute.'); }, 120000);
@@ -216,6 +228,7 @@
@@ -216,6 +228,7 @@
function streamText(text) {
function streamText(text) {
showWait();
showWait();
var w = document.getElementById('vb-wait'); if (!w) return;
var w = document.getElementById('vb-wait'); if (!w) return;
+ addStop(w);
var stick = nearBottom();
var stick = nearBottom();
var body = w.querySelector('.vb-body');
var body = w.querySelector('.vb-body');
var md = body.querySelector('.vb-md');
var md = body.querySelector('.vb-md');
@@ -85,13 +85,17 @@ async function stream(workflow, input, opts = {}, onText = () => {}) {
@@ -85,13 +85,17 @@ async function stream(workflow, input, opts = {}, onText = () => {}) {
try {
try {
if (!_tokens) ai();
if (!_tokens) ai();
const token = await _tokens.getToken();
const token = await _tokens.getToken();
+ const timeout = AbortSignal.timeout(config.ai.waitMs + 10000);
const res = await (opts.fetchImpl || globalThis.fetch)(`${config.ai.internalUrl}/api/v1/runs?stream=1&wait=${Math.max(0, Math.floor(config.ai.waitMs))}`, {
const res = await (opts.fetchImpl || globalThis.fetch)(`${config.ai.internalUrl}/api/v1/runs?stream=1&wait=${Math.max(0, Math.floor(config.ai.waitMs))}`, {
method: 'POST',
method: 'POST',
headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json', Accept: 'text/event-stream', ...(opts.idempotencyKey ? { 'Idempotency-Key': opts.idempotencyKey } : {}) },
headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json', Accept: 'text/event-stream', ...(opts.idempotencyKey ? { 'Idempotency-Key': opts.idempotencyKey } : {}) },
body: JSON.stringify({ workflow, input, ...(opts.credential ? { credential: opts.credential } : {}), ...(opts.attribution !== undefined ? { attribution: opts.attribution } : {}) }),
body: JSON.stringify({ workflow, input, ...(opts.credential ? { credential: opts.credential } : {}), ...(opts.attribution !== undefined ? { attribution: opts.attribution } : {}) }),
- signal: AbortSignal.timeout(config.ai.waitMs + 10000),
+ signal: opts.signal ? AbortSignal.any([opts.signal, timeout]) : timeout,
});
});
if (!res.ok) { console.warn(`[AI] ${workflow}: AI answered ${res.status}`); return null; }
if (!res.ok) { console.warn(`[AI] ${workflow}: AI answered ${res.status}`); return null; }
+ // Stopped by the person (opts.signal): the run is cancelled at AI, so it spends nothing more.
+ const runId = res.headers.get('x-openvibe-run-id');
+ if (opts.signal && runId) opts.signal.addEventListener('abort', () => { try { const { runs } = ai(); if (runs && runs.cancel) runs.cancel(runId).catch(() => {}); } catch { /* AI away */ } }, { once: true });
let run = null;
let run = null;
if (!/event-stream/.test(String(res.headers.get('content-type') || ''))) {
if (!/event-stream/.test(String(res.headers.get('content-type') || ''))) {
const body = await res.json();
const body = await res.json();
@@ -119,7 +123,7 @@ async function stream(workflow, input, opts = {}, onText = () => {}) {
@@ -119,7 +123,7 @@ async function stream(workflow, input, opts = {}, onText = () => {}) {
if (!run || run.status !== 'succeeded' || run.synthetic) return null;
if (!run || run.status !== 'succeeded' || run.synthetic) return null;
return run.output || null;
return run.output || null;
} catch (err) {
} catch (err) {
- console.warn(`[AI] ${workflow}: ${err && err.message}`);
+ if (!(opts.signal && opts.signal.aborted)) console.warn(`[AI] ${workflow}: ${err && err.message}`);
return null;
return null;
}
}
}
}
@@ -45,6 +45,7 @@ const PER_PERSON_HOUR_VIBE = 60;
@@ -45,6 +45,7 @@ const PER_PERSON_HOUR_VIBE = 60;
const MAX_TOOL_ROUNDS = 3;
const MAX_TOOL_ROUNDS = 3;
const MAX_CALLS_PER_ROUND = 4;
const MAX_CALLS_PER_ROUND = 4;
const inFlight = new Set(); // room ids with an answer being written
const inFlight = new Set(); // room ids with an answer being written
+const stoppers = new Map(); // a Vibe chat's room id → the AbortController of the answer being written (stop())
const personHour = new Map(); // `${room}:${user}` → [timestamps]
const personHour = new Map(); // `${room}:${user}` → [timestamps]
const handleOf = (name) => String(name || '').toLowerCase().normalize('NFKD').replace(/[^\p{L}\p{N}]+/gu, '').slice(0, 24);
const handleOf = (name) => String(name || '').toLowerCase().normalize('NFKD').replace(/[^\p{L}\p{N}]+/gu, '').slice(0, 24);
@@ -283,7 +284,7 @@ function userPrompt({ member, context, transcript, askedBy, text }) {
@@ -283,7 +284,7 @@ function userPrompt({ member, context, transcript, askedBy, text }) {
* The answer through chat.member.turn with tools: up to MAX_TOOL_ROUNDS rounds of tool calls run here, then a turn
* The answer through chat.member.turn with tools: up to MAX_TOOL_ROUNDS rounds of tool calls run here, then a turn
* with tool_choice none. → the reply text, or null when a turn failed (the caller falls back to chat.member.reply).
* with tool_choice none. → the reply text, or null when a turn failed (the caller falls back to chat.member.reply).
*/
*/
-async function answerWithTools({ room, member, context, transcript, askedBy, text, opts, subject = null, images = [], user = null, messageId = null, memory = null, called = new Set(), onText = null, steps = null }) {
+async function answerWithTools({ room, member, context, transcript, askedBy, text, opts, subject = null, images = [], user = null, messageId = null, memory = null, called = new Set(), onText = null, steps = null, signal = null }) {
const tools = aiTools.definitions(member, room, { memory: memory !== null });
const tools = aiTools.definitions(member, room, { memory: memory !== null });
const vibe = Number(room.assistant) === 1;
const vibe = Number(room.assistant) === 1;
const prompt = userPrompt({ member, context, transcript, askedBy, text });
const prompt = userPrompt({ member, context, transcript, askedBy, text });
@@ -293,9 +294,10 @@ async function answerWithTools({ room, member, context, transcript, askedBy, tex
@@ -293,9 +294,10 @@ async function answerWithTools({ room, member, context, transcript, askedBy, tex
{ role: 'user', content: images.length ? [{ type: 'text', text: `${prompt}\n\n(${images.length === 1 ? 'The picture they sent is' : 'The pictures they sent are'} attached.)` }, ...images.map((url) => ({ type: 'image_url', image_url: { url } }))] : prompt },
{ role: 'user', content: images.length ? [{ type: 'text', text: `${prompt}\n\n(${images.length === 1 ? 'The picture they sent is' : 'The pictures they sent are'} attached.)` }, ...images.map((url) => ({ type: 'image_url', image_url: { url } }))] : prompt },
];
];
for (let round = 0; round <= MAX_TOOL_ROUNDS; round++) {
for (let round = 0; round <= MAX_TOOL_ROUNDS; round++) {
+ if (signal && signal.aborted) return null; // stopped (stop()): the caller keeps what was written
const last = round === MAX_TOOL_ROUNDS;
const last = round === MAX_TOOL_ROUNDS;
const turn = { messages, tools, tool_choice: last ? 'none' : 'auto', max_tokens: vibe ? 1500 : 600, temperature: 0.7 };
const turn = { messages, tools, tool_choice: last ? 'none' : 'auto', max_tokens: vibe ? 1500 : 600, temperature: 0.7 };
- const turnOpts = { ...opts, idempotencyKey: `${opts.idempotencyKey}-t${round}` };
+ const turnOpts = { ...opts, idempotencyKey: `${opts.idempotencyKey}-t${round}`, ...(signal ? { signal } : {}) };
// Vibe's answer streams (each round's text so far, to the chat as it is written); other members answer whole.
// Vibe's answer streams (each round's text so far, to the chat as it is written); other members answer whole.
const out = onText ? await aiClient.stream('chat.member.turn', turn, turnOpts, onText) : await aiClient.structured('chat.member.turn', turn, turnOpts);
const out = onText ? await aiClient.stream('chat.member.turn', turn, turnOpts, onText) : await aiClient.structured('chat.member.turn', turn, turnOpts);
const msg = out && Array.isArray(out.choices) && out.choices[0] && out.choices[0].message;
const msg = out && Array.isArray(out.choices) && out.choices[0] && out.choices[0].message;
@@ -369,9 +371,17 @@ async function afterPost(room, message, user, deps = {}) {
@@ -369,9 +371,17 @@ async function afterPost(room, message, user, deps = {}) {
lastDelta = now;
lastDelta = now;
broadcast({ type: 'room_ai_delta', room: room.slug, name: member.name, text: String(pending).slice(0, 8000) });
broadcast({ type: 'room_ai_delta', room: room.slug, name: member.name, text: String(pending).slice(0, 8000) });
} : null;
} : null;
+ // Stop (stop(), the chat's Stop button): the answer so far is kept, marked as stopped, and the run is cancelled.
+ const controller = vibe ? new AbortController() : null;
+ if (controller) stoppers.set(room.id, controller);
if (config.ai.memberTools) {
if (config.ai.memberTools) {
const images = await picturesFor(message, history, user).catch(() => []);
const images = await picturesFor(message, history, user).catch(() => []);
- try { text = clean(await answerWithTools({ room, member, context, transcript, askedBy, text: String(message.message).slice(0, vibe ? 12000 : 2000), opts, subject: user.subject_id || null, images, user, messageId: message.id, memory, called, onText, steps }) || '', vibe ? 8000 : 2000); } catch { text = ''; }
+ try { text = clean(await answerWithTools({ room, member, context, transcript, askedBy, text: String(message.message).slice(0, vibe ? 12000 : 2000), opts, subject: user.subject_id || null, images, user, messageId: message.id, memory, called, onText, steps, signal: controller && controller.signal }) || '', vibe ? 8000 : 2000); } catch { text = ''; }
+ }
+ if (controller && controller.signal.aborted) {
+ const sofar = clean(pending || '', 7900);
+ if (sofar && !own) await require('./vibe').countReply(user);
+ return await say(sofar ? `${sofar}\n\n*Stopped.*` : 'Stopped before I wrote anything.', steps && steps.steps.length ? steps : null);
}
}
// A plain "remember that …" is kept even when the model only promised it (small models often do).
// A plain "remember that …" is kept even when the model only promised it (small models often do).
if (memory !== null && text && !called.has('remember')) {
if (memory !== null && text && !called.has('remember')) {
@@ -395,9 +405,18 @@ async function afterPost(room, message, user, deps = {}) {
@@ -395,9 +405,18 @@ async function afterPost(room, message, user, deps = {}) {
return null;
return null;
} finally {
} finally {
inFlight.delete(room.id);
inFlight.delete(room.id);
+ stoppers.delete(room.id);
}
}
}
}
+/** Stop the answer being written in a Vibe chat → true when one was. */
+function stop(roomId) {
+ const c = stoppers.get(Number(roomId)) || stoppers.get(roomId);
+ if (!c || c.signal.aborted) return false;
+ c.abort();
+ return true;
+}
+
/** What the room shows a member thinks with: "OpenVibe", or the own key's provider, model and whose key it is. */
/** What the room shows a member thinks with: "OpenVibe", or the own key's provider, model and whose key it is. */
const modelOf = (m) => (m.engine === 'own' ? (m.engine_label || 'someone\'s own key') : 'OpenVibe');
const modelOf = (m) => (m.engine === 'own' ? (m.engine_label || 'someone\'s own key') : 'OpenVibe');
@@ -407,6 +426,6 @@ const toWire = (m) => ({
@@ -407,6 +426,6 @@ const toWire = (m) => ({
engine: m.engine || 'openvibe', model: modelOf(m), created_at: m.created_at, updated_at: m.updated_at,
engine: m.engine || 'openvibe', model: modelOf(m), created_at: m.created_at, updated_at: m.updated_at,
});
});
-function _reset() { inFlight.clear(); personHour.clear(); }
+function _reset() { inFlight.clear(); personHour.clear(); stoppers.clear(); }
-module.exports = { MAX_PER_ROOM, LIMITS, AiMemberError, enabled, list, byId, add, update, remove, dropKey, relabelKey, modelOf, mentioned, handleOf, afterPost, transcriptOf, vibePrompt, toWire, _reset };
+module.exports = { MAX_PER_ROOM, LIMITS, AiMemberError, enabled, list, byId, add, update, remove, dropKey, relabelKey, modelOf, mentioned, handleOf, afterPost, stop, transcriptOf, vibePrompt, toWire, _reset };
@@ -14,6 +14,7 @@
@@ -14,6 +14,7 @@
* DELETE /:slug 204: the chat and its messages are deleted
* DELETE /:slug 204: the chat and its messages are deleted
* POST /:slug/retry 202 { removed }: Vibe's last answer goes and it answers the person's last message anew (409
* POST /:slug/retry 202 { removed }: Vibe's last answer goes and it answers the person's last message anew (409
* when the last two messages are not that exchange)
* when the last two messages are not that exchange)
+ * POST /:slug/stop 202 { stopped }: the answer being written stops; what was written stays, marked as stopped
* GET /memory { on, notes: [{ id, text, source, created_at }], max }: what Vibe remembers about the person
* GET /memory { on, notes: [{ id, text, source, created_at }], max }: what Vibe remembers about the person
* (server/rooms/vibe-memory.js)
* (server/rooms/vibe-memory.js)
* POST /memory { text } → 201 { note } (a note of their own), 422 when it is empty, too long, secret-looking,
* POST /memory { text } → 201 { note } (a note of their own), 422 when it is empty, too long, secret-looking,
@@ -136,6 +137,14 @@ router.post('/:slug/retry', requireAuth, personOnly, startLimit, async (req, res
@@ -136,6 +137,14 @@ router.post('/:slug/retry', requireAuth, personOnly, startLimit, async (req, res
} catch (err) { send(res, err); }
} catch (err) { send(res, err); }
});
});
+router.post('/:slug/stop', requireAuth, personOnly, manageLimit, async (req, res) => {
+ try {
+ const room = await vibe.ownChat(req.params.slug, req.user);
+ if (!room) return res.status(404).json({ error: 'No such chat', code: 'rooms.not_found' });
+ res.status(202).json({ stopped: require('./ai-members').stop(room.id) });
+ } catch (err) { send(res, err); }
+});
+
router.patch('/:slug', requireAuth, personOnly, manageLimit, async (req, res) => {
router.patch('/:slug', requireAuth, personOnly, manageLimit, async (req, res) => {
try {
try {
const room = await vibe.ownChat(req.params.slug, req.user);
const room = await vibe.ownChat(req.params.slug, req.user);
new file mode 100644
new file mode 100644
@@ -0,0 +1,76 @@
@@ -0,0 +1,76 @@
+'use strict';
+/**
+ * Stop while Vibe writes (server/rooms/ai-members.js stop, vibe-routes.js POST /:slug/stop, server/ai/client.js stream):
+ * the answer so far stays as Vibe's message, marked as stopped, and the run is cancelled at AI; with nothing written,
+ * the message says so; a chat with nothing being written answers stopped: false; another person's chat is not there.
+ */
+const assert = require('assert');
+const http = require('http');
+const { ids } = require('openvibe-contracts');
+const { boot, suite } = require('./helpers');
+
+const t = suite('vibe stop');
+let h, ann, bob, ai, aiClient;
+const cancels = [];
+let wordsBeforeFirst = 0; // how long the fake AI "thinks" before its first word, in 150 ms steps
+const pause = (ms) => new Promise((r) => setTimeout(r, ms));
+const vibeApi = (method, path, u, body) => h.http(method, `/api/chat/vibe${path}`, { token: u.token, body });
+async function lastAi(slug) {
+ for (let i = 0; i < 120; i++) {
+ const ai2 = ((await h.http('GET', `/api/chat/rooms/${slug}/messages`, { token: ann.token })).body.messages || []).filter((m) => m.ai_member_id);
+ if (ai2.length) return ai2[ai2.length - 1];
+ await pause(25);
+ }
+ throw new Error('no answer');
+}
+
+t('boot', async () => {
+ ai = http.createServer((req, res) => {
+ req.resume();
+ req.on('end', async () => {
+ const cancel = /^\/api\/v1\/runs\/(run_\w+)\/cancel$/.exec(req.url);
+ if (cancel) { cancels.push(cancel[1]); res.writeHead(200, { 'content-type': 'application/json' }); return res.end(JSON.stringify({ run: { id: cancel[1], status: 'cancelled' } })); }
+ res.writeHead(200, { 'content-type': 'text/event-stream', 'x-openvibe-run-id': 'run_slow1' });
+ let closed = false;
+ res.on('close', () => { closed = true; });
+ for (let i = 0; i < wordsBeforeFirst && !closed; i++) await pause(150);
+ for (let i = 0; i < 40 && !closed; i++) { res.write(`event: delta\ndata: ${JSON.stringify({ text: `word${i} ` })}\n\n`); await pause(150); }
+ if (!closed) { res.write(`event: run\ndata: ${JSON.stringify({ run: { id: 'run_slow1', status: 'succeeded', output: { choices: [{ message: { content: 'all words' } }] } } })}\n\n`); res.end(); }
+ });
+ });
+ await new Promise((r) => ai.listen(0, '127.0.0.1', r));
+ h = await boot({ env: { CHAT_WEB_URL: 'https://openvibe.chat' } });
+ require('../server/config').ai.internalUrl = `http://127.0.0.1:${ai.address().port}`;
+ aiClient = require('../server/ai/client');
+ aiClient._reset();
+ aiClient._setTokens({ getToken: async () => 'chat-ai-token' });
+ ann = h.addUser('ann', { subject: ids.newId('user') });
+ bob = h.addUser('bob', { subject: ids.newId('user') });
+ for (const u of [ann, bob]) await h.ctx.upsertUser(h.live.users.get(u.id));
+});
+
+t('Stop keeps what was written, marked as stopped, and cancels the run at AI', async () => {
+ const r = await vibeApi('POST', '', ann, { message: 'Write me a long list of words' });
+ assert.strictEqual(r.status, 201, r.text);
+ await pause(700); // a few words in
+ const s = await vibeApi('POST', `/${r.body.room.slug}/stop`, ann);
+ assert.deepStrictEqual([s.status, s.body.stopped], [202, true]);
+ const m = await lastAi(r.body.room.slug);
+ assert.match(m.message, /^word0 word1 [\s\S]*\n\n\*Stopped\.\*$/, m.message);
+ assert.ok(!m.message.includes('word39'), 'not the whole answer');
+ for (let i = 0; i < 40 && !cancels.length; i++) await pause(25);
+ assert.deepStrictEqual(cancels, ['run_slow1'], 'the run is cancelled at AI');
+ assert.strictEqual(require('../server/rooms/ai-members').stop(999999), false, 'a room with nothing being written');
+});
+
+t('stopped before a word: the message says so; nothing being written answers stopped: false; others cannot', async () => {
+ wordsBeforeFirst = 20;
+ const r = await vibeApi('POST', '', ann, { message: 'Think hard first' });
+ await pause(300);
+ assert.strictEqual((await vibeApi('POST', `/${r.body.room.slug}/stop`, bob)).status, 404, 'another person\'s chat');
+ assert.strictEqual((await vibeApi('POST', `/${r.body.room.slug}/stop`, ann)).body.stopped, true);
+ assert.strictEqual((await lastAi(r.body.room.slug)).message, 'Stopped before I wrote anything.');
+ assert.strictEqual((await vibeApi('POST', `/${r.body.room.slug}/stop`, ann)).body.stopped, false, 'nothing is being written now');
+});
+
+t.run(async () => { aiClient._reset(); if (h && h.close) await h.close(); await new Promise((r) => ai.close(r)); });