openvibers/openvibe.live Commits
Agent Streams S7: schedules, party counts and the robot scene (#648)
Changes
nnew file mode 100644@@@ -0,0 +1,53 @@++-- phase: expand++-- Agent Streams S7 (design 2026-11-06, AGENT-STREAMS-S7-REACH-DESIGN): schedules for public agent streams.++--++-- An owner schedules a public agent stream: a time, a task template and the mode. The schedule shows on the++-- channel's Agents tab and in Today 24 hours before its time. When the time comes, the owner's desk claims it: Live++-- creates the stream and hands the desk its publish ticket. A desk that is not online does not claim, and Live++-- records the schedule as missed with 'desk_offline' ("didn't start: the desk was offline") and starts nothing. A++-- chatplays schedule never starts on its own; it waits for the owner's Go (the S6 rule).++--++-- agent_desks is the desk heartbeat: a desk says it is online every 30 s, so the sweep can tell 'the desk was++-- offline' apart from 'the desk was online but the task was not ready'. A heartbeat is a fact, not a stream: it++-- carries no ticket and no secret.++--++-- agent_schedule_reminders is opt-in: a follower who asks for a reminder gets one Today notification. The row names++-- the person, and the schedule's public view carries only the count.++--++-- Expand only: every table is new and an older release ignores all of it.++CREATE TABLE IF NOT EXISTS agent_stream_schedules (++ id text COLLATE "C" PRIMARY KEY, -- scd_<ULID>++ user_id bigint NOT NULL REFERENCES users(id) ON DELETE CASCADE,++ managed_stream_id bigint REFERENCES managed_streams(id) ON DELETE SET NULL,++ title text COLLATE "C" NOT NULL,++ task jsonb NOT NULL,++ mode text COLLATE "C" NOT NULL DEFAULT 'watch' CHECK (mode IN ('watch', 'narrated', 'copilot', 'chatplays')),++ visibility text COLLATE "C" NOT NULL DEFAULT 'public' CHECK (visibility IN ('private', 'unlisted', 'public')),++ starts_at timestamptz NOT NULL,++ state text COLLATE "C" NOT NULL DEFAULT 'scheduled' CHECK (state IN ('scheduled', 'live', 'ended', 'missed', 'cancelled')),++ missed_reason text COLLATE "C" CHECK (missed_reason IN ('desk_offline', 'task_not_ready', 'waits_for_go')),++ waits_for_go boolean NOT NULL DEFAULT false,++ started_at timestamptz,++ ended_at timestamptz,++ created_at timestamptz NOT NULL DEFAULT now(),++ updated_at timestamptz NOT NULL DEFAULT now()++);++CREATE INDEX IF NOT EXISTS agent_stream_schedules_due ON agent_stream_schedules (starts_at) WHERE state = 'scheduled';++CREATE INDEX IF NOT EXISTS agent_stream_schedules_user ON agent_stream_schedules (user_id, starts_at);++CREATE UNIQUE INDEX IF NOT EXISTS agent_stream_schedules_stream ON agent_stream_schedules (managed_stream_id) WHERE managed_stream_id IS NOT NULL;++++-- The desk's heartbeat: online means 'heard from within the last three minutes'.++CREATE TABLE IF NOT EXISTS agent_desks (++ user_id bigint PRIMARY KEY REFERENCES users(id) ON DELETE CASCADE,++ last_seen_at timestamptz NOT NULL DEFAULT now(),++ version text COLLATE "C"++);++++-- A follower's opt-in to a schedule's reminder. The subject is the follower's Network subject, so a schedule's++-- reminder survives a Live account rename; the public view carries the count only.++CREATE TABLE IF NOT EXISTS agent_schedule_reminders (++ schedule_id text COLLATE "C" NOT NULL REFERENCES agent_stream_schedules(id) ON DELETE CASCADE,++ subject_id text COLLATE "C" NOT NULL,++ created_at timestamptz NOT NULL DEFAULT now(),++ PRIMARY KEY (schedule_id, subject_id)++);@@@ -23,7 +23,7 @@"multer": "^1.4.5-lts.1","node-media-server": "^2.7.1","onnxruntime-node": "^1.20.0",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.299.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.305.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.93.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.124.0","pg": "^8.23.1",@@@ -2840,9 +2840,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.299.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.299.0",-- "integrity": "sha512-KoJW79nApPmE7Yoi1kFVDH5gabUwcDmwvtWaPBdM8nHqU2+gaOd7BkkvfYBbtbCR0nwf0nuY7qFTiKGXdme+rg==",++ "version": "0.305.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.305.0",++ "integrity": "sha512-jTU1zlnTmHlydwL3/c5HVdUiPpd8r2KucVdHYUYX5MZgJoqy4M7Hg2jy6UKbJQ6s665uIZyTvswbmJjB2uwc5Q==","license": "AGPL-3.0-only","dependencies": {"ajv": "^8.17.1",@@@ -42,7 +42,7 @@"multer": "^1.4.5-lts.1","node-media-server": "^2.7.1","onnxruntime-node": "^1.20.0",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.299.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.305.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.93.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.124.0","pg": "^8.23.1",@@@ -179,3 +179,19 @@.agent-controls-log-list { margin: 6px 0 0; padding-left: 18px; font-size: .85rem; color: var(--text-secondary, #96a7c2); }.agent-overlay-paused { margin: 2px 0 0; font-size: .8rem; font-weight: 600; color: var(--warning-text, #fbbf24); }@media (max-width: 480px) { .agent-controls-row .btn { flex: 1 1 calc(50% - 8px); } }++++++/* ── S7: the robot camera as a second scene, and the scheduled cards ──────────────────────────── */++.agent-scenes { display: flex; gap: 6px; margin: 10px 0 0; }++.agent-scene { padding: 6px 12px; border-radius: 999px; border: 1px solid var(--ov-border, #2a3140); background: var(--ov-surface, #121722); color: var(--ov-text, #e6e9ef); font: inherit; cursor: pointer; }++.agent-scene.is-on { background: var(--ov-accent, #6ea8fe); color: #06101f; border-color: transparent; }++.agent-robot { position: relative; aspect-ratio: 16 / 9; border-radius: 14px; overflow: hidden; background: #000; margin-top: 10px; }++.agent-robot iframe { width: 100%; height: 100%; border: 0; display: block; }++.agent-schedules { display: grid; grid-template-columns: repeat(auto-fill, minmax(230px, 1fr)); gap: 12px; margin: 0 0 16px; }++.agent-schedules-title { margin: 18px 0 10px; font-size: 1rem; }++.agent-schedule-card { display: flex; flex-direction: column; gap: 5px; padding: 12px 14px; border-radius: 12px; border: 1px solid var(--ov-border, #2a3140); background: var(--ov-surface, #121722); color: inherit; text-decoration: none; }++.agent-schedule-card:hover { border-color: var(--ov-accent, #6ea8fe); }++.agent-schedule-when { font-size: .78rem; text-transform: uppercase; letter-spacing: .04em; opacity: .8; }++.agent-schedule-title { font-size: .98rem; line-height: 1.25; }++.agent-schedule-task, .agent-schedule-meta { font-size: .82rem; opacity: .8; }++@media (max-width: 640px) { .agent-scene { min-height: 44px; } }@@@ -28,6 +28,13 @@<div class="agent-curtain" id="agent-curtain" hidden>Private screen</div><div id="agent-overlay"></div></div>++ <div class="agent-scenes" id="agent-scenes" hidden role="group" aria-label="The stream's scenes">++ <button type="button" class="agent-scene is-on" id="agent-scene-desk" data-agent-scene="desk" aria-pressed="true">Desk</button>++ <button type="button" class="agent-scene" id="agent-scene-robot" data-agent-scene="robot" aria-pressed="false">Robot camera</button>++ </div>++ <div class="agent-robot" id="agent-robot" hidden>++ <iframe id="agent-robot-frame" title="Robot camera" loading="lazy" allow="camera; microphone; fullscreen"></iframe>++ </div><p class="agent-status-line" id="agent-status-line">Loading…</p><section class="agent-chapters" id="agent-chapters" hidden aria-label="The recording's chapters"><h2 class="agent-chapters-title" id="agent-chapters-title">Chapters</h2>@@@ -378,6 +378,8 @@<!-- Agents (Agent Streams S2): this channel's AI agent streams, live and past. --><div class="ch-tab-panel" id="ch-panel-agents" role="tabpanel"><p class="ch-panel-hint"><i class="fa-solid fa-robot"></i> AI agents working on this channel — live now, and what they finished.</p>++ <h3 class="agent-schedules-title" id="ch-agents-schedules-title" hidden>Coming up</h3>++ <div class="agent-schedules" id="ch-agents-schedules"></div><div class="agent-cards" id="ch-agents-grid"></div><button class="btn btn-small btn-outline agent-more" id="ch-agents-more" type="button" hidden onclick="loadMoreChannelAgents(this)">More</button></div>@@@ -93,5 +93,35 @@};}-- window.AgentCards = { card, fill, minutes };++ /** A scheduled stream's card: the time, the task and the reminder count (never who opted in). */++ function scheduleCard(entry) {++ const a = el('a', 'agent-schedule-card');++ a.href = entry.page_url || '#';++ const when = el('span', 'agent-schedule-when');++ const at = new Date(entry.starts_at);++ when.textContent = Number.isNaN(at.getTime()) ? '' : at.toLocaleString(undefined, { weekday: 'short', month: 'short', day: 'numeric', hour: 'numeric', minute: '2-digit' });++ a.appendChild(when);++ a.appendChild(el('b', 'agent-schedule-title', entry.title || 'An AI agent stream'));++ const task = where(entry.task);++ if (task) a.appendChild(el('span', 'agent-schedule-task', task));++ const meta = el('span', 'agent-schedule-meta');++ const bits = [entry.mode === 'chatplays' ? 'Chat plays · waits for Go' : { narrated: 'Narrated', copilot: 'Co-pilot', watch: 'Watch' }[entry.mode] || 'Watch'];++ if (entry.reminders) bits.push(`${entry.reminders} reminder${entry.reminders === 1 ? '' : 's'}`);++ if (entry.waits_for_go) bits.push('waiting for the owner\'s Go');++ if (entry.state === 'missed' && entry.missed_reason === 'desk_offline') bits.push("Didn't start: the desk was offline");++ meta.textContent = bits.join(' · ');++ a.appendChild(meta);++ return a;++ }++++ /** Fill a schedule rail; the caller hides it when it is empty. */++ function fillSchedules(grid, entries) {++ if (!grid) return 0;++ grid.replaceChildren();++ const all = Array.isArray(entries) ? entries : [];++ for (const entry of all) grid.appendChild(scheduleCard(entry));++ return all.length;++ }++++ window.AgentCards = { card, fill, scheduleCard, fillSchedules, minutes };})();@@@ -306,6 +306,34 @@});}++ // S7: a channel bound to a Bot robot may show the robot's camera as a second scene beside the desk. The panel is++ // Bot's own embed; Live only switches which scene is on, and the overlay stays the same on both.++ function renderScenes(stream) {++ const bar = document.getElementById('agent-scenes');++ const robot = document.getElementById('agent-robot');++ const frame = document.getElementById('agent-robot-frame');++ const stage = document.getElementById('agent-stage');++ if (!bar || !robot || !stage) return;++ const scene = stream && stream.robot_scene;++ if (!scene || !scene.embed_url) { bar.hidden = true; robot.hidden = true; stage.hidden = false; return; }++ if (frame && frame.getAttribute('src') !== scene.embed_url) frame.setAttribute('src', scene.embed_url);++ bar.hidden = false;++ const show = (name) => {++ const isRobot = name === 'robot';++ stage.hidden = isRobot;++ robot.hidden = !isRobot;++ for (const btn of bar.querySelectorAll('[data-agent-scene]')) {++ const on = btn.getAttribute('data-agent-scene') === name;++ btn.classList.toggle('is-on', on);++ btn.setAttribute('aria-pressed', on ? 'true' : 'false');++ }++ };++ for (const btn of bar.querySelectorAll('[data-agent-scene]')) {++ btn.onclick = () => show(btn.getAttribute('data-agent-scene'));++ }++ show('desk');++ }++function render(data) {state = data;ticket = data.view_ticket || ticket;@@@ -330,6 +358,7 @@}renderCurtain(!!(stream.curtain && stream.curtain.up));++ renderScenes(stream);renderStatus(stream, liveViewers);mountPlays(data);ensureOverlay(data);@@@ -422,6 +451,7 @@}}renderCurtain(!!(stream.curtain && stream.curtain.up));++ renderScenes(stream);renderVoice();const endBtn = $('agent-end');if (endBtn) endBtn.hidden = stream.state === 'ended';@@@ -1992,9 +1992,14 @@ async function loadChannelAgents(username, { force = false } = {}) {try { data = await api(`/streams/agent?channel=${encodeURIComponent(username)}`); } catch { data = null; }if (username !== currentChannelUsername) return; // the page moved on while this was in flightconst cards = (data && Array.isArray(data.agents)) ? data.agents : [];++ const planned = (data && Array.isArray(data.schedules)) ? data.schedules : [];++ const schedulesEl = document.getElementById('ch-agents-schedules');++ const schedulesTitle = document.getElementById('ch-agents-schedules-title');++ const plannedCount = (window.AgentCards && schedulesEl) ? window.AgentCards.fillSchedules(schedulesEl, planned) : 0;++ if (schedulesTitle) schedulesTitle.hidden = !plannedCount;const tab = document.getElementById('ch-tab-btn-agents');const badge = document.getElementById('ch-tab-badge-agents');-- if (tab) tab.style.display = cards.length ? '' : 'none';++ if (tab) tab.style.display = (cards.length || plannedCount) ? '' : 'none';if (badge) {const live = cards.filter((entry) => entry.state === 'live').length;badge.hidden = !live;@@@ -2005,7 +2010,7 @@ async function loadChannelAgents(username, { force = false } = {}) {if (!cards.length) {const empty = document.createElement('p');empty.className = 'agent-empty';-- empty.textContent = 'No AI agent streams on this channel yet.';++ empty.textContent = plannedCount ? 'No AI agent stream is live right now; the next one is above.' : 'No AI agent streams on this channel yet.';grid.appendChild(empty);if (more) more.hidden = true;return;@@@ -252,6 +252,7 @@ async function _rdTopic(root, slug) {${t.summary ? `<div class="rd-summary"><span class="rd-desk">The desk</span>${_rdParas(t.summary)}</div>` : ''}<h2 class="rd-h2">Every take</h2>${(d.takes || []).map((s) => _rdTake(s, { topic: false })).join('') || _rdEmpty('fa-comment', 'No takes on this topic right now.')}++ ${(d.citations || []).length ? `<h2 class="rd-h2">Agents that worked on this</h2><ul class="rd-onair-list">${d.citations.map((r) => `<li><a class="rd-onair-item" href="https://openvibe.report/s/${encodeURIComponent(r.id)}"><span class="rd-onair-t"><b>${_rdEsc(r.title)}</b><small>${_rdEsc(r.summary || '')} · ${_rdEsc(r.outcome || 'done')}</small></span></a></li>`).join('')}</ul>` : ''}${(d.related || []).length ? `<h2 class="rd-h2">Came up alongside</h2><div class="rd-chips">${d.related.map((r) => _rdTopicChip(r)).join('')}</div>` : ''}</article>`;document.title = `${t.title} · The Rundown · OpenVibe.Live`;@@@ -28,6 +28,7 @@ const tickets = require('./tickets');const frames = require('./frames');const steps = require('./steps');const voice = require('./voice');++const schedules = require('./schedules');const router = express.Router();@@@ -128,7 +129,9 @@ router.get('/', auth.viewer, async (req, res) => {const channel = typeof req.query.channel === 'string' ? req.query.channel.trim().slice(0, 64) : null;if (channel) {const rows = await service.listForChannel(channel, req.viewer ? req.viewer.id : null);-- return res.json({ agents: rows.map(service.cardView) });++ const subject = req.viewer ? await auth.viewerSubject(req.viewer) : null;++ const planned = await schedules.listForChannel(channel, req.viewer ? req.viewer.id : null, { subject });++ return res.json({ agents: rows.map(service.cardView), schedules: planned });}const rows = await service.listPublic({ liveOnly: req.query.live === '1' });res.json({ agents: rows.map(service.cardView) });@@@ -178,6 +181,138 @@ async function watchable(req, row) {});}++++// ── Schedules (S7) ─────────────────────────────────────────────++//++// An owner schedules a public agent stream; the record shows on the channel's Agents tab and in Today 24 hours before++// its time. The desk says it is online with a heartbeat, claims a due schedule and receives the one publish ticket.++// A schedule nobody claims says "didn't start: the desk was offline" and starts nothing. A chatplays schedule waits++// for the owner's Go. Reminders are opt-in and the answer is a count, never a name.++++/** POST /desk/heartbeat — the owner's desk says it is online (a fact, not a stream). */++router.post('/desk/heartbeat', auth.publisher, async (req, res) => {++ try {++ const out = await schedules.heartbeat(req.publisher.user, (req.body || {}).version);++ res.set('Cache-Control', 'no-store');++ res.json(out);++ } catch (err) {++ console.error('[AgentStream] desk heartbeat failed:', err.message);++ fail(res, 500, 'heartbeat_failed', 'The heartbeat could not be stored');++ }++});++++/** POST /schedules — the owner writes a schedule. */++router.post('/schedules', auth.publisher, async (req, res) => {++ try {++ const body = req.body || {};++ const task = schedules.cleanTask(body.task);++ if (!task) return fail(res, 400, 'invalid_task', 'task needs source (codes, actor, desk, node), ref and title');++ const schedule = await schedules.create({++ user: req.publisher.user,++ title: body.title,++ task,++ mode: body.mode,++ startsAt: body.starts_at,++ });++ res.status(201).json({ schedule });++ } catch (err) {++ if (err.status) return fail(res, err.status, err.code, err.message);++ console.error('[AgentStream] schedule create failed:', err.message);++ fail(res, 500, 'schedule_failed', 'The schedule could not be stored');++ }++});++++/** GET /schedules — the caller's own (?mine=1) or a channel's public window (the channel's Agents tab). */++router.get('/schedules', auth.viewer, async (req, res) => {++ try {++ res.set('Cache-Control', 'no-store');++ const subject = req.viewer ? await auth.viewerSubject(req.viewer) : null;++ if (String(req.query.mine || '') === '1') {++ if (!req.viewer) return res.json({ schedules: [] });++ return res.json({ schedules: await schedules.listMine(req.viewer.id, { subject }) });++ }++ if (String(req.query.public || '') === '1') {++ return res.json({ schedules: await schedules.listPublic({ subject }) });++ }++ const channel = typeof req.query.channel === 'string' ? req.query.channel.trim().slice(0, 64) : null;++ if (!channel) return fail(res, 400, 'channel_required', 'channel is required (or mine=1 or public=1)');++ const viewerId = req.viewer ? req.viewer.id : null;++ const rows = await schedules.listForChannel(channel, viewerId, { subject });++ res.json({ schedules: rows });++ } catch (err) {++ console.error('[AgentStream] schedules read failed:', err.message);++ fail(res, 500, 'schedules_failed', 'The schedules could not be read');++ }++});++++/** GET /schedules/:id — one schedule (the owner's own, or a public one inside its 24-hour window). */++router.get('/schedules/:id', auth.viewer, async (req, res) => {++ try {++ const subject = req.viewer ? await auth.viewerSubject(req.viewer) : null;++ const schedule = await schedules.get(req.params.id, { subject });++ if (!schedule) return fail(res, 404, 'schedule_not_found', 'No such schedule');++ const mine = req.viewer && schedule.owner && schedule.owner === subject;++ const soon = schedule.visibility === 'public' && schedule.state === 'scheduled'++ && Date.parse(schedule.starts_at) <= Date.now() + schedules.AHEAD_MS;++ if (!mine && !soon) return fail(res, 404, 'schedule_not_found', 'No such schedule');++ res.set('Cache-Control', 'no-store');++ res.json({ schedule });++ } catch (err) {++ console.error('[AgentStream] schedule read failed:', err.message);++ fail(res, 500, 'schedule_failed', 'The schedule could not be read');++ }++});++++/** DELETE /schedules/:id — cancel it. */++router.delete('/schedules/:id', auth.publisher, async (req, res) => {++ try {++ const schedule = await schedules.cancel(req.params.id, req.publisher.user);++ res.json({ schedule });++ } catch (err) {++ if (err.status) return fail(res, err.status, err.code, err.message);++ console.error('[AgentStream] schedule cancel failed:', err.message);++ fail(res, 500, 'schedule_failed', 'The schedule could not be cancelled');++ }++});++++/** POST /schedules/:id/remind — opt in to the reminder. The answer is the count, never the people. */++router.post('/schedules/:id/remind', auth.viewer, async (req, res) => {++ try {++ if (!req.viewer) return fail(res, 401, 'authentication_required', 'Sign in to set a reminder');++ const subject = await auth.viewerSubject(req.viewer);++ const out = await schedules.remind(req.params.id, subject);++ res.json(out);++ } catch (err) {++ if (err.status) return fail(res, err.status, err.code, err.message);++ console.error('[AgentStream] schedule reminder failed:', err.message);++ fail(res, 500, 'reminder_failed', 'The reminder could not be set');++ }++});++++/** POST /schedules/:id/claim — the owner's online desk starts it: the stream and its one publish ticket. */++router.post('/schedules/:id/claim', auth.publisher, async (req, res) => {++ try {++ const out = await schedules.claim(req.params.id, req.publisher.user, { go: false });++ res.status(201).json(out);++ } catch (err) {++ if (err.status) return fail(res, err.status, err.code, err.message);++ console.error('[AgentStream] schedule claim failed:', err.message);++ fail(res, 500, 'schedule_failed', 'The schedule could not be started');++ }++});++++/** POST /schedules/:id/go — the owner's Go for a chatplays schedule; it starts the stream like a claim. */++router.post('/schedules/:id/go', auth.publisher, async (req, res) => {++ try {++ const out = await schedules.claim(req.params.id, req.publisher.user, { go: true });++ res.status(201).json(out);++ } catch (err) {++ if (err.status) return fail(res, err.status, err.code, err.message);++ console.error('[AgentStream] schedule go failed:', err.message);++ fail(res, 500, 'schedule_failed', 'The schedule could not be started');++ }++});++// ── Read ───────────────────────────────────────────────────────/**@@@ -222,8 +357,14 @@ router.get('/:id', auth.viewer, async (req, res) => {: null;const moderation = await channelModeration(row.channel_id);const isMod = !isOwner && !!(req.viewer && moderation.moderator_ids.includes(Number(req.viewer.id)));++ let parties = null;++ try {++ const count = await require('../chat/chat-client').partyCount(row.public_id);++ if (count && Number.isInteger(Number(count.count))) parties = { count: Math.max(0, Number(count.count)) };++ } catch { /* Chat is optional for a stream */ }res.json({stream,++ parties,live_stream_id: row.live_session_id ? Number(row.live_session_id) : null,can_manage: isOwner,can_moderate: isOwner || isMod,nnew file mode 100644@@@ -0,0 +1,284 @@++'use strict';++/**++ * Scheduled public agent streams (Agent Streams S7, design 2026-11-06).++ *++ * An owner schedules a public agent stream: a time, a task template (a Codes task or an Actor recipe) and the mode.++ * The record shows on the channel's Agents tab and in Today 24 hours before its time. When the time comes the owner's++ * desk claims it: Live creates the stream and hands the desk a publish ticket (the ticket is issued once and never++ * stored raw). A desk that is not online does not claim, and the sweep records the schedule as missed with++ * `desk_offline` — the design's "didn't start: the desk was offline" — and starts nothing.++ *++ * A chatplays schedule never starts on its own: it keeps its time and `waits_for_go` until the owner presses Go (the++ * S6 rule). Reminders are opt-in only: a follower asks, the row names them, and every public view carries the count.++ *++ * The desk heartbeat (agent_desks) tells the sweep whether the owner's desk was online at the time, so the card can++ * say which of the two happened. A heartbeat is a fact, not a stream.++ */++const db = require('../db/database');++const contracts = require('openvibe-contracts');++const service = require('./service');++++const MODES = service.MODES;++const MAX_TITLE = 200;++const MAX_TASK_REF = 200;++const AHEAD_MS = 24 * 60 * 60 * 1000;++const SCHEDULE_WINDOW_MS = 30 * 24 * 60 * 60 * 1000;++const DESK_ONLINE_MS = 3 * 60 * 1000;++const MISS_GRACE_MS = 60 * 1000;++const TASK_SOURCES = service.TASK_SOURCES;++++class ScheduleError extends Error {++ constructor(status, code, message) {++ super(message);++ this.status = status;++ this.code = code;++ }++}++++function text(value, max) {++ if (typeof value !== 'string') return null;++ const s = value.trim();++ return s && Array.from(s).length <= max ? s : null;++}++++function cleanTask(task) {++ if (!task || typeof task !== 'object' || Array.isArray(task)) return null;++ const source = TASK_SOURCES.includes(task.source) ? task.source : null;++ const ref = text(task.ref, MAX_TASK_REF);++ const title = text(task.title, MAX_TITLE);++ if (!source || !ref || !title) return null;++ const url = task.url == null ? null : (typeof task.url === 'string' && /^https?:\/\//i.test(task.url) && task.url.length <= 2000 ? task.url : undefined);++ if (url === undefined) return null;++ return { source, ref, title, url };++}++++function cleanStart(value, now = Date.now()) {++ const at = new Date(String(value || ''));++ if (Number.isNaN(at.getTime())) throw new ScheduleError(400, 'invalid_starts_at', 'starts_at is a date-time');++ if (at.getTime() < now + 60 * 1000) throw new ScheduleError(400, 'starts_at_past', 'A schedule starts at least a minute from now');++ if (at.getTime() > now + SCHEDULE_WINDOW_MS) throw new ScheduleError(400, 'starts_at_far', 'A schedule is at most 30 days away');++ return at;++}++++const ID_RE = /^scd_[0-9A-HJKMNP-TV-Z]{26}$/;++const newId = () => 'scd_' + contracts.ids.ulid();++++const SELECT = `++ SELECT sc.*, u.username, u.display_name,++ la.subject_id AS owner_subject,++ ms.public_id AS agent_stream_id,++ (SELECT COUNT(*) FROM agent_schedule_reminders r WHERE r.schedule_id = sc.id) AS reminders,++ EXISTS(SELECT 1 FROM agent_schedule_reminders r WHERE r.schedule_id = sc.id AND r.subject_id = ?) AS reminded++ FROM agent_stream_schedules sc++ JOIN users u ON u.id = sc.user_id++ LEFT JOIN linked_accounts la ON la.service = 'network' AND la.user_id = sc.user_id++ LEFT JOIN managed_streams ms ON ms.id = sc.managed_stream_id`;++++function pageUrl(row) {++ const base = String(require('../config').baseUrl || 'https://openvibe.live').replace(/\/+$/, '');++ return `${base}/@${encodeURIComponent(row.username || '')}/agents`;++}++++function scheduleView(row, { subject = null } = {}) {++ if (!row) return null;++ return {++ id: row.id,++ owner: row.owner_subject || null,++ channel: row.username,++ title: row.title,++ task: typeof row.task === 'string' ? JSON.parse(row.task) : (row.task || {}),++ mode: row.mode,++ visibility: row.visibility,++ starts_at: service.iso(row.starts_at),++ state: row.state,++ agent_stream_id: row.agent_stream_id || null,++ missed_reason: row.missed_reason || null,++ waits_for_go: !!row.waits_for_go,++ started_at: service.iso(row.started_at),++ ended_at: service.iso(row.ended_at),++ reminders: Number(row.reminders) || 0,++ ...(subject ? { reminded: !!Number(row.reminded) } : {}),++ page_url: pageUrl(row),++ created_at: service.iso(row.created_at),++ };++}++++async function get(id, { subject = null } = {}) {++ if (!ID_RE.test(String(id || ''))) return null;++ const row = await db.get(`${SELECT} WHERE sc.id = ?`, [subject || null, id]);++ return row ? scheduleView(row, { subject }) : null;++}++++async function rowOf(id) {++ if (!ID_RE.test(String(id || ''))) return null;++ return await db.get('SELECT * FROM agent_stream_schedules WHERE id = ?', [id]) || null;++}++++/** A new schedule. The route has already checked the owner's permission; this checks the record's own rules. */++async function create({ user, title, task, mode = 'watch', startsAt }) {++ const cleanTitle = text(title, MAX_TITLE);++ if (!cleanTitle) throw new ScheduleError(400, 'invalid_title', 'A title of 1–200 characters is required');++ const clean = cleanTask(task);++ if (!clean) throw new ScheduleError(400, 'invalid_task', 'task needs source (codes, actor, desk, node), ref and title');++ if (!MODES.includes(mode)) throw new ScheduleError(400, 'invalid_mode', 'mode is watch, narrated, copilot or chatplays');++ const at = cleanStart(startsAt);++ const id = newId();++ await db.run(++ `INSERT INTO agent_stream_schedules (id, user_id, title, task, mode, visibility, starts_at)++ VALUES (?, ?, ?, ?::jsonb, ?, 'public', ?)`,++ [id, user.id, cleanTitle, JSON.stringify(clean), mode, at]++ );++ return await get(id, { subject: await service.ownerSubject(user.id) });++}++++/** The owner's own schedules, the next one first. */++async function listMine(userId, { subject = null } = {}) {++ const rows = await db.all(`${SELECT} WHERE sc.user_id = ? ORDER BY (sc.state = 'scheduled') DESC, sc.starts_at ASC LIMIT 100`, [subject || null, userId]);++ return rows.map((row) => scheduleView(row, { subject }));++}++++/**++ * The public schedules a page may show: public, still waiting, and inside the next 24 hours (the design's window).++ * The owner's own page also sees their whole list (the channel's Agents tab).++ */++async function listPublic({ channel = null, now = Date.now(), subject = null, limit = 100 } = {}) {++ const params = [subject || null, new Date(now), new Date(now + AHEAD_MS)];++ let where = "sc.visibility = 'public' AND sc.state = 'scheduled' AND sc.starts_at >= ? AND sc.starts_at <= ?";++ if (channel) { where += ' AND LOWER(u.username) = LOWER(?)'; params.push(String(channel)); }++ params.push(Math.min(200, Math.max(1, Number(limit) || 100)));++ const rows = await db.all(`${SELECT} WHERE ${where} ORDER BY sc.starts_at ASC, sc.id ASC LIMIT ?`, params);++ return rows.map((row) => scheduleView(row, { subject }));++}++++/** The owner's whole list plus the public window, for the channel Agents tab. */++async function listForChannel(username, viewerId, { subject = null } = {}) {++ const rows = await db.all(`${SELECT} WHERE LOWER(u.username) = LOWER(?) ORDER BY sc.starts_at ASC LIMIT 200`, [subject || null, String(username)]);++ return rows.filter((row) => Number(row.user_id) === Number(viewerId || 0)++ || (row.visibility === 'public' && row.state === 'scheduled' && Date.parse(service.iso(row.starts_at)) <= Date.now() + AHEAD_MS));++}++++/** Cancel one of the owner's schedules. A started or finished one is not cancellable here. */++async function cancel(id, user) {++ const row = await rowOf(id);++ if (!row) throw new ScheduleError(404, 'schedule_not_found', 'No such schedule');++ if (Number(row.user_id) !== Number(user.id)) throw new ScheduleError(403, 'not_your_schedule', 'This schedule belongs to another account');++ if (row.state !== 'scheduled') throw new ScheduleError(409, 'schedule_not_cancellable', 'This schedule already started or ended');++ await db.run("UPDATE agent_stream_schedules SET state = 'cancelled', updated_at = now() WHERE id = ?", [id]);++ return await get(id);++}++++/** Opt in to a reminder (idempotent). The answer is the count, never the names. */++async function remind(id, subject) {++ if (!/^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(subject || ''))) throw new ScheduleError(409, 'no_subject', 'This account has no OpenVibe subject yet; sign in again');++ const row = await rowOf(id);++ if (!row) throw new ScheduleError(404, 'schedule_not_found', 'No such schedule');++ // Only a schedule this person may see: a public one inside its 24-hour window (a private one's id is not an++ // invitation; the answer is the same 404 as a missing one).++ if (row.visibility !== 'public' || Date.parse(service.iso(row.starts_at)) > Date.now() + AHEAD_MS) throw new ScheduleError(404, 'schedule_not_found', 'No such schedule');++ if (row.state !== 'scheduled' || Date.parse(service.iso(row.starts_at)) < Date.now()) throw new ScheduleError(409, 'schedule_over', 'This schedule has already started or ended');++ await db.run('INSERT INTO agent_schedule_reminders (schedule_id, subject_id) VALUES (?, ?) ON CONFLICT (schedule_id, subject_id) DO NOTHING', [id, subject]);++ const n = Number((await db.get('SELECT COUNT(*) AS n FROM agent_schedule_reminders WHERE schedule_id = ?', [id]))?.n || 0);++ return { reminders: n, reminded: true };++}++++/** The desk says it is online. A heartbeat is a fact, not a stream: it carries no ticket. */++async function heartbeat(user, version = null) {++ await db.run(++ `INSERT INTO agent_desks (user_id, last_seen_at, version) VALUES (?, now(), ?)++ ON CONFLICT (user_id) DO UPDATE SET last_seen_at = now(), version = COALESCE(excluded.version, agent_desks.version)`,++ [user.id, typeof version === 'string' ? version.slice(0, 80) : null]++ );++ return { online: true, at: new Date().toISOString() };++}++++async function deskSeen(userId) {++ const row = await db.get('SELECT last_seen_at FROM agent_desks WHERE user_id = ?', [userId]);++ const at = row ? Date.parse(service.iso(row.last_seen_at)) : NaN;++ return Number.isFinite(at) ? at : null;++}++++/**++ * The sweep's part: a schedule whose time passed and that nobody claimed becomes missed. A chatplays one keeps its++ * time and says it waits for Go; nothing is started in either case.++ */++async function startDue({ now = Date.now() } = {}) {++ const rows = await db.all(++ "SELECT * FROM agent_stream_schedules WHERE state = 'scheduled' AND starts_at <= ? ORDER BY starts_at ASC LIMIT 200",++ [new Date(now - MISS_GRACE_MS)]++ );++ const out = { missed: 0, waiting: 0 };++ for (const row of rows) {++ if (row.mode === 'chatplays') {++ if (!row.waits_for_go || row.missed_reason !== 'waits_for_go') {++ await db.run("UPDATE agent_stream_schedules SET waits_for_go = true, missed_reason = 'waits_for_go', updated_at = now() WHERE id = ?", [row.id]);++ }++ out.waiting++;++ continue;++ }++ const seen = await deskSeen(row.user_id);++ const online = seen != null && now - seen <= DESK_ONLINE_MS;++ await db.run(++ "UPDATE agent_stream_schedules SET state = 'missed', missed_reason = ?, updated_at = now() WHERE id = ?",++ [online ? 'task_not_ready' : 'desk_offline', row.id]++ );++ out.missed++;++ }++ return out;++}++++/**++ * The owner's desk claims a due schedule: Live creates the stream and answers the one publish ticket. A chatplays++ * schedule is refused until the owner's Go; a schedule that is not due, or not the caller's, is refused too.++ */++async function claim(id, user, { go = false, now = Date.now() } = {}) {++ const row = await db.get(++ `SELECT sc.*, u.username, la.subject_id AS owner_subject FROM agent_stream_schedules sc++ JOIN users u ON u.id = sc.user_id++ LEFT JOIN linked_accounts la ON la.service = 'network' AND la.user_id = sc.user_id++ WHERE sc.id = ?`, [id]++ );++ if (!row) throw new ScheduleError(404, 'schedule_not_found', 'No such schedule');++ if (Number(row.user_id) !== Number(user.id)) throw new ScheduleError(403, 'not_your_schedule', 'This schedule belongs to another account');++ if (row.state !== 'scheduled') throw new ScheduleError(409, 'schedule_not_scheduled', 'This schedule already started or ended');++ if (Date.parse(service.iso(row.starts_at)) > now + 60 * 1000) throw new ScheduleError(409, 'schedule_not_due', 'This schedule is not due yet');++ if (row.mode === 'chatplays' && !go) {++ await db.run("UPDATE agent_stream_schedules SET waits_for_go = true, missed_reason = 'waits_for_go', updated_at = now() WHERE id = ?", [id]);++ throw new ScheduleError(409, 'waits_for_go', 'A chat plays schedule waits for the owner\'s Go');++ }++ // The stream is the schedule's, and only one stream may come from it: the row flips first, in one statement.++ const claimed = await db.run(++ `UPDATE agent_stream_schedules SET state = 'live', started_at = now(), waits_for_go = false, missed_reason = NULL, updated_at = now()++ WHERE id = ? AND state = 'scheduled' RETURNING id`, [id]++ );++ if (!claimed.changes) throw new ScheduleError(409, 'schedule_not_scheduled', 'This schedule already started or ended');++ let stream;++ let publish;++ try {++ stream = await service.create({++ user, title: row.title, task: typeof row.task === 'string' ? JSON.parse(row.task) : row.task,++ visibility: 'public', mode: row.mode,++ });++ publish = await require('./tickets').issuePublish(stream.managed_stream_id);++ await db.run('UPDATE agent_stream_schedules SET managed_stream_id = ?, updated_at = now() WHERE id = ?', [stream.managed_stream_id, id]);++ } catch (err) {++ await db.run("UPDATE agent_stream_schedules SET state = 'scheduled', started_at = NULL, updated_at = now() WHERE id = ? AND state = 'live' AND managed_stream_id IS NULL", [id]);++ throw err;++ }++ return {++ schedule: await get(id, { subject: row.owner_subject }),++ stream: service.publicView(stream),++ publish: { whip_url: service.whipUrl(stream.public_id), ...publish },++ };++}++++/** The stream a schedule started ended with it. Called from service.end, in the same request. */++async function markEnded(managedStreamId) {++ if (!managedStreamId) return;++ await db.run("UPDATE agent_stream_schedules SET state = 'ended', ended_at = now(), updated_at = now() WHERE managed_stream_id = ? AND state = 'live'", [managedStreamId]);++}++++module.exports = {++ AHEAD_MS, DESK_ONLINE_MS, MISS_GRACE_MS, SCHEDULE_WINDOW_MS, ScheduleError,++ cleanTask, newId, get, create, listMine, listPublic, listForChannel, cancel, remind, heartbeat, deskSeen, startDue, claim, markEnded,++ scheduleView,++};@@@ -83,10 +83,12 @@ const SELECT = `(SELECT st.title FROM agent_stream_steps st WHERE st.managed_stream_id = ms.id ORDER BY st.id DESC LIMIT 1) AS latest_step_title,(SELECT st.status FROM agent_stream_steps st WHERE st.managed_stream_id = ms.id ORDER BY st.id DESC LIMIT 1) AS latest_step_status,(SELECT la.subject_id FROM linked_accounts la WHERE la.service = 'network' AND la.user_id = ms.user_id) AS owner_subject,-- u.username, u.display_name, u.avatar_url, u.profile_color, u.is_banned, u.is_owner++ u.username, u.display_name, u.avatar_url, u.profile_color, u.is_banned, u.is_owner,++ c.bot_robot_idFROM managed_streams msJOIN agent_streams a ON a.managed_stream_id = ms.id-- JOIN users u ON u.id = ms.user_id`;++ JOIN users u ON u.id = ms.user_id++ LEFT JOIN channels c ON c.id = ms.channel_id`;async function getByPublicId(publicId) {if (!AGENT_ID_RE.test(String(publicId || ''))) return null;@@@ -167,6 +169,9 @@ function publicView(row) {label: 'AI agent',// The chat room a public or unlisted stream carries (S2): people only, the agent reads it in S5.chat_room: (row.visibility === 'public' || row.visibility === 'unlisted') ? `agent:${row.public_id}` : null,++ // S7: a channel bound to a Bot robot can show the robot's camera as a second scene beside the desk. The++ // binding and Bot's own embed_public decide whether it appears; Live only names the panel.++ robot_scene: robotScene(row),};}@@@ -196,6 +201,19 @@ function cardView(row) {};}++/** The robot camera scene for a channel bound to a Bot robot, or null (S7). */++function robotScene(row) {++ try {++ const bot = require('../bot/embed');++ if (!bot.enabled() || !bot.validRobotId(row.bot_robot_id)) return null;++ return {++ robot_id: row.bot_robot_id,++ label: 'Robot camera',++ embed_url: bot.embedUrl(row.bot_robot_id),++ };++ } catch { return null; }++}++/** Public agent streams (live or ended), newest first, for the home row and a channel's Agents tab. */async function listPublic({ channel = null, liveOnly = false, limit = 60 } = {}) {const params = [];@@@ -309,6 +327,9 @@ async function end(row, { reason = 'ended' } = {}) {event: { kind: 'ended', extra: { reason } },});await require('./tickets').revokeAll(row.managed_stream_id);++ // S7: the schedule that started this stream ends with it, and a robot the channel is bound to is stopped.++ try { await require('./schedules').markEnded(row.managed_stream_id); } catch (err) { console.warn('[AgentStream] schedule end: ' + err.message); }++ try { require('../bot/kill').stop(row); } catch (err) { console.warn('[AgentStream] bot stop: ' + err.message); }if (session) {try {const whip = require('../streaming/whip-handler');@@@ -47,6 +47,13 @@ async function sweep() {}} catch (err) { console.warn('[AgentStream] idle voice failed:', err.message); }++ // S7: a due schedule nobody claimed is marked missed ('didn't start: the desk was offline'); a chatplays one++ // keeps its time and waits for the owner's Go. Nothing is started here.++ try {++ const out = await require('./schedules').startDue();++ if (out.missed || out.waiting) console.warn(`[AgentStream] schedules: ${out.missed} missed, ${out.waiting} waiting for Go`);++ } catch (err) { console.warn('[AgentStream] schedule sweep failed:', err.message); }++try { await tickets.prune(); } catch (err) { console.warn('[AgentStream] ticket prune failed:', err.message); }try {const sharing = require('./sharing');nnew file mode 100644@@@ -0,0 +1,57 @@++'use strict';++/**++ * The kill switch's reach into OpenVibe.Bot (Agent Streams S7, design 2026-11-06 §6).++ *++ * When an agent stream ends — the owner's End, a moderator's end control, a ticket revoke or the sweep — and the++ * stream's channel is bound to a robot (`channels.bot_robot_id`), Live tells Bot to e_stop that robot. Bot's own rules++ * still hold: its owner may always stop it, and Live's internal call carries the deployment's shared secret. With++ * `BOT_KILL_URL` unset (and in tests) this is a no-op: a stream ending never fails because Bot is down.++ *++ * This is not a second control path for chat: nothing in Chat, and neither S5 nor S6, reaches this. Only the owner's++ * kill switch (and the stream ending, which the owner can press) does.++ */++const db = require('../db/database');++++const URL_BASE = () => String(process.env.BOT_KILL_URL || '').trim().replace(/\/+$/, '');++const SECRET = () => String(process.env.BOT_KILL_SECRET || '');++const TIMEOUT_MS = 2500;++++function configured() { return !!URL_BASE() && !!SECRET(); }++++/** The robot bound to this agent stream's channel, or null. */++async function robotFor(row) {++ if (!row || !row.channel_id) return null;++ try {++ const channel = await db.get('SELECT bot_robot_id FROM channels WHERE id = ?', [row.channel_id]);++ const id = channel && String(channel.bot_robot_id || '');++ return /^rob_[A-Za-z0-9_-]{4,60}$/.test(id) ? id : null;++ } catch { return null; }++}++++/**++ * Stop the robot bound to this stream's channel, if there is one and Bot is configured. Resolves to++ * { stopped, robot_id } either way; a Bot failure is logged, never thrown into the stream's end.++ */++async function stop(row, { fetchImpl = globalThis.fetch } = {}) {++ const robotId = await robotFor(row);++ if (!robotId) return { stopped: false, robot_id: null };++ if (!configured()) return { stopped: false, robot_id: robotId, reason: 'bot_kill_unconfigured' };++ try {++ const res = await fetchImpl(`${URL_BASE()}/internal/robots/${encodeURIComponent(robotId)}/estop`, {++ method: 'POST',++ headers: { 'Content-Type': 'application/json', 'X-OV-Bot-Kill': SECRET() },++ body: JSON.stringify({ reason: 'agent_stream_ended', stream_id: row.public_id || null }),++ signal: AbortSignal.timeout(TIMEOUT_MS),++ });++ if (!res.ok) {++ console.warn(`[AgentStream] Bot did not stop ${robotId}: ${res.status}`);++ return { stopped: false, robot_id: robotId, reason: `bot_${res.status}` };++ }++ return { stopped: true, robot_id: robotId };++ } catch (err) {++ console.warn(`[AgentStream] Bot stop of ${robotId} failed: ${err.message}`);++ return { stopped: false, robot_id: robotId, reason: 'bot_unreachable' };++ }++}++++module.exports = { stop, robotFor, configured, URL_BASE };@@@ -199,6 +199,17 @@ async function readSoundByCommand(params) {} catch (err) { note(`sound-by-command: ${err.message}`); }return null;}++/** How many listed watch parties a public agent stream has (S7). The answer is a count; Chat never names the people. */++async function partyCount(agentStreamId) {++ if (!available()) return null;++ try {++ const { status, data } = await request('GET', `/internal/chat/parties/count${qs({ agent_stream: agentStreamId })}`);++ if (status === 200 && data && data.ok !== false) return data;++ note(`party-count: Chat answered ${status}${data && data.error ? ` (${data.error})` : ''}`);++ } catch (err) { note(`party-count: ${err.message}`); }++ return null;++}++/** Record where Live's asset sync uploaded a channel sound (idempotent; Chat applies the same write once). */async function soundAsset(body) {if (!available()) return null;@@@ -231,6 +242,6 @@ module.exports = {invalidate: async (body) => await send('invalidate', body),presence,readStats, readMessages, readTimeline, readFirstChat, readPendingIp, readRelayUsers, readRelayUser, readTtsOverride,-- readSounds, readSoundByCommand, readDmBlockState, soundAsset, agentRoom,++ readSounds, readSoundByCommand, readDmBlockState, soundAsset, agentRoom, partyCount,_setRetryMs(ms) { retryMs = ms || RETRY_MS; },};nnew file mode 100644@@@ -0,0 +1,67 @@++'use strict';++/**++ * Agent report citations on a Rundown topic page (Agent Streams S7, design 2026-11-06 §2).++ *++ * The Rundown's topic pages may cite a public session report when an agent worked on the topic. Only Everyone's++ * reports are read (Report's public list asks the sharing system), and the topic's whole name must appear as a whole++ * phrase — the same whole-name rule the streamers' write-ups keep. A Report outage is an empty list, never a broken++ * topic page.++ *++ * The read is cached for five minutes per topic: Report answers its public list for everyone, and a topic page is++ * refreshed by the same reader many times a day.++ */++const REPORT_URL = () => String(process.env.OV_REPORT_URL || 'https://openvibe.report').replace(/\/+$/, '');++const TTL_MS = 5 * 60_000;++const cache = new Map(); // topic phrase -> { at, list, expires }++++function normalize(value) {++ return String(value || '').toLowerCase().normalize('NFKD').replace(/[^a-z0-9]+/g, ' ').trim();++}++++/** Whether this report speaks of the topic by its whole name (never a partial word). */++function speaksOf(report, topic) {++ const phrase = normalize(topic);++ if (!phrase) return false;++ const text = [report && report.task && report.task.title, report && report.summary, ...(report && Array.isArray(report.highlights) ? report.highlights : [])]++ .map(normalize).filter(Boolean).join(' ');++ return new RegExp(`(^| )${phrase.replace(/[.*+?^${}()|[\]\\]/g, '\\$&')}( |$)`).test(text);++}++++function cite(report) {++ return {++ id: report.id,++ title: (report.task && report.task.title) || report.summary || 'An agent session',++ summary: report.summary || null,++ outcome: report.outcome || 'done',++ worked_seconds: Number(report.worked_seconds) || 0,++ };++}++++function createCitations({ fetchImpl = globalThis.fetch, log = console, now = () => Date.now(), url = REPORT_URL(), ttl = TTL_MS } = {}) {++ async function forTopic(topic, { limit = 6 } = {}) {++ const phrase = normalize(topic);++ if (!phrase) return [];++ const hit = cache.get(phrase);++ if (hit && hit.expires > now()) return hit.list.slice(0, limit);++ let list = [];++ try {++ const res = await fetchImpl(`${url}/api/v1/session-reports`, {++ headers: { Accept: 'application/json' },++ signal: AbortSignal.timeout(2500),++ });++ if (!res.ok) throw new Error(`report ${res.status}`);++ const body = await res.json().catch(() => null);++ const reports = body && Array.isArray(body.reports) ? body.reports : [];++ list = reports.filter((r) => speaksOf(r, phrase)).map(cite);++ } catch (err) {++ log.warn(`[Rundown] agent citations for "${phrase}" unavailable: ${(err && err.message) || err}`);++ list = [];++ }++ cache.set(phrase, { at: now(), expires: now() + ttl, list });++ if (cache.size > 1000) cache.clear();++ return list.slice(0, Math.max(1, Math.min(20, Number(limit) || 6)));++ }++ return { forTopic, clear: () => cache.clear() };++}++++module.exports = { createCitations, speaksOf, normalize, TTL_MS };@@@ -30,6 +30,7 @@ const read = require('./read');const browse = require('./browse');const settings = require('./settings');const del = require('./delete');++const { createCitations } = require('./citations');const router = express.Router();const fail = (res, err, msg) => {@@@ -79,8 +80,14 @@ router.get('/topics', optionalAuth, async (req, res) => {} catch (err) { fail(res, err, 'Failed to load the topics'); }});router.get('/t/:slug', optionalAuth, async (req, res) => {-- try { const t = await read.topic(req.params.slug, { viewer: await viewerOf(req) }); if (!t) return res.status(404).json({ error: 'No such topic' }); fresh(res); res.json(t); }-- catch (err) { fail(res, err, 'Failed to load the topic'); }++ try {++ const t = await read.topic(req.params.slug, { viewer: await viewerOf(req) });++ if (!t) return res.status(404).json({ error: 'No such topic' });++ // S7: a public agent report may cite the topic when the agent worked on it. Report being down is no citations.++ t.citations = await createCitations().forTopic((t.topic && t.topic.title) || t.title);++ fresh(res);++ res.json(t);++ } catch (err) { fail(res, err, 'Failed to load the topic'); }});router.get('/e/:streamId', optionalAuth, async (req, res) => {try { const e = await read.episode(req.params.streamId, { viewer: await viewerOf(req) }); if (!e) return res.status(404).json({ error: 'No such episode' }); fresh(res); res.json(e); }nnew file mode 100644@@@ -0,0 +1,298 @@++'use strict';++/**++ * Agent streams S7, Live's part (design 2026-11-06): schedules, the desk heartbeat, the miss and the Go; the party++ * count on a stream; and the robot camera as a second scene.++ *++ * - a schedule shows on the channel's Agents tab 24 hours before its time and nowhere earlier;++ * - reminders are opt-in, and the answer is a count, never a name;++ * - a due schedule nobody claimed says "didn't start: the desk was offline" and starts nothing;++ * - an online desk claims it and gets the one publish ticket; a chatplays schedule waits for the owner's Go;++ * - the stream page carries the party count (a number) and, on a channel bound to a robot, the robot scene.++ *++ * Sign-in is stubbed with an x-test-user header, as agent-streams-s2.test.js does; everything after that is production.++ */++const assert = require('assert');++const fs = require('node:fs');++const path = require('node:path');++const http = require('http');++const express = require('express');++++process.env.NODE_ENV = 'test';++const quiet = console.log;++console.log = (...a) => { if (!/^\[/.test(String(a[0]))) quiet(...a); };++console.warn = () => {};++console.error = () => {};++++const SID = {++ ann: 'usr_01JAB2C3D4E5F6G7H8J9K0MNA1',++ bob: 'usr_01JAB2C3D4E5F6G7H8J9K0MNB2',++};++++let failures = 0;++async function check(name, fn) {++ try { await fn(); quiet(' ✓', name); }++ catch (e) { failures++; quiet(' ✗', name, '\n ', (e.stack || String(e)).split('\n').slice(0, 8).join('\n ')); }++}++++(async () => {++ const db = require('../server/db/database');++ await db.initDb();++ const raw = db.getDb();++ const addUser = async (id, username, role) => await raw.prepare(++ `INSERT INTO users (id, username, display_name, email, password_hash, role, is_owner) OVERRIDING SYSTEM VALUE++ VALUES (?, ?, ?, ?, 'x', ?, 0)`).run(id, username, username, `${username}@x`, role);++ await addUser(1, 'ann', 'streamer');++ await addUser(2, 'bob', 'user');++ await raw.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (1, 'network', '101', ?)").run(SID.ann);++ await raw.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (2, 'network', '102', ?)").run(SID.bob);++++ const config = require('../server/config');++ config.baseUrl = 'https://openvibe.live';++ const service = require('../server/agent-streams/service');++ const schedules = require('../server/agent-streams/schedules');++ require('../server/agent-streams/sharing')._setNetwork({ json: async () => ({}) });++ const ann = await db.getUserById(1);++ const bob = await db.getUserById(2);++ const channel = await db.ensureChannel(1);++++ const app = express();++ app.use(express.json());++ app.use((req, res, next) => { req.user = { 1: ann, 2: bob }[req.get('x-test-user')] || null; next(); });++ app.use('/api/streams/agent', require('../server/agent-streams/routes'));++ const server = await new Promise((ok) => { const s = http.createServer(app).listen(0, '127.0.0.1', () => ok(s)); });++ const base = `http://127.0.0.1:${server.address().port}`;++ const call = async (method, url, { as = 1, body, headers = {} } = {}) => {++ const res = await fetch(`${base}${url}`, {++ method,++ headers: { 'Content-Type': 'application/json', 'x-test-user': String(as), ...headers },++ body: body === undefined ? undefined : JSON.stringify(body),++ });++ const text = await res.text();++ let data = {};++ try { data = text ? JSON.parse(text) : {}; } catch { data = { text }; }++ return { status: res.status, data, headers: res.headers };++ };++ const makeSchedule = async ({ title = 'Fix the login page', at = Date.now() + 2 * 60 * 60 * 1000, mode = 'watch' } = {}) => {++ const r = await call('POST', '/api/streams/agent/schedules', {++ body: {++ title,++ task: { source: 'codes', ref: 'tsk_1', title: 'Fix the login page', url: 'https://openvibe.codes/tasks/1' },++ mode,++ starts_at: new Date(at).toISOString(),++ },++ });++ assert.strictEqual(r.status, 201, JSON.stringify(r.data));++ return r.data.schedule;++ };++ const due = async (id) => await db.run("UPDATE agent_stream_schedules SET starts_at = now() - interval '2 minutes' WHERE id = ?", [id]);++++ quiet('agent streams S7: schedules');++ const soon = await makeSchedule({ title: 'Soon', at: Date.now() + 23 * 60 * 60 * 1000 });++ const later = await makeSchedule({ title: 'Later', at: Date.now() + 25 * 60 * 60 * 1000 });++ await check('a schedule answers the contract shape', async () => {++ assert.match(soon.id, /^scd_[0-9A-HJKMNP-TV-Z]{26}$/);++ assert.strictEqual(soon.owner, SID.ann);++ assert.strictEqual(soon.channel, 'ann');++ assert.strictEqual(soon.visibility, 'public');++ assert.strictEqual(soon.state, 'scheduled');++ assert.strictEqual(soon.reminders, 0);++ assert.ok(soon.page_url.endsWith('/@ann/agents'));++ });++ await check('Today sees it 24 hours ahead, not a day early', async () => {++ const r = await call('GET', '/api/streams/agent/schedules?channel=ann', { as: 2 });++ assert.strictEqual(r.status, 200, JSON.stringify(r.data));++ const ids = r.data.schedules.map((s) => s.id);++ assert.ok(ids.includes(soon.id), '23 hours ahead is shown');++ assert.ok(!ids.includes(later.id), '25 hours ahead is not');++ const mine = await call('GET', '/api/streams/agent/schedules?mine=1', { as: 1 });++ assert.ok(mine.data.schedules.some((s) => s.id === later.id), 'the owner sees their whole list');++ });++ await check('reminders are opt-in only, and the answer is a count', async () => {++ const before = await call('GET', `/api/streams/agent/schedules/${soon.id}`, { as: 2 });++ assert.strictEqual(before.data.schedule.reminded, false);++ const r = await call('POST', `/api/streams/agent/schedules/${soon.id}/remind`, { as: 2 });++ assert.strictEqual(r.status, 200, JSON.stringify(r.data));++ assert.strictEqual(r.data.reminders, 1);++ assert.strictEqual(r.data.reminded, true);++ const asAnn = await call('GET', `/api/streams/agent/schedules/${soon.id}`, { as: 1 });++ assert.strictEqual(asAnn.data.schedule.reminded, false, 'ann did not opt in');++ assert.strictEqual(asAnn.data.schedule.reminders, 1, 'the count is there');++ const all = JSON.stringify(asAnn.data);++ assert.ok(!all.includes('bob'), 'no name in a schedule view');++ const signedOut = await call('POST', `/api/streams/agent/schedules/${soon.id}/remind`, { as: 0 });++ assert.strictEqual(signedOut.status, 401);++ // Only a schedule the person may see: one beyond the 24-hour window, or a private one, is a 404.++ const tooEarly = await call('POST', `/api/streams/agent/schedules/${later.id}/remind`, { as: 2 });++ assert.strictEqual(tooEarly.status, 404, 'a schedule outside its public window is not offered');++ await db.run("UPDATE agent_stream_schedules SET visibility = 'private' WHERE id = ?", [soon.id]);++ const hidden = await call('POST', `/api/streams/agent/schedules/${soon.id}/remind`, { as: 2 });++ assert.strictEqual(hidden.status, 404, "a private schedule's id is not an invitation");++ await db.run("UPDATE agent_stream_schedules SET visibility = 'public' WHERE id = ?", [soon.id]);++ });++ await check('a due schedule with the desk offline says so and starts nothing', async () => {++ const missed = await makeSchedule({ title: 'Offline desk' });++ await due(missed.id);++ const streamsBefore = (await service.listByUser(1)).length;++ const out = await schedules.startDue();++ assert.ok(out.missed >= 1, JSON.stringify(out));++ const after = await call('GET', `/api/streams/agent/schedules/${missed.id}`, { as: 1 });++ assert.strictEqual(after.data.schedule.state, 'missed');++ assert.strictEqual(after.data.schedule.missed_reason, 'desk_offline');++ assert.strictEqual(after.data.schedule.agent_stream_id, null);++ assert.strictEqual((await service.listByUser(1)).length, streamsBefore, 'nothing ran');++ });++ await check('an online desk claims it and gets one publish ticket; ending the stream ends the schedule', async () => {++ const ready = await makeSchedule({ title: 'Online desk' });++ await schedules.heartbeat(ann, 'node/0.41.0');++ await due(ready.id);++ const r = await call('POST', `/api/streams/agent/schedules/${ready.id}/claim`, { as: 1 });++ assert.strictEqual(r.status, 201, JSON.stringify(r.data));++ assert.match(r.data.publish.ticket, /^wpt_/);++ assert.match(r.data.publish.whip_url, /\/whip\/agent\/ags_/);++ assert.strictEqual(r.data.schedule.state, 'live');++ assert.strictEqual(r.data.stream.id, r.data.schedule.agent_stream_id);++ await service.markPublishing(await service.getByPublicId(r.data.stream.id));++ const ended = await call('POST', `/api/streams/agent/${r.data.stream.id}/end`, { as: 1 });++ assert.strictEqual(ended.status, 200, JSON.stringify(ended.data));++ const after = await call('GET', `/api/streams/agent/schedules/${ready.id}`, { as: 1 });++ assert.strictEqual(after.data.schedule.state, 'ended');++ assert.ok(after.data.schedule.ended_at);++ });++ await check('a chat plays schedule waits for the owner\'s Go and starts nothing before it', async () => {++ const plays = await makeSchedule({ title: 'Chat plays a tiny game', mode: 'chatplays' });++ await schedules.heartbeat(ann);++ await due(plays.id);++ const before = (await service.listByUser(1)).length;++ const refused = await call('POST', `/api/streams/agent/schedules/${plays.id}/claim`, { as: 1 });++ assert.strictEqual(refused.status, 409, JSON.stringify(refused.data));++ assert.strictEqual(refused.data.code, 'waits_for_go');++ const waiting = await call('GET', `/api/streams/agent/schedules/${plays.id}`, { as: 1 });++ assert.strictEqual(waiting.data.schedule.waits_for_go, true);++ assert.strictEqual(waiting.data.schedule.missed_reason, 'waits_for_go');++ assert.strictEqual(waiting.data.schedule.state, 'scheduled');++ assert.strictEqual((await service.listByUser(1)).length, before, 'nothing ran without Go');++ const go = await call('POST', `/api/streams/agent/schedules/${plays.id}/go`, { as: 1 });++ assert.strictEqual(go.status, 201, JSON.stringify(go.data));++ assert.strictEqual(go.data.schedule.state, 'live');++ });++ await check('only the owner may cancel; a cancelled schedule leaves the window', async () => {++ const own = await makeSchedule({ title: 'Cancel me' });++ const other = await call('DELETE', `/api/streams/agent/schedules/${own.id}`, { as: 2 });++ assert.strictEqual(other.status, 403);++ const ok = await call('DELETE', `/api/streams/agent/schedules/${own.id}`, { as: 1 });++ assert.strictEqual(ok.status, 200, JSON.stringify(ok.data));++ assert.strictEqual(ok.data.schedule.state, 'cancelled');++ });++++ quiet('agent streams S7: parties and the robot scene');++ const partyStream = await service.create({++ user: ann, title: 'A party stream', task: { source: 'codes', ref: 'tsk_party', title: 'A party' }, visibility: 'public',++ });++ await check('the stream carries the party count and never a name', async () => {++ const chat = require('../server/chat/chat-client');++ const original = chat.partyCount;++ chat.partyCount = async () => ({ ok: true, count: 3 });++ try {++ const r = await call('GET', `/api/streams/agent/${partyStream.public_id}`, { as: 2 });++ assert.strictEqual(r.status, 200, JSON.stringify(r.data));++ assert.deepStrictEqual(r.data.parties, { count: 3 });++ const text = JSON.stringify(r.data);++ assert.ok(!/host|people|names|members/i.test(text), 'no party member list');++ } finally { chat.partyCount = original; }++ process.env.OV_CHAT_INTERNAL_URL = '';++ const noChat = await call('GET', `/api/streams/agent/${partyStream.public_id}`, { as: 2 });++ assert.strictEqual(noChat.data.parties, null, 'Chat down: no count, the stream still answers');++ process.env.OV_CHAT_INTERNAL_URL = 'http://127.0.0.1:4400';++ });++ await check('a channel bound to a robot shows the robot camera as a second scene', async () => {++ process.env.LIVE_BOT_EMBED = '1';++ try {++ await raw.prepare('UPDATE channels SET bot_robot_id = ? WHERE id = ?').run('rob_sim123', channel.id);++ const r = await call('GET', `/api/streams/agent/${partyStream.public_id}`, { as: 2 });++ assert.strictEqual(r.status, 200, JSON.stringify(r.data));++ assert.ok(r.data.stream.robot_scene, 'the robot scene is offered');++ assert.strictEqual(r.data.stream.robot_scene.robot_id, 'rob_sim123');++ assert.match(r.data.stream.robot_scene.embed_url, /\/panel\/rob_sim123\/embed$/);++ const fragment = fs.readFileSync(path.join(__dirname, '../public/fragments/agent-stream.html'), 'utf8');++ assert.match(fragment, /id="agent-scenes"/);++ assert.match(fragment, /id="agent-robot"/);++ const js = fs.readFileSync(path.join(__dirname, '../public/js/agent-stream.js'), 'utf8');++ assert.match(js, /function renderScenes/);++ assert.match(js, /robot_scene/);++ const css = fs.readFileSync(path.join(__dirname, '../public/css/features/agent-stream.css'), 'utf8');++ assert.match(css, /\.agent-robot/);++ } finally {++ await raw.prepare('UPDATE channels SET bot_robot_id = NULL WHERE id = ?').run(channel.id);++ delete process.env.LIVE_BOT_EMBED;++ }++ });++ await check('the kill switch reaches the robot through Bot', async () => {++ process.env.BOT_KILL_URL = 'https://bot.test';++ process.env.BOT_KILL_SECRET = 'kill-secret';++ try {++ await raw.prepare('UPDATE channels SET bot_robot_id = ? WHERE id = ?').run('rob_sim123', channel.id);++ const calls = [];++ const out = await require('../server/bot/kill').stop(++ { channel_id: channel.id, public_id: partyStream.public_id },++ { fetchImpl: async (url, opts) => { calls.push({ url, opts }); return { ok: true, status: 200 }; } }++ );++ assert.deepStrictEqual(out, { stopped: true, robot_id: 'rob_sim123' });++ assert.strictEqual(calls.length, 1);++ assert.strictEqual(calls[0].url, 'https://bot.test/internal/robots/rob_sim123/estop');++ assert.strictEqual(calls[0].opts.headers['X-OV-Bot-Kill'], 'kill-secret');++ assert.strictEqual(JSON.parse(calls[0].opts.body).reason, 'agent_stream_ended');++ } finally {++ await raw.prepare('UPDATE channels SET bot_robot_id = NULL WHERE id = ?').run(channel.id);++ delete process.env.BOT_KILL_URL;++ delete process.env.BOT_KILL_SECRET;++ }++ });++++ await check('the channel Agents tab draws the schedules above the cards', async () => {++ const fragment = fs.readFileSync(path.join(__dirname, '../public/fragments/channel.html'), 'utf8');++ assert.match(fragment, /id="ch-agents-schedules"/);++ const cards = fs.readFileSync(path.join(__dirname, '../public/js/agent-cards.js'), 'utf8');++ assert.match(cards, /scheduleCard/);++ assert.match(cards, /Didn't start: the desk was offline/);++ const channelJs = fs.readFileSync(path.join(__dirname, '../public/js/app-channel.js'), 'utf8');++ assert.match(channelJs, /fillSchedules/);++ });++ await check('a Rundown topic cites a public agent report by its whole name only', async () => {++ const { createCitations, speaksOf } = require('../server/rundown/citations');++ assert.ok(speaksOf({ task: { title: 'Fix the login page' }, summary: 'Done.' }, 'Login page'));++ assert.ok(!speaksOf({ task: { title: 'Fix the login page' } }, 'login pag'), 'a partial name is not the topic');++ assert.ok(!speaksOf({ task: { title: 'Fix the login page' } }, 'ogin'), 'a name inside a word is not the topic');++ const asked = [];++ const fetchImpl = async (url) => {++ asked.push(url);++ return { ok: true, json: async () => ({ reports: [++ { id: 'rps_01JAB2C3D4E5F6G7H8J9K0MNPS', task: { title: 'Fix the login page' }, summary: 'Done.', outcome: 'done' },++ { id: 'rps_01JAB2C3D4E5F6G7H8J9K0MNPT', task: { title: 'Fix the checkout page' }, summary: 'Done.', outcome: 'done' },++ ] }) };++ };++ const cites = createCitations({ fetchImpl, url: 'https://report.test', log: { warn() {} } });++ const first = await cites.forTopic('Login page');++ assert.deepStrictEqual(first.map((c) => c.id), ['rps_01JAB2C3D4E5F6G7H8J9K0MNPS']);++ assert.strictEqual(first[0].title, 'Fix the login page');++ await cites.forTopic('Login page');++ assert.strictEqual(asked.length, 1, 'the topic is cached');++ assert.deepStrictEqual(await cites.forTopic('Login page'), first);++ const down = createCitations({ fetchImpl: async () => { throw new Error('down'); }, url: 'https://report.test', log: { warn() {} } });++ down.clear();++ assert.deepStrictEqual(await down.forTopic('Login page'), [], 'Report down: no citations, no broken page');++ const javascript = fs.readFileSync(path.join(__dirname, '../public/js/rundown.js'), 'utf8');++ assert.match(javascript, /Agents that worked on this/);++ assert.match(javascript, /openvibe\.report\/s\//);++ });++++ await check('the schedule card survives 390 px', async () => {++ const css = fs.readFileSync(path.join(__dirname, '../public/css/features/agent-stream.css'), 'utf8');++ assert.match(css, /@media \(max-width: 640px\)[\s\S]*agent-scene/);++ assert.match(css, /\.agent-schedules \{ display: grid/);++ });++++ server.close();++ await db.closeDb?.();++ quiet(failures ? `\n${failures} check(s) failed` : '\nAll agent stream S7 checks passed');++ process.exit(failures ? 1 : 0);++})().catch((e) => { quiet(e.stack || String(e)); process.exit(1); });@@@ -351,6 +351,7 @@ async function check(name, fn) {'server/social/preview.js': 'OpenVibe.Network public profile API (configured OV_NETWORK_INTERNAL_URL, fixed path, username checked); every other preview goes through egress','server/monetization/inventory-client.js': 'OpenVibe.Inventory API for Live cosmetics (configured INVENTORY_URL)','server/bot/bot-client.js': 'OpenVibe.Bot command API for a bound channel\'s chat robot commands (configured LIVE_BOT_INTERNAL_URL)',++ 'server/bot/kill.js': 'OpenVibe.Bot internal e-stop for the robot a killed agent stream is bound to (configured BOT_KILL_URL, fixed path, robot id checked)','server/cam/cam-client.js': 'OpenVibe.Cam stills API for the cam button, as the viewer (configured CAM_URL)','server/chat/insight-client.js': 'OpenVibe.Chat public chat-AI reads (configured OV_CHAT_INTERNAL_URL, fixed paths, ids validated)','server/chat/chat-client.js': 'OpenVibe.Chat typed ingress (configured OV_CHAT_INTERNAL_URL, fixed /internal/chat/* paths)',@@@ -360,6 +361,7 @@ async function check(name, fn) {'server/auth/identity-sync.js': 'OpenVibe.Network (configured)','server/auth/account-data.js': 'OpenVibe.Network (configured): export parts and deletion confirmations','server/agent-streams/sharing.js': 'OpenVibe.Network (configured): the sharing system\'s name lookup and person resolution',++ 'server/rundown/citations.js': 'OpenVibe.Report public session-report list (OV_REPORT_URL, configured; fixed path, no URL from a person)','server/ai/viewer-quota.js': 'OpenVibe.AI (configured OV_AI_INTERNAL_URL): a streamer\'s daily AI-viewer cap','server/ai/byo-credentials.js': 'OpenVibe.AI (configured OV_AI_INTERNAL_URL): a streamer\'s own key stored there','server/monetization/billing-client.js': 'OpenVibe.Billing (configured)',
new file mode 100644
new file mode 100644
@@ -0,0 +1,53 @@
@@ -0,0 +1,53 @@
+-- phase: expand
+-- Agent Streams S7 (design 2026-11-06, AGENT-STREAMS-S7-REACH-DESIGN): schedules for public agent streams.
+--
+-- An owner schedules a public agent stream: a time, a task template and the mode. The schedule shows on the
+-- channel's Agents tab and in Today 24 hours before its time. When the time comes, the owner's desk claims it: Live
+-- creates the stream and hands the desk its publish ticket. A desk that is not online does not claim, and Live
+-- records the schedule as missed with 'desk_offline' ("didn't start: the desk was offline") and starts nothing. A
+-- chatplays schedule never starts on its own; it waits for the owner's Go (the S6 rule).
+--
+-- agent_desks is the desk heartbeat: a desk says it is online every 30 s, so the sweep can tell 'the desk was
+-- offline' apart from 'the desk was online but the task was not ready'. A heartbeat is a fact, not a stream: it
+-- carries no ticket and no secret.
+--
+-- agent_schedule_reminders is opt-in: a follower who asks for a reminder gets one Today notification. The row names
+-- the person, and the schedule's public view carries only the count.
+--
+-- Expand only: every table is new and an older release ignores all of it.
+CREATE TABLE IF NOT EXISTS agent_stream_schedules (
+ id text COLLATE "C" PRIMARY KEY, -- scd_<ULID>
+ user_id bigint NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+ managed_stream_id bigint REFERENCES managed_streams(id) ON DELETE SET NULL,
+ title text COLLATE "C" NOT NULL,
+ task jsonb NOT NULL,
+ mode text COLLATE "C" NOT NULL DEFAULT 'watch' CHECK (mode IN ('watch', 'narrated', 'copilot', 'chatplays')),
+ visibility text COLLATE "C" NOT NULL DEFAULT 'public' CHECK (visibility IN ('private', 'unlisted', 'public')),
+ starts_at timestamptz NOT NULL,
+ state text COLLATE "C" NOT NULL DEFAULT 'scheduled' CHECK (state IN ('scheduled', 'live', 'ended', 'missed', 'cancelled')),
+ missed_reason text COLLATE "C" CHECK (missed_reason IN ('desk_offline', 'task_not_ready', 'waits_for_go')),
+ waits_for_go boolean NOT NULL DEFAULT false,
+ started_at timestamptz,
+ ended_at timestamptz,
+ created_at timestamptz NOT NULL DEFAULT now(),
+ updated_at timestamptz NOT NULL DEFAULT now()
+);
+CREATE INDEX IF NOT EXISTS agent_stream_schedules_due ON agent_stream_schedules (starts_at) WHERE state = 'scheduled';
+CREATE INDEX IF NOT EXISTS agent_stream_schedules_user ON agent_stream_schedules (user_id, starts_at);
+CREATE UNIQUE INDEX IF NOT EXISTS agent_stream_schedules_stream ON agent_stream_schedules (managed_stream_id) WHERE managed_stream_id IS NOT NULL;
+
+-- The desk's heartbeat: online means 'heard from within the last three minutes'.
+CREATE TABLE IF NOT EXISTS agent_desks (
+ user_id bigint PRIMARY KEY REFERENCES users(id) ON DELETE CASCADE,
+ last_seen_at timestamptz NOT NULL DEFAULT now(),
+ version text COLLATE "C"
+);
+
+-- A follower's opt-in to a schedule's reminder. The subject is the follower's Network subject, so a schedule's
+-- reminder survives a Live account rename; the public view carries the count only.
+CREATE TABLE IF NOT EXISTS agent_schedule_reminders (
+ schedule_id text COLLATE "C" NOT NULL REFERENCES agent_stream_schedules(id) ON DELETE CASCADE,
+ subject_id text COLLATE "C" NOT NULL,
+ created_at timestamptz NOT NULL DEFAULT now(),
+ PRIMARY KEY (schedule_id, subject_id)
+);
@@ -23,7 +23,7 @@
@@ -23,7 +23,7 @@
"multer": "^1.4.5-lts.1",
"multer": "^1.4.5-lts.1",
"node-media-server": "^2.7.1",
"node-media-server": "^2.7.1",
"onnxruntime-node": "^1.20.0",
"onnxruntime-node": "^1.20.0",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.299.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.305.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.93.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.93.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.124.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.124.0",
"pg": "^8.23.1",
"pg": "^8.23.1",
@@ -2840,9 +2840,9 @@
@@ -2840,9 +2840,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.299.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.299.0",
- "integrity": "sha512-KoJW79nApPmE7Yoi1kFVDH5gabUwcDmwvtWaPBdM8nHqU2+gaOd7BkkvfYBbtbCR0nwf0nuY7qFTiKGXdme+rg==",
+ "version": "0.305.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.305.0",
+ "integrity": "sha512-jTU1zlnTmHlydwL3/c5HVdUiPpd8r2KucVdHYUYX5MZgJoqy4M7Hg2jy6UKbJQ6s665uIZyTvswbmJjB2uwc5Q==",
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"dependencies": {
"dependencies": {
"ajv": "^8.17.1",
"ajv": "^8.17.1",
@@ -42,7 +42,7 @@
@@ -42,7 +42,7 @@
"multer": "^1.4.5-lts.1",
"multer": "^1.4.5-lts.1",
"node-media-server": "^2.7.1",
"node-media-server": "^2.7.1",
"onnxruntime-node": "^1.20.0",
"onnxruntime-node": "^1.20.0",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.299.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.305.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.93.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.93.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.124.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.124.0",
"pg": "^8.23.1",
"pg": "^8.23.1",
@@ -179,3 +179,19 @@
@@ -179,3 +179,19 @@
.agent-controls-log-list { margin: 6px 0 0; padding-left: 18px; font-size: .85rem; color: var(--text-secondary, #96a7c2); }
.agent-controls-log-list { margin: 6px 0 0; padding-left: 18px; font-size: .85rem; color: var(--text-secondary, #96a7c2); }
.agent-overlay-paused { margin: 2px 0 0; font-size: .8rem; font-weight: 600; color: var(--warning-text, #fbbf24); }
.agent-overlay-paused { margin: 2px 0 0; font-size: .8rem; font-weight: 600; color: var(--warning-text, #fbbf24); }
@media (max-width: 480px) { .agent-controls-row .btn { flex: 1 1 calc(50% - 8px); } }
@media (max-width: 480px) { .agent-controls-row .btn { flex: 1 1 calc(50% - 8px); } }
+
+
+/* ── S7: the robot camera as a second scene, and the scheduled cards ──────────────────────────── */
+.agent-scenes { display: flex; gap: 6px; margin: 10px 0 0; }
+.agent-scene { padding: 6px 12px; border-radius: 999px; border: 1px solid var(--ov-border, #2a3140); background: var(--ov-surface, #121722); color: var(--ov-text, #e6e9ef); font: inherit; cursor: pointer; }
+.agent-scene.is-on { background: var(--ov-accent, #6ea8fe); color: #06101f; border-color: transparent; }
+.agent-robot { position: relative; aspect-ratio: 16 / 9; border-radius: 14px; overflow: hidden; background: #000; margin-top: 10px; }
+.agent-robot iframe { width: 100%; height: 100%; border: 0; display: block; }
+.agent-schedules { display: grid; grid-template-columns: repeat(auto-fill, minmax(230px, 1fr)); gap: 12px; margin: 0 0 16px; }
+.agent-schedules-title { margin: 18px 0 10px; font-size: 1rem; }
+.agent-schedule-card { display: flex; flex-direction: column; gap: 5px; padding: 12px 14px; border-radius: 12px; border: 1px solid var(--ov-border, #2a3140); background: var(--ov-surface, #121722); color: inherit; text-decoration: none; }
+.agent-schedule-card:hover { border-color: var(--ov-accent, #6ea8fe); }
+.agent-schedule-when { font-size: .78rem; text-transform: uppercase; letter-spacing: .04em; opacity: .8; }
+.agent-schedule-title { font-size: .98rem; line-height: 1.25; }
+.agent-schedule-task, .agent-schedule-meta { font-size: .82rem; opacity: .8; }
+@media (max-width: 640px) { .agent-scene { min-height: 44px; } }
@@ -28,6 +28,13 @@
@@ -28,6 +28,13 @@
<div class="agent-curtain" id="agent-curtain" hidden>Private screen</div>
<div class="agent-curtain" id="agent-curtain" hidden>Private screen</div>
<div id="agent-overlay"></div>
<div id="agent-overlay"></div>
</div>
</div>
+ <div class="agent-scenes" id="agent-scenes" hidden role="group" aria-label="The stream's scenes">
+ <button type="button" class="agent-scene is-on" id="agent-scene-desk" data-agent-scene="desk" aria-pressed="true">Desk</button>
+ <button type="button" class="agent-scene" id="agent-scene-robot" data-agent-scene="robot" aria-pressed="false">Robot camera</button>
+ </div>
+ <div class="agent-robot" id="agent-robot" hidden>
+ <iframe id="agent-robot-frame" title="Robot camera" loading="lazy" allow="camera; microphone; fullscreen"></iframe>
+ </div>
<p class="agent-status-line" id="agent-status-line">Loading…</p>
<p class="agent-status-line" id="agent-status-line">Loading…</p>
<section class="agent-chapters" id="agent-chapters" hidden aria-label="The recording's chapters">
<section class="agent-chapters" id="agent-chapters" hidden aria-label="The recording's chapters">
<h2 class="agent-chapters-title" id="agent-chapters-title">Chapters</h2>
<h2 class="agent-chapters-title" id="agent-chapters-title">Chapters</h2>
@@ -378,6 +378,8 @@
@@ -378,6 +378,8 @@
<!-- Agents (Agent Streams S2): this channel's AI agent streams, live and past. -->
<!-- Agents (Agent Streams S2): this channel's AI agent streams, live and past. -->
<div class="ch-tab-panel" id="ch-panel-agents" role="tabpanel">
<div class="ch-tab-panel" id="ch-panel-agents" role="tabpanel">
<p class="ch-panel-hint"><i class="fa-solid fa-robot"></i> AI agents working on this channel — live now, and what they finished.</p>
<p class="ch-panel-hint"><i class="fa-solid fa-robot"></i> AI agents working on this channel — live now, and what they finished.</p>
+ <h3 class="agent-schedules-title" id="ch-agents-schedules-title" hidden>Coming up</h3>
+ <div class="agent-schedules" id="ch-agents-schedules"></div>
<div class="agent-cards" id="ch-agents-grid"></div>
<div class="agent-cards" id="ch-agents-grid"></div>
<button class="btn btn-small btn-outline agent-more" id="ch-agents-more" type="button" hidden onclick="loadMoreChannelAgents(this)">More</button>
<button class="btn btn-small btn-outline agent-more" id="ch-agents-more" type="button" hidden onclick="loadMoreChannelAgents(this)">More</button>
</div>
</div>
@@ -93,5 +93,35 @@
@@ -93,5 +93,35 @@
};
};
}
}
- window.AgentCards = { card, fill, minutes };
+ /** A scheduled stream's card: the time, the task and the reminder count (never who opted in). */
+ function scheduleCard(entry) {
+ const a = el('a', 'agent-schedule-card');
+ a.href = entry.page_url || '#';
+ const when = el('span', 'agent-schedule-when');
+ const at = new Date(entry.starts_at);
+ when.textContent = Number.isNaN(at.getTime()) ? '' : at.toLocaleString(undefined, { weekday: 'short', month: 'short', day: 'numeric', hour: 'numeric', minute: '2-digit' });
+ a.appendChild(when);
+ a.appendChild(el('b', 'agent-schedule-title', entry.title || 'An AI agent stream'));
+ const task = where(entry.task);
+ if (task) a.appendChild(el('span', 'agent-schedule-task', task));
+ const meta = el('span', 'agent-schedule-meta');
+ const bits = [entry.mode === 'chatplays' ? 'Chat plays · waits for Go' : { narrated: 'Narrated', copilot: 'Co-pilot', watch: 'Watch' }[entry.mode] || 'Watch'];
+ if (entry.reminders) bits.push(`${entry.reminders} reminder${entry.reminders === 1 ? '' : 's'}`);
+ if (entry.waits_for_go) bits.push('waiting for the owner\'s Go');
+ if (entry.state === 'missed' && entry.missed_reason === 'desk_offline') bits.push("Didn't start: the desk was offline");
+ meta.textContent = bits.join(' · ');
+ a.appendChild(meta);
+ return a;
+ }
+
+ /** Fill a schedule rail; the caller hides it when it is empty. */
+ function fillSchedules(grid, entries) {
+ if (!grid) return 0;
+ grid.replaceChildren();
+ const all = Array.isArray(entries) ? entries : [];
+ for (const entry of all) grid.appendChild(scheduleCard(entry));
+ return all.length;
+ }
+
+ window.AgentCards = { card, fill, scheduleCard, fillSchedules, minutes };
})();
})();
@@ -306,6 +306,34 @@
@@ -306,6 +306,34 @@
});
});
}
}
+ // S7: a channel bound to a Bot robot may show the robot's camera as a second scene beside the desk. The panel is
+ // Bot's own embed; Live only switches which scene is on, and the overlay stays the same on both.
+ function renderScenes(stream) {
+ const bar = document.getElementById('agent-scenes');
+ const robot = document.getElementById('agent-robot');
+ const frame = document.getElementById('agent-robot-frame');
+ const stage = document.getElementById('agent-stage');
+ if (!bar || !robot || !stage) return;
+ const scene = stream && stream.robot_scene;
+ if (!scene || !scene.embed_url) { bar.hidden = true; robot.hidden = true; stage.hidden = false; return; }
+ if (frame && frame.getAttribute('src') !== scene.embed_url) frame.setAttribute('src', scene.embed_url);
+ bar.hidden = false;
+ const show = (name) => {
+ const isRobot = name === 'robot';
+ stage.hidden = isRobot;
+ robot.hidden = !isRobot;
+ for (const btn of bar.querySelectorAll('[data-agent-scene]')) {
+ const on = btn.getAttribute('data-agent-scene') === name;
+ btn.classList.toggle('is-on', on);
+ btn.setAttribute('aria-pressed', on ? 'true' : 'false');
+ }
+ };
+ for (const btn of bar.querySelectorAll('[data-agent-scene]')) {
+ btn.onclick = () => show(btn.getAttribute('data-agent-scene'));
+ }
+ show('desk');
+ }
+
function render(data) {
function render(data) {
state = data;
state = data;
ticket = data.view_ticket || ticket;
ticket = data.view_ticket || ticket;
@@ -330,6 +358,7 @@
@@ -330,6 +358,7 @@
}
}
renderCurtain(!!(stream.curtain && stream.curtain.up));
renderCurtain(!!(stream.curtain && stream.curtain.up));
+ renderScenes(stream);
renderStatus(stream, liveViewers);
renderStatus(stream, liveViewers);
mountPlays(data);
mountPlays(data);
ensureOverlay(data);
ensureOverlay(data);
@@ -422,6 +451,7 @@
@@ -422,6 +451,7 @@
}
}
}
}
renderCurtain(!!(stream.curtain && stream.curtain.up));
renderCurtain(!!(stream.curtain && stream.curtain.up));
+ renderScenes(stream);
renderVoice();
renderVoice();
const endBtn = $('agent-end');
const endBtn = $('agent-end');
if (endBtn) endBtn.hidden = stream.state === 'ended';
if (endBtn) endBtn.hidden = stream.state === 'ended';
@@ -1992,9 +1992,14 @@ async function loadChannelAgents(username, { force = false } = {}) {
@@ -1992,9 +1992,14 @@ async function loadChannelAgents(username, { force = false } = {}) {
try { data = await api(`/streams/agent?channel=${encodeURIComponent(username)}`); } catch { data = null; }
try { data = await api(`/streams/agent?channel=${encodeURIComponent(username)}`); } catch { data = null; }
if (username !== currentChannelUsername) return; // the page moved on while this was in flight
if (username !== currentChannelUsername) return; // the page moved on while this was in flight
const cards = (data && Array.isArray(data.agents)) ? data.agents : [];
const cards = (data && Array.isArray(data.agents)) ? data.agents : [];
+ const planned = (data && Array.isArray(data.schedules)) ? data.schedules : [];
+ const schedulesEl = document.getElementById('ch-agents-schedules');
+ const schedulesTitle = document.getElementById('ch-agents-schedules-title');
+ const plannedCount = (window.AgentCards && schedulesEl) ? window.AgentCards.fillSchedules(schedulesEl, planned) : 0;
+ if (schedulesTitle) schedulesTitle.hidden = !plannedCount;
const tab = document.getElementById('ch-tab-btn-agents');
const tab = document.getElementById('ch-tab-btn-agents');
const badge = document.getElementById('ch-tab-badge-agents');
const badge = document.getElementById('ch-tab-badge-agents');
- if (tab) tab.style.display = cards.length ? '' : 'none';
+ if (tab) tab.style.display = (cards.length || plannedCount) ? '' : 'none';
if (badge) {
if (badge) {
const live = cards.filter((entry) => entry.state === 'live').length;
const live = cards.filter((entry) => entry.state === 'live').length;
badge.hidden = !live;
badge.hidden = !live;
@@ -2005,7 +2010,7 @@ async function loadChannelAgents(username, { force = false } = {}) {
@@ -2005,7 +2010,7 @@ async function loadChannelAgents(username, { force = false } = {}) {
if (!cards.length) {
if (!cards.length) {
const empty = document.createElement('p');
const empty = document.createElement('p');
empty.className = 'agent-empty';
empty.className = 'agent-empty';
- empty.textContent = 'No AI agent streams on this channel yet.';
+ empty.textContent = plannedCount ? 'No AI agent stream is live right now; the next one is above.' : 'No AI agent streams on this channel yet.';
grid.appendChild(empty);
grid.appendChild(empty);
if (more) more.hidden = true;
if (more) more.hidden = true;
return;
return;
@@ -252,6 +252,7 @@ async function _rdTopic(root, slug) {
@@ -252,6 +252,7 @@ async function _rdTopic(root, slug) {
${t.summary ? `<div class="rd-summary"><span class="rd-desk">The desk</span>${_rdParas(t.summary)}</div>` : ''}
${t.summary ? `<div class="rd-summary"><span class="rd-desk">The desk</span>${_rdParas(t.summary)}</div>` : ''}
<h2 class="rd-h2">Every take</h2>
<h2 class="rd-h2">Every take</h2>
${(d.takes || []).map((s) => _rdTake(s, { topic: false })).join('') || _rdEmpty('fa-comment', 'No takes on this topic right now.')}
${(d.takes || []).map((s) => _rdTake(s, { topic: false })).join('') || _rdEmpty('fa-comment', 'No takes on this topic right now.')}
+ ${(d.citations || []).length ? `<h2 class="rd-h2">Agents that worked on this</h2><ul class="rd-onair-list">${d.citations.map((r) => `<li><a class="rd-onair-item" href="https://openvibe.report/s/${encodeURIComponent(r.id)}"><span class="rd-onair-t"><b>${_rdEsc(r.title)}</b><small>${_rdEsc(r.summary || '')} · ${_rdEsc(r.outcome || 'done')}</small></span></a></li>`).join('')}</ul>` : ''}
${(d.related || []).length ? `<h2 class="rd-h2">Came up alongside</h2><div class="rd-chips">${d.related.map((r) => _rdTopicChip(r)).join('')}</div>` : ''}
${(d.related || []).length ? `<h2 class="rd-h2">Came up alongside</h2><div class="rd-chips">${d.related.map((r) => _rdTopicChip(r)).join('')}</div>` : ''}
</article>`;
</article>`;
document.title = `${t.title} · The Rundown · OpenVibe.Live`;
document.title = `${t.title} · The Rundown · OpenVibe.Live`;
@@ -28,6 +28,7 @@ const tickets = require('./tickets');
@@ -28,6 +28,7 @@ const tickets = require('./tickets');
const frames = require('./frames');
const frames = require('./frames');
const steps = require('./steps');
const steps = require('./steps');
const voice = require('./voice');
const voice = require('./voice');
+const schedules = require('./schedules');
const router = express.Router();
const router = express.Router();
@@ -128,7 +129,9 @@ router.get('/', auth.viewer, async (req, res) => {
@@ -128,7 +129,9 @@ router.get('/', auth.viewer, async (req, res) => {
const channel = typeof req.query.channel === 'string' ? req.query.channel.trim().slice(0, 64) : null;
const channel = typeof req.query.channel === 'string' ? req.query.channel.trim().slice(0, 64) : null;
if (channel) {
if (channel) {
const rows = await service.listForChannel(channel, req.viewer ? req.viewer.id : null);
const rows = await service.listForChannel(channel, req.viewer ? req.viewer.id : null);
- return res.json({ agents: rows.map(service.cardView) });
+ const subject = req.viewer ? await auth.viewerSubject(req.viewer) : null;
+ const planned = await schedules.listForChannel(channel, req.viewer ? req.viewer.id : null, { subject });
+ return res.json({ agents: rows.map(service.cardView), schedules: planned });
}
}
const rows = await service.listPublic({ liveOnly: req.query.live === '1' });
const rows = await service.listPublic({ liveOnly: req.query.live === '1' });
res.json({ agents: rows.map(service.cardView) });
res.json({ agents: rows.map(service.cardView) });
@@ -178,6 +181,138 @@ async function watchable(req, row) {
@@ -178,6 +181,138 @@ async function watchable(req, row) {
});
});
}
}
+
+// ── Schedules (S7) ─────────────────────────────────────────────
+//
+// An owner schedules a public agent stream; the record shows on the channel's Agents tab and in Today 24 hours before
+// its time. The desk says it is online with a heartbeat, claims a due schedule and receives the one publish ticket.
+// A schedule nobody claims says "didn't start: the desk was offline" and starts nothing. A chatplays schedule waits
+// for the owner's Go. Reminders are opt-in and the answer is a count, never a name.
+
+/** POST /desk/heartbeat — the owner's desk says it is online (a fact, not a stream). */
+router.post('/desk/heartbeat', auth.publisher, async (req, res) => {
+ try {
+ const out = await schedules.heartbeat(req.publisher.user, (req.body || {}).version);
+ res.set('Cache-Control', 'no-store');
+ res.json(out);
+ } catch (err) {
+ console.error('[AgentStream] desk heartbeat failed:', err.message);
+ fail(res, 500, 'heartbeat_failed', 'The heartbeat could not be stored');
+ }
+});
+
+/** POST /schedules — the owner writes a schedule. */
+router.post('/schedules', auth.publisher, async (req, res) => {
+ try {
+ const body = req.body || {};
+ const task = schedules.cleanTask(body.task);
+ if (!task) return fail(res, 400, 'invalid_task', 'task needs source (codes, actor, desk, node), ref and title');
+ const schedule = await schedules.create({
+ user: req.publisher.user,
+ title: body.title,
+ task,
+ mode: body.mode,
+ startsAt: body.starts_at,
+ });
+ res.status(201).json({ schedule });
+ } catch (err) {
+ if (err.status) return fail(res, err.status, err.code, err.message);
+ console.error('[AgentStream] schedule create failed:', err.message);
+ fail(res, 500, 'schedule_failed', 'The schedule could not be stored');
+ }
+});
+
+/** GET /schedules — the caller's own (?mine=1) or a channel's public window (the channel's Agents tab). */
+router.get('/schedules', auth.viewer, async (req, res) => {
+ try {
+ res.set('Cache-Control', 'no-store');
+ const subject = req.viewer ? await auth.viewerSubject(req.viewer) : null;
+ if (String(req.query.mine || '') === '1') {
+ if (!req.viewer) return res.json({ schedules: [] });
+ return res.json({ schedules: await schedules.listMine(req.viewer.id, { subject }) });
+ }
+ if (String(req.query.public || '') === '1') {
+ return res.json({ schedules: await schedules.listPublic({ subject }) });
+ }
+ const channel = typeof req.query.channel === 'string' ? req.query.channel.trim().slice(0, 64) : null;
+ if (!channel) return fail(res, 400, 'channel_required', 'channel is required (or mine=1 or public=1)');
+ const viewerId = req.viewer ? req.viewer.id : null;
+ const rows = await schedules.listForChannel(channel, viewerId, { subject });
+ res.json({ schedules: rows });
+ } catch (err) {
+ console.error('[AgentStream] schedules read failed:', err.message);
+ fail(res, 500, 'schedules_failed', 'The schedules could not be read');
+ }
+});
+
+/** GET /schedules/:id — one schedule (the owner's own, or a public one inside its 24-hour window). */
+router.get('/schedules/:id', auth.viewer, async (req, res) => {
+ try {
+ const subject = req.viewer ? await auth.viewerSubject(req.viewer) : null;
+ const schedule = await schedules.get(req.params.id, { subject });
+ if (!schedule) return fail(res, 404, 'schedule_not_found', 'No such schedule');
+ const mine = req.viewer && schedule.owner && schedule.owner === subject;
+ const soon = schedule.visibility === 'public' && schedule.state === 'scheduled'
+ && Date.parse(schedule.starts_at) <= Date.now() + schedules.AHEAD_MS;
+ if (!mine && !soon) return fail(res, 404, 'schedule_not_found', 'No such schedule');
+ res.set('Cache-Control', 'no-store');
+ res.json({ schedule });
+ } catch (err) {
+ console.error('[AgentStream] schedule read failed:', err.message);
+ fail(res, 500, 'schedule_failed', 'The schedule could not be read');
+ }
+});
+
+/** DELETE /schedules/:id — cancel it. */
+router.delete('/schedules/:id', auth.publisher, async (req, res) => {
+ try {
+ const schedule = await schedules.cancel(req.params.id, req.publisher.user);
+ res.json({ schedule });
+ } catch (err) {
+ if (err.status) return fail(res, err.status, err.code, err.message);
+ console.error('[AgentStream] schedule cancel failed:', err.message);
+ fail(res, 500, 'schedule_failed', 'The schedule could not be cancelled');
+ }
+});
+
+/** POST /schedules/:id/remind — opt in to the reminder. The answer is the count, never the people. */
+router.post('/schedules/:id/remind', auth.viewer, async (req, res) => {
+ try {
+ if (!req.viewer) return fail(res, 401, 'authentication_required', 'Sign in to set a reminder');
+ const subject = await auth.viewerSubject(req.viewer);
+ const out = await schedules.remind(req.params.id, subject);
+ res.json(out);
+ } catch (err) {
+ if (err.status) return fail(res, err.status, err.code, err.message);
+ console.error('[AgentStream] schedule reminder failed:', err.message);
+ fail(res, 500, 'reminder_failed', 'The reminder could not be set');
+ }
+});
+
+/** POST /schedules/:id/claim — the owner's online desk starts it: the stream and its one publish ticket. */
+router.post('/schedules/:id/claim', auth.publisher, async (req, res) => {
+ try {
+ const out = await schedules.claim(req.params.id, req.publisher.user, { go: false });
+ res.status(201).json(out);
+ } catch (err) {
+ if (err.status) return fail(res, err.status, err.code, err.message);
+ console.error('[AgentStream] schedule claim failed:', err.message);
+ fail(res, 500, 'schedule_failed', 'The schedule could not be started');
+ }
+});
+
+/** POST /schedules/:id/go — the owner's Go for a chatplays schedule; it starts the stream like a claim. */
+router.post('/schedules/:id/go', auth.publisher, async (req, res) => {
+ try {
+ const out = await schedules.claim(req.params.id, req.publisher.user, { go: true });
+ res.status(201).json(out);
+ } catch (err) {
+ if (err.status) return fail(res, err.status, err.code, err.message);
+ console.error('[AgentStream] schedule go failed:', err.message);
+ fail(res, 500, 'schedule_failed', 'The schedule could not be started');
+ }
+});
+
// ── Read ───────────────────────────────────────────────────────
// ── Read ───────────────────────────────────────────────────────
/**
/**
@@ -222,8 +357,14 @@ router.get('/:id', auth.viewer, async (req, res) => {
@@ -222,8 +357,14 @@ router.get('/:id', auth.viewer, async (req, res) => {
: null;
: null;
const moderation = await channelModeration(row.channel_id);
const moderation = await channelModeration(row.channel_id);
const isMod = !isOwner && !!(req.viewer && moderation.moderator_ids.includes(Number(req.viewer.id)));
const isMod = !isOwner && !!(req.viewer && moderation.moderator_ids.includes(Number(req.viewer.id)));
+ let parties = null;
+ try {
+ const count = await require('../chat/chat-client').partyCount(row.public_id);
+ if (count && Number.isInteger(Number(count.count))) parties = { count: Math.max(0, Number(count.count)) };
+ } catch { /* Chat is optional for a stream */ }
res.json({
res.json({
stream,
stream,
+ parties,
live_stream_id: row.live_session_id ? Number(row.live_session_id) : null,
live_stream_id: row.live_session_id ? Number(row.live_session_id) : null,
can_manage: isOwner,
can_manage: isOwner,
can_moderate: isOwner || isMod,
can_moderate: isOwner || isMod,
new file mode 100644
new file mode 100644
@@ -0,0 +1,284 @@
@@ -0,0 +1,284 @@
+'use strict';
+/**
+ * Scheduled public agent streams (Agent Streams S7, design 2026-11-06).
+ *
+ * An owner schedules a public agent stream: a time, a task template (a Codes task or an Actor recipe) and the mode.
+ * The record shows on the channel's Agents tab and in Today 24 hours before its time. When the time comes the owner's
+ * desk claims it: Live creates the stream and hands the desk a publish ticket (the ticket is issued once and never
+ * stored raw). A desk that is not online does not claim, and the sweep records the schedule as missed with
+ * `desk_offline` — the design's "didn't start: the desk was offline" — and starts nothing.
+ *
+ * A chatplays schedule never starts on its own: it keeps its time and `waits_for_go` until the owner presses Go (the
+ * S6 rule). Reminders are opt-in only: a follower asks, the row names them, and every public view carries the count.
+ *
+ * The desk heartbeat (agent_desks) tells the sweep whether the owner's desk was online at the time, so the card can
+ * say which of the two happened. A heartbeat is a fact, not a stream.
+ */
+const db = require('../db/database');
+const contracts = require('openvibe-contracts');
+const service = require('./service');
+
+const MODES = service.MODES;
+const MAX_TITLE = 200;
+const MAX_TASK_REF = 200;
+const AHEAD_MS = 24 * 60 * 60 * 1000;
+const SCHEDULE_WINDOW_MS = 30 * 24 * 60 * 60 * 1000;
+const DESK_ONLINE_MS = 3 * 60 * 1000;
+const MISS_GRACE_MS = 60 * 1000;
+const TASK_SOURCES = service.TASK_SOURCES;
+
+class ScheduleError extends Error {
+ constructor(status, code, message) {
+ super(message);
+ this.status = status;
+ this.code = code;
+ }
+}
+
+function text(value, max) {
+ if (typeof value !== 'string') return null;
+ const s = value.trim();
+ return s && Array.from(s).length <= max ? s : null;
+}
+
+function cleanTask(task) {
+ if (!task || typeof task !== 'object' || Array.isArray(task)) return null;
+ const source = TASK_SOURCES.includes(task.source) ? task.source : null;
+ const ref = text(task.ref, MAX_TASK_REF);
+ const title = text(task.title, MAX_TITLE);
+ if (!source || !ref || !title) return null;
+ const url = task.url == null ? null : (typeof task.url === 'string' && /^https?:\/\//i.test(task.url) && task.url.length <= 2000 ? task.url : undefined);
+ if (url === undefined) return null;
+ return { source, ref, title, url };
+}
+
+function cleanStart(value, now = Date.now()) {
+ const at = new Date(String(value || ''));
+ if (Number.isNaN(at.getTime())) throw new ScheduleError(400, 'invalid_starts_at', 'starts_at is a date-time');
+ if (at.getTime() < now + 60 * 1000) throw new ScheduleError(400, 'starts_at_past', 'A schedule starts at least a minute from now');
+ if (at.getTime() > now + SCHEDULE_WINDOW_MS) throw new ScheduleError(400, 'starts_at_far', 'A schedule is at most 30 days away');
+ return at;
+}
+
+const ID_RE = /^scd_[0-9A-HJKMNP-TV-Z]{26}$/;
+const newId = () => 'scd_' + contracts.ids.ulid();
+
+const SELECT = `
+ SELECT sc.*, u.username, u.display_name,
+ la.subject_id AS owner_subject,
+ ms.public_id AS agent_stream_id,
+ (SELECT COUNT(*) FROM agent_schedule_reminders r WHERE r.schedule_id = sc.id) AS reminders,
+ EXISTS(SELECT 1 FROM agent_schedule_reminders r WHERE r.schedule_id = sc.id AND r.subject_id = ?) AS reminded
+ FROM agent_stream_schedules sc
+ JOIN users u ON u.id = sc.user_id
+ LEFT JOIN linked_accounts la ON la.service = 'network' AND la.user_id = sc.user_id
+ LEFT JOIN managed_streams ms ON ms.id = sc.managed_stream_id`;
+
+function pageUrl(row) {
+ const base = String(require('../config').baseUrl || 'https://openvibe.live').replace(/\/+$/, '');
+ return `${base}/@${encodeURIComponent(row.username || '')}/agents`;
+}
+
+function scheduleView(row, { subject = null } = {}) {
+ if (!row) return null;
+ return {
+ id: row.id,
+ owner: row.owner_subject || null,
+ channel: row.username,
+ title: row.title,
+ task: typeof row.task === 'string' ? JSON.parse(row.task) : (row.task || {}),
+ mode: row.mode,
+ visibility: row.visibility,
+ starts_at: service.iso(row.starts_at),
+ state: row.state,
+ agent_stream_id: row.agent_stream_id || null,
+ missed_reason: row.missed_reason || null,
+ waits_for_go: !!row.waits_for_go,
+ started_at: service.iso(row.started_at),
+ ended_at: service.iso(row.ended_at),
+ reminders: Number(row.reminders) || 0,
+ ...(subject ? { reminded: !!Number(row.reminded) } : {}),
+ page_url: pageUrl(row),
+ created_at: service.iso(row.created_at),
+ };
+}
+
+async function get(id, { subject = null } = {}) {
+ if (!ID_RE.test(String(id || ''))) return null;
+ const row = await db.get(`${SELECT} WHERE sc.id = ?`, [subject || null, id]);
+ return row ? scheduleView(row, { subject }) : null;
+}
+
+async function rowOf(id) {
+ if (!ID_RE.test(String(id || ''))) return null;
+ return await db.get('SELECT * FROM agent_stream_schedules WHERE id = ?', [id]) || null;
+}
+
+/** A new schedule. The route has already checked the owner's permission; this checks the record's own rules. */
+async function create({ user, title, task, mode = 'watch', startsAt }) {
+ const cleanTitle = text(title, MAX_TITLE);
+ if (!cleanTitle) throw new ScheduleError(400, 'invalid_title', 'A title of 1–200 characters is required');
+ const clean = cleanTask(task);
+ if (!clean) throw new ScheduleError(400, 'invalid_task', 'task needs source (codes, actor, desk, node), ref and title');
+ if (!MODES.includes(mode)) throw new ScheduleError(400, 'invalid_mode', 'mode is watch, narrated, copilot or chatplays');
+ const at = cleanStart(startsAt);
+ const id = newId();
+ await db.run(
+ `INSERT INTO agent_stream_schedules (id, user_id, title, task, mode, visibility, starts_at)
+ VALUES (?, ?, ?, ?::jsonb, ?, 'public', ?)`,
+ [id, user.id, cleanTitle, JSON.stringify(clean), mode, at]
+ );
+ return await get(id, { subject: await service.ownerSubject(user.id) });
+}
+
+/** The owner's own schedules, the next one first. */
+async function listMine(userId, { subject = null } = {}) {
+ const rows = await db.all(`${SELECT} WHERE sc.user_id = ? ORDER BY (sc.state = 'scheduled') DESC, sc.starts_at ASC LIMIT 100`, [subject || null, userId]);
+ return rows.map((row) => scheduleView(row, { subject }));
+}
+
+/**
+ * The public schedules a page may show: public, still waiting, and inside the next 24 hours (the design's window).
+ * The owner's own page also sees their whole list (the channel's Agents tab).
+ */
+async function listPublic({ channel = null, now = Date.now(), subject = null, limit = 100 } = {}) {
+ const params = [subject || null, new Date(now), new Date(now + AHEAD_MS)];
+ let where = "sc.visibility = 'public' AND sc.state = 'scheduled' AND sc.starts_at >= ? AND sc.starts_at <= ?";
+ if (channel) { where += ' AND LOWER(u.username) = LOWER(?)'; params.push(String(channel)); }
+ params.push(Math.min(200, Math.max(1, Number(limit) || 100)));
+ const rows = await db.all(`${SELECT} WHERE ${where} ORDER BY sc.starts_at ASC, sc.id ASC LIMIT ?`, params);
+ return rows.map((row) => scheduleView(row, { subject }));
+}
+
+/** The owner's whole list plus the public window, for the channel Agents tab. */
+async function listForChannel(username, viewerId, { subject = null } = {}) {
+ const rows = await db.all(`${SELECT} WHERE LOWER(u.username) = LOWER(?) ORDER BY sc.starts_at ASC LIMIT 200`, [subject || null, String(username)]);
+ return rows.filter((row) => Number(row.user_id) === Number(viewerId || 0)
+ || (row.visibility === 'public' && row.state === 'scheduled' && Date.parse(service.iso(row.starts_at)) <= Date.now() + AHEAD_MS));
+}
+
+/** Cancel one of the owner's schedules. A started or finished one is not cancellable here. */
+async function cancel(id, user) {
+ const row = await rowOf(id);
+ if (!row) throw new ScheduleError(404, 'schedule_not_found', 'No such schedule');
+ if (Number(row.user_id) !== Number(user.id)) throw new ScheduleError(403, 'not_your_schedule', 'This schedule belongs to another account');
+ if (row.state !== 'scheduled') throw new ScheduleError(409, 'schedule_not_cancellable', 'This schedule already started or ended');
+ await db.run("UPDATE agent_stream_schedules SET state = 'cancelled', updated_at = now() WHERE id = ?", [id]);
+ return await get(id);
+}
+
+/** Opt in to a reminder (idempotent). The answer is the count, never the names. */
+async function remind(id, subject) {
+ if (!/^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(subject || ''))) throw new ScheduleError(409, 'no_subject', 'This account has no OpenVibe subject yet; sign in again');
+ const row = await rowOf(id);
+ if (!row) throw new ScheduleError(404, 'schedule_not_found', 'No such schedule');
+ // Only a schedule this person may see: a public one inside its 24-hour window (a private one's id is not an
+ // invitation; the answer is the same 404 as a missing one).
+ if (row.visibility !== 'public' || Date.parse(service.iso(row.starts_at)) > Date.now() + AHEAD_MS) throw new ScheduleError(404, 'schedule_not_found', 'No such schedule');
+ if (row.state !== 'scheduled' || Date.parse(service.iso(row.starts_at)) < Date.now()) throw new ScheduleError(409, 'schedule_over', 'This schedule has already started or ended');
+ await db.run('INSERT INTO agent_schedule_reminders (schedule_id, subject_id) VALUES (?, ?) ON CONFLICT (schedule_id, subject_id) DO NOTHING', [id, subject]);
+ const n = Number((await db.get('SELECT COUNT(*) AS n FROM agent_schedule_reminders WHERE schedule_id = ?', [id]))?.n || 0);
+ return { reminders: n, reminded: true };
+}
+
+/** The desk says it is online. A heartbeat is a fact, not a stream: it carries no ticket. */
+async function heartbeat(user, version = null) {
+ await db.run(
+ `INSERT INTO agent_desks (user_id, last_seen_at, version) VALUES (?, now(), ?)
+ ON CONFLICT (user_id) DO UPDATE SET last_seen_at = now(), version = COALESCE(excluded.version, agent_desks.version)`,
+ [user.id, typeof version === 'string' ? version.slice(0, 80) : null]
+ );
+ return { online: true, at: new Date().toISOString() };
+}
+
+async function deskSeen(userId) {
+ const row = await db.get('SELECT last_seen_at FROM agent_desks WHERE user_id = ?', [userId]);
+ const at = row ? Date.parse(service.iso(row.last_seen_at)) : NaN;
+ return Number.isFinite(at) ? at : null;
+}
+
+/**
+ * The sweep's part: a schedule whose time passed and that nobody claimed becomes missed. A chatplays one keeps its
+ * time and says it waits for Go; nothing is started in either case.
+ */
+async function startDue({ now = Date.now() } = {}) {
+ const rows = await db.all(
+ "SELECT * FROM agent_stream_schedules WHERE state = 'scheduled' AND starts_at <= ? ORDER BY starts_at ASC LIMIT 200",
+ [new Date(now - MISS_GRACE_MS)]
+ );
+ const out = { missed: 0, waiting: 0 };
+ for (const row of rows) {
+ if (row.mode === 'chatplays') {
+ if (!row.waits_for_go || row.missed_reason !== 'waits_for_go') {
+ await db.run("UPDATE agent_stream_schedules SET waits_for_go = true, missed_reason = 'waits_for_go', updated_at = now() WHERE id = ?", [row.id]);
+ }
+ out.waiting++;
+ continue;
+ }
+ const seen = await deskSeen(row.user_id);
+ const online = seen != null && now - seen <= DESK_ONLINE_MS;
+ await db.run(
+ "UPDATE agent_stream_schedules SET state = 'missed', missed_reason = ?, updated_at = now() WHERE id = ?",
+ [online ? 'task_not_ready' : 'desk_offline', row.id]
+ );
+ out.missed++;
+ }
+ return out;
+}
+
+/**
+ * The owner's desk claims a due schedule: Live creates the stream and answers the one publish ticket. A chatplays
+ * schedule is refused until the owner's Go; a schedule that is not due, or not the caller's, is refused too.
+ */
+async function claim(id, user, { go = false, now = Date.now() } = {}) {
+ const row = await db.get(
+ `SELECT sc.*, u.username, la.subject_id AS owner_subject FROM agent_stream_schedules sc
+ JOIN users u ON u.id = sc.user_id
+ LEFT JOIN linked_accounts la ON la.service = 'network' AND la.user_id = sc.user_id
+ WHERE sc.id = ?`, [id]
+ );
+ if (!row) throw new ScheduleError(404, 'schedule_not_found', 'No such schedule');
+ if (Number(row.user_id) !== Number(user.id)) throw new ScheduleError(403, 'not_your_schedule', 'This schedule belongs to another account');
+ if (row.state !== 'scheduled') throw new ScheduleError(409, 'schedule_not_scheduled', 'This schedule already started or ended');
+ if (Date.parse(service.iso(row.starts_at)) > now + 60 * 1000) throw new ScheduleError(409, 'schedule_not_due', 'This schedule is not due yet');
+ if (row.mode === 'chatplays' && !go) {
+ await db.run("UPDATE agent_stream_schedules SET waits_for_go = true, missed_reason = 'waits_for_go', updated_at = now() WHERE id = ?", [id]);
+ throw new ScheduleError(409, 'waits_for_go', 'A chat plays schedule waits for the owner\'s Go');
+ }
+ // The stream is the schedule's, and only one stream may come from it: the row flips first, in one statement.
+ const claimed = await db.run(
+ `UPDATE agent_stream_schedules SET state = 'live', started_at = now(), waits_for_go = false, missed_reason = NULL, updated_at = now()
+ WHERE id = ? AND state = 'scheduled' RETURNING id`, [id]
+ );
+ if (!claimed.changes) throw new ScheduleError(409, 'schedule_not_scheduled', 'This schedule already started or ended');
+ let stream;
+ let publish;
+ try {
+ stream = await service.create({
+ user, title: row.title, task: typeof row.task === 'string' ? JSON.parse(row.task) : row.task,
+ visibility: 'public', mode: row.mode,
+ });
+ publish = await require('./tickets').issuePublish(stream.managed_stream_id);
+ await db.run('UPDATE agent_stream_schedules SET managed_stream_id = ?, updated_at = now() WHERE id = ?', [stream.managed_stream_id, id]);
+ } catch (err) {
+ await db.run("UPDATE agent_stream_schedules SET state = 'scheduled', started_at = NULL, updated_at = now() WHERE id = ? AND state = 'live' AND managed_stream_id IS NULL", [id]);
+ throw err;
+ }
+ return {
+ schedule: await get(id, { subject: row.owner_subject }),
+ stream: service.publicView(stream),
+ publish: { whip_url: service.whipUrl(stream.public_id), ...publish },
+ };
+}
+
+/** The stream a schedule started ended with it. Called from service.end, in the same request. */
+async function markEnded(managedStreamId) {
+ if (!managedStreamId) return;
+ await db.run("UPDATE agent_stream_schedules SET state = 'ended', ended_at = now(), updated_at = now() WHERE managed_stream_id = ? AND state = 'live'", [managedStreamId]);
+}
+
+module.exports = {
+ AHEAD_MS, DESK_ONLINE_MS, MISS_GRACE_MS, SCHEDULE_WINDOW_MS, ScheduleError,
+ cleanTask, newId, get, create, listMine, listPublic, listForChannel, cancel, remind, heartbeat, deskSeen, startDue, claim, markEnded,
+ scheduleView,
+};
@@ -83,10 +83,12 @@ const SELECT = `
@@ -83,10 +83,12 @@ const SELECT = `
(SELECT st.title FROM agent_stream_steps st WHERE st.managed_stream_id = ms.id ORDER BY st.id DESC LIMIT 1) AS latest_step_title,
(SELECT st.title FROM agent_stream_steps st WHERE st.managed_stream_id = ms.id ORDER BY st.id DESC LIMIT 1) AS latest_step_title,
(SELECT st.status FROM agent_stream_steps st WHERE st.managed_stream_id = ms.id ORDER BY st.id DESC LIMIT 1) AS latest_step_status,
(SELECT st.status FROM agent_stream_steps st WHERE st.managed_stream_id = ms.id ORDER BY st.id DESC LIMIT 1) AS latest_step_status,
(SELECT la.subject_id FROM linked_accounts la WHERE la.service = 'network' AND la.user_id = ms.user_id) AS owner_subject,
(SELECT la.subject_id FROM linked_accounts la WHERE la.service = 'network' AND la.user_id = ms.user_id) AS owner_subject,
- u.username, u.display_name, u.avatar_url, u.profile_color, u.is_banned, u.is_owner
+ u.username, u.display_name, u.avatar_url, u.profile_color, u.is_banned, u.is_owner,
+ c.bot_robot_id
FROM managed_streams ms
FROM managed_streams ms
JOIN agent_streams a ON a.managed_stream_id = ms.id
JOIN agent_streams a ON a.managed_stream_id = ms.id
- JOIN users u ON u.id = ms.user_id`;
+ JOIN users u ON u.id = ms.user_id
+ LEFT JOIN channels c ON c.id = ms.channel_id`;
async function getByPublicId(publicId) {
async function getByPublicId(publicId) {
if (!AGENT_ID_RE.test(String(publicId || ''))) return null;
if (!AGENT_ID_RE.test(String(publicId || ''))) return null;
@@ -167,6 +169,9 @@ function publicView(row) {
@@ -167,6 +169,9 @@ function publicView(row) {
label: 'AI agent',
label: 'AI agent',
// The chat room a public or unlisted stream carries (S2): people only, the agent reads it in S5.
// The chat room a public or unlisted stream carries (S2): people only, the agent reads it in S5.
chat_room: (row.visibility === 'public' || row.visibility === 'unlisted') ? `agent:${row.public_id}` : null,
chat_room: (row.visibility === 'public' || row.visibility === 'unlisted') ? `agent:${row.public_id}` : null,
+ // S7: a channel bound to a Bot robot can show the robot's camera as a second scene beside the desk. The
+ // binding and Bot's own embed_public decide whether it appears; Live only names the panel.
+ robot_scene: robotScene(row),
};
};
}
}
@@ -196,6 +201,19 @@ function cardView(row) {
@@ -196,6 +201,19 @@ function cardView(row) {
};
};
}
}
+/** The robot camera scene for a channel bound to a Bot robot, or null (S7). */
+function robotScene(row) {
+ try {
+ const bot = require('../bot/embed');
+ if (!bot.enabled() || !bot.validRobotId(row.bot_robot_id)) return null;
+ return {
+ robot_id: row.bot_robot_id,
+ label: 'Robot camera',
+ embed_url: bot.embedUrl(row.bot_robot_id),
+ };
+ } catch { return null; }
+}
+
/** Public agent streams (live or ended), newest first, for the home row and a channel's Agents tab. */
/** Public agent streams (live or ended), newest first, for the home row and a channel's Agents tab. */
async function listPublic({ channel = null, liveOnly = false, limit = 60 } = {}) {
async function listPublic({ channel = null, liveOnly = false, limit = 60 } = {}) {
const params = [];
const params = [];
@@ -309,6 +327,9 @@ async function end(row, { reason = 'ended' } = {}) {
@@ -309,6 +327,9 @@ async function end(row, { reason = 'ended' } = {}) {
event: { kind: 'ended', extra: { reason } },
event: { kind: 'ended', extra: { reason } },
});
});
await require('./tickets').revokeAll(row.managed_stream_id);
await require('./tickets').revokeAll(row.managed_stream_id);
+ // S7: the schedule that started this stream ends with it, and a robot the channel is bound to is stopped.
+ try { await require('./schedules').markEnded(row.managed_stream_id); } catch (err) { console.warn('[AgentStream] schedule end: ' + err.message); }
+ try { require('../bot/kill').stop(row); } catch (err) { console.warn('[AgentStream] bot stop: ' + err.message); }
if (session) {
if (session) {
try {
try {
const whip = require('../streaming/whip-handler');
const whip = require('../streaming/whip-handler');
@@ -47,6 +47,13 @@ async function sweep() {
@@ -47,6 +47,13 @@ async function sweep() {
}
}
} catch (err) { console.warn('[AgentStream] idle voice failed:', err.message); }
} catch (err) { console.warn('[AgentStream] idle voice failed:', err.message); }
+ // S7: a due schedule nobody claimed is marked missed ('didn't start: the desk was offline'); a chatplays one
+ // keeps its time and waits for the owner's Go. Nothing is started here.
+ try {
+ const out = await require('./schedules').startDue();
+ if (out.missed || out.waiting) console.warn(`[AgentStream] schedules: ${out.missed} missed, ${out.waiting} waiting for Go`);
+ } catch (err) { console.warn('[AgentStream] schedule sweep failed:', err.message); }
+
try { await tickets.prune(); } catch (err) { console.warn('[AgentStream] ticket prune failed:', err.message); }
try { await tickets.prune(); } catch (err) { console.warn('[AgentStream] ticket prune failed:', err.message); }
try {
try {
const sharing = require('./sharing');
const sharing = require('./sharing');
new file mode 100644
new file mode 100644
@@ -0,0 +1,57 @@
@@ -0,0 +1,57 @@
+'use strict';
+/**
+ * The kill switch's reach into OpenVibe.Bot (Agent Streams S7, design 2026-11-06 §6).
+ *
+ * When an agent stream ends — the owner's End, a moderator's end control, a ticket revoke or the sweep — and the
+ * stream's channel is bound to a robot (`channels.bot_robot_id`), Live tells Bot to e_stop that robot. Bot's own rules
+ * still hold: its owner may always stop it, and Live's internal call carries the deployment's shared secret. With
+ * `BOT_KILL_URL` unset (and in tests) this is a no-op: a stream ending never fails because Bot is down.
+ *
+ * This is not a second control path for chat: nothing in Chat, and neither S5 nor S6, reaches this. Only the owner's
+ * kill switch (and the stream ending, which the owner can press) does.
+ */
+const db = require('../db/database');
+
+const URL_BASE = () => String(process.env.BOT_KILL_URL || '').trim().replace(/\/+$/, '');
+const SECRET = () => String(process.env.BOT_KILL_SECRET || '');
+const TIMEOUT_MS = 2500;
+
+function configured() { return !!URL_BASE() && !!SECRET(); }
+
+/** The robot bound to this agent stream's channel, or null. */
+async function robotFor(row) {
+ if (!row || !row.channel_id) return null;
+ try {
+ const channel = await db.get('SELECT bot_robot_id FROM channels WHERE id = ?', [row.channel_id]);
+ const id = channel && String(channel.bot_robot_id || '');
+ return /^rob_[A-Za-z0-9_-]{4,60}$/.test(id) ? id : null;
+ } catch { return null; }
+}
+
+/**
+ * Stop the robot bound to this stream's channel, if there is one and Bot is configured. Resolves to
+ * { stopped, robot_id } either way; a Bot failure is logged, never thrown into the stream's end.
+ */
+async function stop(row, { fetchImpl = globalThis.fetch } = {}) {
+ const robotId = await robotFor(row);
+ if (!robotId) return { stopped: false, robot_id: null };
+ if (!configured()) return { stopped: false, robot_id: robotId, reason: 'bot_kill_unconfigured' };
+ try {
+ const res = await fetchImpl(`${URL_BASE()}/internal/robots/${encodeURIComponent(robotId)}/estop`, {
+ method: 'POST',
+ headers: { 'Content-Type': 'application/json', 'X-OV-Bot-Kill': SECRET() },
+ body: JSON.stringify({ reason: 'agent_stream_ended', stream_id: row.public_id || null }),
+ signal: AbortSignal.timeout(TIMEOUT_MS),
+ });
+ if (!res.ok) {
+ console.warn(`[AgentStream] Bot did not stop ${robotId}: ${res.status}`);
+ return { stopped: false, robot_id: robotId, reason: `bot_${res.status}` };
+ }
+ return { stopped: true, robot_id: robotId };
+ } catch (err) {
+ console.warn(`[AgentStream] Bot stop of ${robotId} failed: ${err.message}`);
+ return { stopped: false, robot_id: robotId, reason: 'bot_unreachable' };
+ }
+}
+
+module.exports = { stop, robotFor, configured, URL_BASE };
@@ -199,6 +199,17 @@ async function readSoundByCommand(params) {
@@ -199,6 +199,17 @@ async function readSoundByCommand(params) {
} catch (err) { note(`sound-by-command: ${err.message}`); }
} catch (err) { note(`sound-by-command: ${err.message}`); }
return null;
return null;
}
}
+/** How many listed watch parties a public agent stream has (S7). The answer is a count; Chat never names the people. */
+async function partyCount(agentStreamId) {
+ if (!available()) return null;
+ try {
+ const { status, data } = await request('GET', `/internal/chat/parties/count${qs({ agent_stream: agentStreamId })}`);
+ if (status === 200 && data && data.ok !== false) return data;
+ note(`party-count: Chat answered ${status}${data && data.error ? ` (${data.error})` : ''}`);
+ } catch (err) { note(`party-count: ${err.message}`); }
+ return null;
+}
+
/** Record where Live's asset sync uploaded a channel sound (idempotent; Chat applies the same write once). */
/** Record where Live's asset sync uploaded a channel sound (idempotent; Chat applies the same write once). */
async function soundAsset(body) {
async function soundAsset(body) {
if (!available()) return null;
if (!available()) return null;
@@ -231,6 +242,6 @@ module.exports = {
@@ -231,6 +242,6 @@ module.exports = {
invalidate: async (body) => await send('invalidate', body),
invalidate: async (body) => await send('invalidate', body),
presence,
presence,
readStats, readMessages, readTimeline, readFirstChat, readPendingIp, readRelayUsers, readRelayUser, readTtsOverride,
readStats, readMessages, readTimeline, readFirstChat, readPendingIp, readRelayUsers, readRelayUser, readTtsOverride,
- readSounds, readSoundByCommand, readDmBlockState, soundAsset, agentRoom,
+ readSounds, readSoundByCommand, readDmBlockState, soundAsset, agentRoom, partyCount,
_setRetryMs(ms) { retryMs = ms || RETRY_MS; },
_setRetryMs(ms) { retryMs = ms || RETRY_MS; },
};
};
new file mode 100644
new file mode 100644
@@ -0,0 +1,67 @@
@@ -0,0 +1,67 @@
+'use strict';
+/**
+ * Agent report citations on a Rundown topic page (Agent Streams S7, design 2026-11-06 §2).
+ *
+ * The Rundown's topic pages may cite a public session report when an agent worked on the topic. Only Everyone's
+ * reports are read (Report's public list asks the sharing system), and the topic's whole name must appear as a whole
+ * phrase — the same whole-name rule the streamers' write-ups keep. A Report outage is an empty list, never a broken
+ * topic page.
+ *
+ * The read is cached for five minutes per topic: Report answers its public list for everyone, and a topic page is
+ * refreshed by the same reader many times a day.
+ */
+const REPORT_URL = () => String(process.env.OV_REPORT_URL || 'https://openvibe.report').replace(/\/+$/, '');
+const TTL_MS = 5 * 60_000;
+const cache = new Map(); // topic phrase -> { at, list, expires }
+
+function normalize(value) {
+ return String(value || '').toLowerCase().normalize('NFKD').replace(/[^a-z0-9]+/g, ' ').trim();
+}
+
+/** Whether this report speaks of the topic by its whole name (never a partial word). */
+function speaksOf(report, topic) {
+ const phrase = normalize(topic);
+ if (!phrase) return false;
+ const text = [report && report.task && report.task.title, report && report.summary, ...(report && Array.isArray(report.highlights) ? report.highlights : [])]
+ .map(normalize).filter(Boolean).join(' ');
+ return new RegExp(`(^| )${phrase.replace(/[.*+?^${}()|[\]\\]/g, '\\$&')}( |$)`).test(text);
+}
+
+function cite(report) {
+ return {
+ id: report.id,
+ title: (report.task && report.task.title) || report.summary || 'An agent session',
+ summary: report.summary || null,
+ outcome: report.outcome || 'done',
+ worked_seconds: Number(report.worked_seconds) || 0,
+ };
+}
+
+function createCitations({ fetchImpl = globalThis.fetch, log = console, now = () => Date.now(), url = REPORT_URL(), ttl = TTL_MS } = {}) {
+ async function forTopic(topic, { limit = 6 } = {}) {
+ const phrase = normalize(topic);
+ if (!phrase) return [];
+ const hit = cache.get(phrase);
+ if (hit && hit.expires > now()) return hit.list.slice(0, limit);
+ let list = [];
+ try {
+ const res = await fetchImpl(`${url}/api/v1/session-reports`, {
+ headers: { Accept: 'application/json' },
+ signal: AbortSignal.timeout(2500),
+ });
+ if (!res.ok) throw new Error(`report ${res.status}`);
+ const body = await res.json().catch(() => null);
+ const reports = body && Array.isArray(body.reports) ? body.reports : [];
+ list = reports.filter((r) => speaksOf(r, phrase)).map(cite);
+ } catch (err) {
+ log.warn(`[Rundown] agent citations for "${phrase}" unavailable: ${(err && err.message) || err}`);
+ list = [];
+ }
+ cache.set(phrase, { at: now(), expires: now() + ttl, list });
+ if (cache.size > 1000) cache.clear();
+ return list.slice(0, Math.max(1, Math.min(20, Number(limit) || 6)));
+ }
+ return { forTopic, clear: () => cache.clear() };
+}
+
+module.exports = { createCitations, speaksOf, normalize, TTL_MS };
@@ -30,6 +30,7 @@ const read = require('./read');
@@ -30,6 +30,7 @@ const read = require('./read');
const browse = require('./browse');
const browse = require('./browse');
const settings = require('./settings');
const settings = require('./settings');
const del = require('./delete');
const del = require('./delete');
+const { createCitations } = require('./citations');
const router = express.Router();
const router = express.Router();
const fail = (res, err, msg) => {
const fail = (res, err, msg) => {
@@ -79,8 +80,14 @@ router.get('/topics', optionalAuth, async (req, res) => {
@@ -79,8 +80,14 @@ router.get('/topics', optionalAuth, async (req, res) => {
} catch (err) { fail(res, err, 'Failed to load the topics'); }
} catch (err) { fail(res, err, 'Failed to load the topics'); }
});
});
router.get('/t/:slug', optionalAuth, async (req, res) => {
router.get('/t/:slug', optionalAuth, async (req, res) => {
- try { const t = await read.topic(req.params.slug, { viewer: await viewerOf(req) }); if (!t) return res.status(404).json({ error: 'No such topic' }); fresh(res); res.json(t); }
- catch (err) { fail(res, err, 'Failed to load the topic'); }
+ try {
+ const t = await read.topic(req.params.slug, { viewer: await viewerOf(req) });
+ if (!t) return res.status(404).json({ error: 'No such topic' });
+ // S7: a public agent report may cite the topic when the agent worked on it. Report being down is no citations.
+ t.citations = await createCitations().forTopic((t.topic && t.topic.title) || t.title);
+ fresh(res);
+ res.json(t);
+ } catch (err) { fail(res, err, 'Failed to load the topic'); }
});
});
router.get('/e/:streamId', optionalAuth, async (req, res) => {
router.get('/e/:streamId', optionalAuth, async (req, res) => {
try { const e = await read.episode(req.params.streamId, { viewer: await viewerOf(req) }); if (!e) return res.status(404).json({ error: 'No such episode' }); fresh(res); res.json(e); }
try { const e = await read.episode(req.params.streamId, { viewer: await viewerOf(req) }); if (!e) return res.status(404).json({ error: 'No such episode' }); fresh(res); res.json(e); }
new file mode 100644
new file mode 100644
@@ -0,0 +1,298 @@
@@ -0,0 +1,298 @@
+'use strict';
+/**
+ * Agent streams S7, Live's part (design 2026-11-06): schedules, the desk heartbeat, the miss and the Go; the party
+ * count on a stream; and the robot camera as a second scene.
+ *
+ * - a schedule shows on the channel's Agents tab 24 hours before its time and nowhere earlier;
+ * - reminders are opt-in, and the answer is a count, never a name;
+ * - a due schedule nobody claimed says "didn't start: the desk was offline" and starts nothing;
+ * - an online desk claims it and gets the one publish ticket; a chatplays schedule waits for the owner's Go;
+ * - the stream page carries the party count (a number) and, on a channel bound to a robot, the robot scene.
+ *
+ * Sign-in is stubbed with an x-test-user header, as agent-streams-s2.test.js does; everything after that is production.
+ */
+const assert = require('assert');
+const fs = require('node:fs');
+const path = require('node:path');
+const http = require('http');
+const express = require('express');
+
+process.env.NODE_ENV = 'test';
+const quiet = console.log;
+console.log = (...a) => { if (!/^\[/.test(String(a[0]))) quiet(...a); };
+console.warn = () => {};
+console.error = () => {};
+
+const SID = {
+ ann: 'usr_01JAB2C3D4E5F6G7H8J9K0MNA1',
+ bob: 'usr_01JAB2C3D4E5F6G7H8J9K0MNB2',
+};
+
+let failures = 0;
+async function check(name, fn) {
+ try { await fn(); quiet(' ✓', name); }
+ catch (e) { failures++; quiet(' ✗', name, '\n ', (e.stack || String(e)).split('\n').slice(0, 8).join('\n ')); }
+}
+
+(async () => {
+ const db = require('../server/db/database');
+ await db.initDb();
+ const raw = db.getDb();
+ const addUser = async (id, username, role) => await raw.prepare(
+ `INSERT INTO users (id, username, display_name, email, password_hash, role, is_owner) OVERRIDING SYSTEM VALUE
+ VALUES (?, ?, ?, ?, 'x', ?, 0)`).run(id, username, username, `${username}@x`, role);
+ await addUser(1, 'ann', 'streamer');
+ await addUser(2, 'bob', 'user');
+ await raw.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (1, 'network', '101', ?)").run(SID.ann);
+ await raw.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (2, 'network', '102', ?)").run(SID.bob);
+
+ const config = require('../server/config');
+ config.baseUrl = 'https://openvibe.live';
+ const service = require('../server/agent-streams/service');
+ const schedules = require('../server/agent-streams/schedules');
+ require('../server/agent-streams/sharing')._setNetwork({ json: async () => ({}) });
+ const ann = await db.getUserById(1);
+ const bob = await db.getUserById(2);
+ const channel = await db.ensureChannel(1);
+
+ const app = express();
+ app.use(express.json());
+ app.use((req, res, next) => { req.user = { 1: ann, 2: bob }[req.get('x-test-user')] || null; next(); });
+ app.use('/api/streams/agent', require('../server/agent-streams/routes'));
+ const server = await new Promise((ok) => { const s = http.createServer(app).listen(0, '127.0.0.1', () => ok(s)); });
+ const base = `http://127.0.0.1:${server.address().port}`;
+ const call = async (method, url, { as = 1, body, headers = {} } = {}) => {
+ const res = await fetch(`${base}${url}`, {
+ method,
+ headers: { 'Content-Type': 'application/json', 'x-test-user': String(as), ...headers },
+ body: body === undefined ? undefined : JSON.stringify(body),
+ });
+ const text = await res.text();
+ let data = {};
+ try { data = text ? JSON.parse(text) : {}; } catch { data = { text }; }
+ return { status: res.status, data, headers: res.headers };
+ };
+ const makeSchedule = async ({ title = 'Fix the login page', at = Date.now() + 2 * 60 * 60 * 1000, mode = 'watch' } = {}) => {
+ const r = await call('POST', '/api/streams/agent/schedules', {
+ body: {
+ title,
+ task: { source: 'codes', ref: 'tsk_1', title: 'Fix the login page', url: 'https://openvibe.codes/tasks/1' },
+ mode,
+ starts_at: new Date(at).toISOString(),
+ },
+ });
+ assert.strictEqual(r.status, 201, JSON.stringify(r.data));
+ return r.data.schedule;
+ };
+ const due = async (id) => await db.run("UPDATE agent_stream_schedules SET starts_at = now() - interval '2 minutes' WHERE id = ?", [id]);
+
+ quiet('agent streams S7: schedules');
+ const soon = await makeSchedule({ title: 'Soon', at: Date.now() + 23 * 60 * 60 * 1000 });
+ const later = await makeSchedule({ title: 'Later', at: Date.now() + 25 * 60 * 60 * 1000 });
+ await check('a schedule answers the contract shape', async () => {
+ assert.match(soon.id, /^scd_[0-9A-HJKMNP-TV-Z]{26}$/);
+ assert.strictEqual(soon.owner, SID.ann);
+ assert.strictEqual(soon.channel, 'ann');
+ assert.strictEqual(soon.visibility, 'public');
+ assert.strictEqual(soon.state, 'scheduled');
+ assert.strictEqual(soon.reminders, 0);
+ assert.ok(soon.page_url.endsWith('/@ann/agents'));
+ });
+ await check('Today sees it 24 hours ahead, not a day early', async () => {
+ const r = await call('GET', '/api/streams/agent/schedules?channel=ann', { as: 2 });
+ assert.strictEqual(r.status, 200, JSON.stringify(r.data));
+ const ids = r.data.schedules.map((s) => s.id);
+ assert.ok(ids.includes(soon.id), '23 hours ahead is shown');
+ assert.ok(!ids.includes(later.id), '25 hours ahead is not');
+ const mine = await call('GET', '/api/streams/agent/schedules?mine=1', { as: 1 });
+ assert.ok(mine.data.schedules.some((s) => s.id === later.id), 'the owner sees their whole list');
+ });
+ await check('reminders are opt-in only, and the answer is a count', async () => {
+ const before = await call('GET', `/api/streams/agent/schedules/${soon.id}`, { as: 2 });
+ assert.strictEqual(before.data.schedule.reminded, false);
+ const r = await call('POST', `/api/streams/agent/schedules/${soon.id}/remind`, { as: 2 });
+ assert.strictEqual(r.status, 200, JSON.stringify(r.data));
+ assert.strictEqual(r.data.reminders, 1);
+ assert.strictEqual(r.data.reminded, true);
+ const asAnn = await call('GET', `/api/streams/agent/schedules/${soon.id}`, { as: 1 });
+ assert.strictEqual(asAnn.data.schedule.reminded, false, 'ann did not opt in');
+ assert.strictEqual(asAnn.data.schedule.reminders, 1, 'the count is there');
+ const all = JSON.stringify(asAnn.data);
+ assert.ok(!all.includes('bob'), 'no name in a schedule view');
+ const signedOut = await call('POST', `/api/streams/agent/schedules/${soon.id}/remind`, { as: 0 });
+ assert.strictEqual(signedOut.status, 401);
+ // Only a schedule the person may see: one beyond the 24-hour window, or a private one, is a 404.
+ const tooEarly = await call('POST', `/api/streams/agent/schedules/${later.id}/remind`, { as: 2 });
+ assert.strictEqual(tooEarly.status, 404, 'a schedule outside its public window is not offered');
+ await db.run("UPDATE agent_stream_schedules SET visibility = 'private' WHERE id = ?", [soon.id]);
+ const hidden = await call('POST', `/api/streams/agent/schedules/${soon.id}/remind`, { as: 2 });
+ assert.strictEqual(hidden.status, 404, "a private schedule's id is not an invitation");
+ await db.run("UPDATE agent_stream_schedules SET visibility = 'public' WHERE id = ?", [soon.id]);
+ });
+ await check('a due schedule with the desk offline says so and starts nothing', async () => {
+ const missed = await makeSchedule({ title: 'Offline desk' });
+ await due(missed.id);
+ const streamsBefore = (await service.listByUser(1)).length;
+ const out = await schedules.startDue();
+ assert.ok(out.missed >= 1, JSON.stringify(out));
+ const after = await call('GET', `/api/streams/agent/schedules/${missed.id}`, { as: 1 });
+ assert.strictEqual(after.data.schedule.state, 'missed');
+ assert.strictEqual(after.data.schedule.missed_reason, 'desk_offline');
+ assert.strictEqual(after.data.schedule.agent_stream_id, null);
+ assert.strictEqual((await service.listByUser(1)).length, streamsBefore, 'nothing ran');
+ });
+ await check('an online desk claims it and gets one publish ticket; ending the stream ends the schedule', async () => {
+ const ready = await makeSchedule({ title: 'Online desk' });
+ await schedules.heartbeat(ann, 'node/0.41.0');
+ await due(ready.id);
+ const r = await call('POST', `/api/streams/agent/schedules/${ready.id}/claim`, { as: 1 });
+ assert.strictEqual(r.status, 201, JSON.stringify(r.data));
+ assert.match(r.data.publish.ticket, /^wpt_/);
+ assert.match(r.data.publish.whip_url, /\/whip\/agent\/ags_/);
+ assert.strictEqual(r.data.schedule.state, 'live');
+ assert.strictEqual(r.data.stream.id, r.data.schedule.agent_stream_id);
+ await service.markPublishing(await service.getByPublicId(r.data.stream.id));
+ const ended = await call('POST', `/api/streams/agent/${r.data.stream.id}/end`, { as: 1 });
+ assert.strictEqual(ended.status, 200, JSON.stringify(ended.data));
+ const after = await call('GET', `/api/streams/agent/schedules/${ready.id}`, { as: 1 });
+ assert.strictEqual(after.data.schedule.state, 'ended');
+ assert.ok(after.data.schedule.ended_at);
+ });
+ await check('a chat plays schedule waits for the owner\'s Go and starts nothing before it', async () => {
+ const plays = await makeSchedule({ title: 'Chat plays a tiny game', mode: 'chatplays' });
+ await schedules.heartbeat(ann);
+ await due(plays.id);
+ const before = (await service.listByUser(1)).length;
+ const refused = await call('POST', `/api/streams/agent/schedules/${plays.id}/claim`, { as: 1 });
+ assert.strictEqual(refused.status, 409, JSON.stringify(refused.data));
+ assert.strictEqual(refused.data.code, 'waits_for_go');
+ const waiting = await call('GET', `/api/streams/agent/schedules/${plays.id}`, { as: 1 });
+ assert.strictEqual(waiting.data.schedule.waits_for_go, true);
+ assert.strictEqual(waiting.data.schedule.missed_reason, 'waits_for_go');
+ assert.strictEqual(waiting.data.schedule.state, 'scheduled');
+ assert.strictEqual((await service.listByUser(1)).length, before, 'nothing ran without Go');
+ const go = await call('POST', `/api/streams/agent/schedules/${plays.id}/go`, { as: 1 });
+ assert.strictEqual(go.status, 201, JSON.stringify(go.data));
+ assert.strictEqual(go.data.schedule.state, 'live');
+ });
+ await check('only the owner may cancel; a cancelled schedule leaves the window', async () => {
+ const own = await makeSchedule({ title: 'Cancel me' });
+ const other = await call('DELETE', `/api/streams/agent/schedules/${own.id}`, { as: 2 });
+ assert.strictEqual(other.status, 403);
+ const ok = await call('DELETE', `/api/streams/agent/schedules/${own.id}`, { as: 1 });
+ assert.strictEqual(ok.status, 200, JSON.stringify(ok.data));
+ assert.strictEqual(ok.data.schedule.state, 'cancelled');
+ });
+
+ quiet('agent streams S7: parties and the robot scene');
+ const partyStream = await service.create({
+ user: ann, title: 'A party stream', task: { source: 'codes', ref: 'tsk_party', title: 'A party' }, visibility: 'public',
+ });
+ await check('the stream carries the party count and never a name', async () => {
+ const chat = require('../server/chat/chat-client');
+ const original = chat.partyCount;
+ chat.partyCount = async () => ({ ok: true, count: 3 });
+ try {
+ const r = await call('GET', `/api/streams/agent/${partyStream.public_id}`, { as: 2 });
+ assert.strictEqual(r.status, 200, JSON.stringify(r.data));
+ assert.deepStrictEqual(r.data.parties, { count: 3 });
+ const text = JSON.stringify(r.data);
+ assert.ok(!/host|people|names|members/i.test(text), 'no party member list');
+ } finally { chat.partyCount = original; }
+ process.env.OV_CHAT_INTERNAL_URL = '';
+ const noChat = await call('GET', `/api/streams/agent/${partyStream.public_id}`, { as: 2 });
+ assert.strictEqual(noChat.data.parties, null, 'Chat down: no count, the stream still answers');
+ process.env.OV_CHAT_INTERNAL_URL = 'http://127.0.0.1:4400';
+ });
+ await check('a channel bound to a robot shows the robot camera as a second scene', async () => {
+ process.env.LIVE_BOT_EMBED = '1';
+ try {
+ await raw.prepare('UPDATE channels SET bot_robot_id = ? WHERE id = ?').run('rob_sim123', channel.id);
+ const r = await call('GET', `/api/streams/agent/${partyStream.public_id}`, { as: 2 });
+ assert.strictEqual(r.status, 200, JSON.stringify(r.data));
+ assert.ok(r.data.stream.robot_scene, 'the robot scene is offered');
+ assert.strictEqual(r.data.stream.robot_scene.robot_id, 'rob_sim123');
+ assert.match(r.data.stream.robot_scene.embed_url, /\/panel\/rob_sim123\/embed$/);
+ const fragment = fs.readFileSync(path.join(__dirname, '../public/fragments/agent-stream.html'), 'utf8');
+ assert.match(fragment, /id="agent-scenes"/);
+ assert.match(fragment, /id="agent-robot"/);
+ const js = fs.readFileSync(path.join(__dirname, '../public/js/agent-stream.js'), 'utf8');
+ assert.match(js, /function renderScenes/);
+ assert.match(js, /robot_scene/);
+ const css = fs.readFileSync(path.join(__dirname, '../public/css/features/agent-stream.css'), 'utf8');
+ assert.match(css, /\.agent-robot/);
+ } finally {
+ await raw.prepare('UPDATE channels SET bot_robot_id = NULL WHERE id = ?').run(channel.id);
+ delete process.env.LIVE_BOT_EMBED;
+ }
+ });
+ await check('the kill switch reaches the robot through Bot', async () => {
+ process.env.BOT_KILL_URL = 'https://bot.test';
+ process.env.BOT_KILL_SECRET = 'kill-secret';
+ try {
+ await raw.prepare('UPDATE channels SET bot_robot_id = ? WHERE id = ?').run('rob_sim123', channel.id);
+ const calls = [];
+ const out = await require('../server/bot/kill').stop(
+ { channel_id: channel.id, public_id: partyStream.public_id },
+ { fetchImpl: async (url, opts) => { calls.push({ url, opts }); return { ok: true, status: 200 }; } }
+ );
+ assert.deepStrictEqual(out, { stopped: true, robot_id: 'rob_sim123' });
+ assert.strictEqual(calls.length, 1);
+ assert.strictEqual(calls[0].url, 'https://bot.test/internal/robots/rob_sim123/estop');
+ assert.strictEqual(calls[0].opts.headers['X-OV-Bot-Kill'], 'kill-secret');
+ assert.strictEqual(JSON.parse(calls[0].opts.body).reason, 'agent_stream_ended');
+ } finally {
+ await raw.prepare('UPDATE channels SET bot_robot_id = NULL WHERE id = ?').run(channel.id);
+ delete process.env.BOT_KILL_URL;
+ delete process.env.BOT_KILL_SECRET;
+ }
+ });
+
+ await check('the channel Agents tab draws the schedules above the cards', async () => {
+ const fragment = fs.readFileSync(path.join(__dirname, '../public/fragments/channel.html'), 'utf8');
+ assert.match(fragment, /id="ch-agents-schedules"/);
+ const cards = fs.readFileSync(path.join(__dirname, '../public/js/agent-cards.js'), 'utf8');
+ assert.match(cards, /scheduleCard/);
+ assert.match(cards, /Didn't start: the desk was offline/);
+ const channelJs = fs.readFileSync(path.join(__dirname, '../public/js/app-channel.js'), 'utf8');
+ assert.match(channelJs, /fillSchedules/);
+ });
+ await check('a Rundown topic cites a public agent report by its whole name only', async () => {
+ const { createCitations, speaksOf } = require('../server/rundown/citations');
+ assert.ok(speaksOf({ task: { title: 'Fix the login page' }, summary: 'Done.' }, 'Login page'));
+ assert.ok(!speaksOf({ task: { title: 'Fix the login page' } }, 'login pag'), 'a partial name is not the topic');
+ assert.ok(!speaksOf({ task: { title: 'Fix the login page' } }, 'ogin'), 'a name inside a word is not the topic');
+ const asked = [];
+ const fetchImpl = async (url) => {
+ asked.push(url);
+ return { ok: true, json: async () => ({ reports: [
+ { id: 'rps_01JAB2C3D4E5F6G7H8J9K0MNPS', task: { title: 'Fix the login page' }, summary: 'Done.', outcome: 'done' },
+ { id: 'rps_01JAB2C3D4E5F6G7H8J9K0MNPT', task: { title: 'Fix the checkout page' }, summary: 'Done.', outcome: 'done' },
+ ] }) };
+ };
+ const cites = createCitations({ fetchImpl, url: 'https://report.test', log: { warn() {} } });
+ const first = await cites.forTopic('Login page');
+ assert.deepStrictEqual(first.map((c) => c.id), ['rps_01JAB2C3D4E5F6G7H8J9K0MNPS']);
+ assert.strictEqual(first[0].title, 'Fix the login page');
+ await cites.forTopic('Login page');
+ assert.strictEqual(asked.length, 1, 'the topic is cached');
+ assert.deepStrictEqual(await cites.forTopic('Login page'), first);
+ const down = createCitations({ fetchImpl: async () => { throw new Error('down'); }, url: 'https://report.test', log: { warn() {} } });
+ down.clear();
+ assert.deepStrictEqual(await down.forTopic('Login page'), [], 'Report down: no citations, no broken page');
+ const javascript = fs.readFileSync(path.join(__dirname, '../public/js/rundown.js'), 'utf8');
+ assert.match(javascript, /Agents that worked on this/);
+ assert.match(javascript, /openvibe\.report\/s\//);
+ });
+
+ await check('the schedule card survives 390 px', async () => {
+ const css = fs.readFileSync(path.join(__dirname, '../public/css/features/agent-stream.css'), 'utf8');
+ assert.match(css, /@media \(max-width: 640px\)[\s\S]*agent-scene/);
+ assert.match(css, /\.agent-schedules \{ display: grid/);
+ });
+
+ server.close();
+ await db.closeDb?.();
+ quiet(failures ? `\n${failures} check(s) failed` : '\nAll agent stream S7 checks passed');
+ process.exit(failures ? 1 : 0);
+})().catch((e) => { quiet(e.stack || String(e)); process.exit(1); });
@@ -351,6 +351,7 @@ async function check(name, fn) {
@@ -351,6 +351,7 @@ async function check(name, fn) {
'server/social/preview.js': 'OpenVibe.Network public profile API (configured OV_NETWORK_INTERNAL_URL, fixed path, username checked); every other preview goes through egress',
'server/social/preview.js': 'OpenVibe.Network public profile API (configured OV_NETWORK_INTERNAL_URL, fixed path, username checked); every other preview goes through egress',
'server/monetization/inventory-client.js': 'OpenVibe.Inventory API for Live cosmetics (configured INVENTORY_URL)',
'server/monetization/inventory-client.js': 'OpenVibe.Inventory API for Live cosmetics (configured INVENTORY_URL)',
'server/bot/bot-client.js': 'OpenVibe.Bot command API for a bound channel\'s chat robot commands (configured LIVE_BOT_INTERNAL_URL)',
'server/bot/bot-client.js': 'OpenVibe.Bot command API for a bound channel\'s chat robot commands (configured LIVE_BOT_INTERNAL_URL)',
+ 'server/bot/kill.js': 'OpenVibe.Bot internal e-stop for the robot a killed agent stream is bound to (configured BOT_KILL_URL, fixed path, robot id checked)',
'server/cam/cam-client.js': 'OpenVibe.Cam stills API for the cam button, as the viewer (configured CAM_URL)',
'server/cam/cam-client.js': 'OpenVibe.Cam stills API for the cam button, as the viewer (configured CAM_URL)',
'server/chat/insight-client.js': 'OpenVibe.Chat public chat-AI reads (configured OV_CHAT_INTERNAL_URL, fixed paths, ids validated)',
'server/chat/insight-client.js': 'OpenVibe.Chat public chat-AI reads (configured OV_CHAT_INTERNAL_URL, fixed paths, ids validated)',
'server/chat/chat-client.js': 'OpenVibe.Chat typed ingress (configured OV_CHAT_INTERNAL_URL, fixed /internal/chat/* paths)',
'server/chat/chat-client.js': 'OpenVibe.Chat typed ingress (configured OV_CHAT_INTERNAL_URL, fixed /internal/chat/* paths)',
@@ -360,6 +361,7 @@ async function check(name, fn) {
@@ -360,6 +361,7 @@ async function check(name, fn) {
'server/auth/identity-sync.js': 'OpenVibe.Network (configured)',
'server/auth/identity-sync.js': 'OpenVibe.Network (configured)',
'server/auth/account-data.js': 'OpenVibe.Network (configured): export parts and deletion confirmations',
'server/auth/account-data.js': 'OpenVibe.Network (configured): export parts and deletion confirmations',
'server/agent-streams/sharing.js': 'OpenVibe.Network (configured): the sharing system\'s name lookup and person resolution',
'server/agent-streams/sharing.js': 'OpenVibe.Network (configured): the sharing system\'s name lookup and person resolution',
+ 'server/rundown/citations.js': 'OpenVibe.Report public session-report list (OV_REPORT_URL, configured; fixed path, no URL from a person)',
'server/ai/viewer-quota.js': 'OpenVibe.AI (configured OV_AI_INTERNAL_URL): a streamer\'s daily AI-viewer cap',
'server/ai/viewer-quota.js': 'OpenVibe.AI (configured OV_AI_INTERNAL_URL): a streamer\'s daily AI-viewer cap',
'server/ai/byo-credentials.js': 'OpenVibe.AI (configured OV_AI_INTERNAL_URL): a streamer\'s own key stored there',
'server/ai/byo-credentials.js': 'OpenVibe.AI (configured OV_AI_INTERNAL_URL): a streamer\'s own key stored there',
'server/monetization/billing-client.js': 'OpenVibe.Billing (configured)',
'server/monetization/billing-client.js': 'OpenVibe.Billing (configured)',