openvibers/openvibe.live Commits
Agent Streams S5: Co-pilot, the four controls and their log (#642)
Changes
@@@ -163,3 +163,35 @@ counted "More (N)" after twelve.**Not in S2:** the agent's own chat room (`agent:<ags_id>`), which needs a Chat-side room kind, and the voice work ofS4. Chat is for people; the agent reading it is S5.++++## S5: Co-pilot (Contracts 0.303.0)++++A stream can be in **Co-pilot** mode, which implies Narrated and turns on the agent's reading of its chat room. The++owner chooses it with the voice controls (`PUT /api/streams/agent/<ags_id>/voice { "mode": "copilot" }`); it needs a++public or unlisted stream and a task's service that says it can read a digest:++++```++POST /api/streams/agent/<ags_id>/copilot { "ready": true } (the task's service, with its publish token)++```++++Live makes the Chat room `agent:<ags_id>` when the stream becomes public or unlisted; a private stream has none.++Codes and Actor read Chat's ten-item, 1,500-character digest between steps, always inside the fixed untrusted frame,++and post an answer line, a poll or a credit ("Suggested by @mira") through Chat's `chat.agent-digest@1` and++`chat.agent-poll@1`.++++**The four controls.** The owner and the channel's moderators have Pause, Curtain, Mute and End on the stream page,++in the Dock and on the phone:++++```++POST /api/streams/agent/<ags_id>/control { "control": "pause", "paused": true }++ { "control": "curtain", "up": true }++ { "control": "mute", "muted": true }++ { "control": "end" }++GET /api/streams/agent/<ags_id>/controls the log (newest first), for the owner and moderators++```++++Pause is relayed to the task's service in the state it polls (`paused`, `paused_by`); it stops at the next safe point.++A moderator may not resume a pause the owner set. The overlay says "Paused by @mod". Every press is stored in++`agent_stream_controls` with who pressed it, which control and when; End also stops the room's Co-pilot reading while++the task keeps running off camera. Curtain is S1's curtain, mute is S4's voice mute, and all four act within their++design limits (2 s for pause and end, 1 s for curtain and mute).nnew file mode 100644@@@ -0,0 +1,31 @@++-- phase: expand++-- Agent streams, S5 (design 2026-11-06, plan-state-2026-10-22/AGENT-STREAMS-S5-COPILOT-DESIGN-2026-11-06.md):++-- Co-pilot. The stream's mode can be `copilot` (Contracts 0.303.0, live.agent-stream@1), which implies Narrated and++-- turns on the chat digest reads and polls on OpenVibe.Chat. The owner and the channel's moderators have four++-- controls (pause, curtain, mute, end); every press is logged with who pressed it.++--++-- The pause is the task's service reading it and stopping at the next safe point: `paused`, `paused_by` and++-- `paused_by_role` travel in the state the desk polls. A moderator may not resume a pause the owner set.++--++-- The task's service declares it can read a digest (`copilot_ready`): Codes and Actor do, a desk does not. Co-pilot++-- needs a public or unlisted stream, a ready service and the narrated voice.++--++-- ADD-only: an older release ignores the columns and the table.++ALTER TABLE agent_streams ADD COLUMN IF NOT EXISTS copilot_ready boolean NOT NULL DEFAULT false;++ALTER TABLE agent_streams ADD COLUMN IF NOT EXISTS paused boolean NOT NULL DEFAULT false;++ALTER TABLE agent_streams ADD COLUMN IF NOT EXISTS paused_by text;++ALTER TABLE agent_streams ADD COLUMN IF NOT EXISTS paused_by_role text CHECK (paused_by_role IN ('owner', 'moderator'));++ALTER TABLE agent_streams ADD COLUMN IF NOT EXISTS paused_at timestamptz;++++-- Every control the owner and the channel's moderators pressed over a Co-pilot stream, and who pressed it.++CREATE TABLE IF NOT EXISTS agent_stream_controls (++ id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,++ managed_stream_id bigint NOT NULL REFERENCES managed_streams(id) ON DELETE CASCADE,++ control text NOT NULL CHECK (control IN ('pause', 'curtain', 'mute', 'end')),++ by_handle text NOT NULL,++ by_role text NOT NULL CHECK (by_role IN ('owner', 'moderator')),++ paused boolean, -- for a pause: true = paused, false = resumed++ paused_by text, -- whose pause is in force after this press++ at timestamptz NOT NULL DEFAULT now()++);++CREATE INDEX IF NOT EXISTS agent_stream_controls_stream ON agent_stream_controls (managed_stream_id, id DESC);@@@ -159,3 +159,15 @@.home-agents-title { margin: 0; font-size: 1.05rem; font-weight: 800; }.home-agents-note { color: var(--text-muted); font-size: 0.78rem; }.home-agents .agent-cards { grid-template-columns: repeat(auto-fill, minmax(200px, 1fr)); }++++/* S5: the moderators' bar and the paused line (Agent Streams S5). */++.agent-controls { margin: 0 0 12px; padding: 10px 12px; border: 1px solid var(--border, rgba(255,255,255,.12)); border-radius: 12px;++ background: var(--bg-card, rgba(18,24,38,.72)); }++.agent-controls-title { margin: 0 0 8px; font-size: .8rem; letter-spacing: .06em; text-transform: uppercase; color: var(--text-secondary, #96a7c2); }++.agent-controls-row { display: flex; flex-wrap: wrap; gap: 8px; }++.agent-controls-row .btn { min-height: 44px; }++.agent-controls-log { margin-top: 8px; }++.agent-controls-log summary { cursor: pointer; color: var(--text-secondary, #96a7c2); font-size: .85rem; }++.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); } }@@@ -4,6 +4,20 @@<h1 id="agent-title" class="agent-title">Agent stream</h1><p id="agent-task" class="agent-task"></p></header>++ <section class="agent-controls" id="agent-controls" aria-labelledby="agent-controls-title" hidden>++ <h2 class="agent-controls-title" id="agent-controls-title">Stream controls</h2>++ <div class="agent-controls-row">++ <button type="button" class="btn btn-outline" id="agent-ctl-pause">Pause the agent</button>++ <button type="button" class="btn btn-outline" id="agent-ctl-curtain">Curtain</button>++ <button type="button" class="btn btn-outline" id="agent-ctl-mute">Mute voice</button>++ <button type="button" class="btn btn-danger" id="agent-ctl-end">End</button>++ </div>++ <p class="agent-hint" id="agent-controls-state"></p>++ <details class="agent-controls-log" id="agent-controls-log" hidden>++ <summary>Control log</summary>++ <ol class="agent-controls-log-list" id="agent-controls-log-list"></ol>++ </details>++ </section><div class="agent-layout"><div class="agent-main"><div class="agent-stage" id="agent-stage">@@@ -46,7 +60,9 @@<select id="agent-voice-mode" class="agent-select"><option value="watch">Watch — the agent stays quiet</option><option value="narrated">Narrated — the agent says what it is doing</option>++ <option value="copilot">Co-pilot — it also reads the chat and may answer</option></select>++ <p class="agent-note" id="agent-copilot-note" hidden>Co-pilot needs a public or unlisted stream and a task's service that can read a digest.</p></div><div class="agent-field"><label class="agent-field-label" for="agent-voice-personality">How it says it</label>@@@ -44,7 +44,7 @@const host = $('agent-embed-overlay');if (!host || !window.AgentOverlay) return;if (!overlay) overlay = window.AgentOverlay.create({ host, title: stream.title, initial: steps });-- overlay.update({ steps, task: stream.task, workedSeconds: stream.worked_seconds, state: stream.state });++ overlay.update({ steps, task: stream.task, workedSeconds: stream.worked_seconds, state: stream.state, pausedBy: streamView && streamView.__pausedBy });}function pollSteps() {@@@ -71,6 +71,7 @@ticket = data.view_ticket || ticket;streamState = stream.state;streamView = stream;++ streamView.__pausedBy = data.paused_by || null;const title = $('agent-embed-title');if (title) title.textContent = stream.title || '';@@@ -82,6 +82,9 @@const dot = el('span', 'agent-overlay-dot');const stepTitle = el('span', 'agent-overlay-step-title');stepLine.append(dot, stepTitle);++ // S5: whose pause is in force ("Paused by @mod"), so a viewer knows why the work stopped.++ const pausedLine = el('p', 'agent-overlay-paused');++ pausedLine.hidden = true;const ticks = el('div', 'agent-overlay-ticks');ticks.setAttribute('role', 'list');@@@ -100,7 +103,7 @@const sheetList = el('ol', 'agent-overlay-sheet-list');sheet.appendChild(sheetList);-- body.append(taskLine, stepLine, captionLine, ticks, foot);++ body.append(taskLine, stepLine, pausedLine, captionLine, ticks, foot);root.append(foldBtn, body, sheet);host.appendChild(root);@@@ -178,6 +181,9 @@voiceLine.textContent = '';voiceLine.hidden = true;}++ const pausedBy = data.pausedBy ? String(data.pausedBy).replace(/^@/, '') : '';++ pausedLine.textContent = pausedBy ? `Paused by @${pausedBy}` : '';++ pausedLine.hidden = !pausedBy;const cur = current(steps);const task = data.task && data.task.title ? data.task.title : (options.title || 'An AI agent is working');taskLine.textContent = task;@@@ -26,6 +26,7 @@let linesTimer = null;let overlay = null;let chaptersFor = null; // the stream id whose chapters are rendered (they never change while the page lives)++ let controlsLogFor = null; // the stream id whose control log is renderedlet generation = 0; // bumped on every page leave, so a slow request from the last page changes nothingfunction $(id) { return document.getElementById(id); }@@@ -45,6 +46,7 @@voice = null;voicesLoaded = false;chaptersFor = null;++ controlsLogFor = null;state = null;ticket = null;}@@@ -145,6 +147,7 @@state: data.stream.state,caption: captionText(),voice,++ pausedBy: data.paused_by || null,});}@@@ -331,6 +334,7 @@mountPlays(data);ensureOverlay(data);renderOwner(data);++ renderControls(data);renderVoice();loadVoices();scheduleRenew(data);@@@ -341,6 +345,51 @@if (stream.state === 'ended') renderRecording(data);}++ /** S5: the moderators' bar. The owner and the channel's moderators see it; the owner also sees the log. */++ function renderControls(data) {++ const bar = $('agent-controls');++ if (!bar) return;++ const allowed = !!(data && (data.can_manage || data.can_moderate));++ bar.hidden = !allowed;++ if (!allowed) return;++ const paused = !!(data && data.paused);++ const pauseBtn = $('agent-ctl-pause');++ if (pauseBtn) {++ pauseBtn.textContent = paused ? 'Resume the agent' : 'Pause the agent';++ // A moderator cannot resume a pause the owner set (the server refuses it too).++ pauseBtn.disabled = paused && data.paused_by_role === 'owner' && !data.can_manage;++ }++ const curtain = data && data.stream ? data.stream.curtain : null;++ const curtainBtn = $('agent-ctl-curtain');++ if (curtainBtn) curtainBtn.textContent = curtain && curtain.up ? 'Lift the curtain' : 'Curtain';++ const muteBtn = $('agent-ctl-mute');++ if (muteBtn) muteBtn.textContent = voice && voice.muted ? 'Unmute voice' : 'Mute voice';++ const stateEl = $('agent-controls-state');++ if (stateEl) {++ if (paused && data.paused_by) stateEl.textContent = `Paused by @${data.paused_by}`;++ else if (curtain && curtain.up) stateEl.textContent = 'The curtain is up: viewers see a private screen.';++ else if (voice && voice.muted) stateEl.textContent = 'The voice is muted.';++ else stateEl.textContent = '';++ }++ const log = $('agent-controls-log');++ if (log && data.can_manage && data.stream && controlsLogFor !== data.stream.id) {++ controlsLogFor = data.stream.id;++ log.hidden = false;++ const list = $('agent-controls-log-list');++ if (list) list.replaceChildren();++ api(`/streams/agent/${encodeURIComponent(data.stream.id)}/controls`).then((out) => {++ if (!out || !Array.isArray(out.controls) || !list) return;++ if (!out.controls.length) { list.appendChild(Object.assign(document.createElement('li'), { textContent: 'No control has been pressed.' })); return; }++ for (const entry of out.controls) {++ const li = document.createElement('li');++ const when = new Date(entry.at);++ li.textContent = `${entry.control} by @${entry.by.handle} (${entry.by.role}) · ${when.toLocaleString()}`;++ list.appendChild(li);++ }++ }).catch(() => { controlsLogFor = null; });++ } else if (log && !data.can_manage) log.hidden = true;++ }++function renderOwner(data) {const panel = $('agent-owner');if (!panel) return;@@@ -385,7 +434,7 @@const panel = $('agent-owner');if (!panel || panel.hidden || !voice) return;const mode = $('agent-voice-mode');-- if (mode) mode.value = voice.mode === 'narrated' ? 'narrated' : 'watch';++ if (mode) mode.value = ['narrated', 'copilot'].includes(voice.mode) ? voice.mode : 'watch';const personality = $('agent-voice-personality');if (personality) personality.value = voice.personality || 'calm';const own = $('agent-voice-own');@@@ -404,6 +453,8 @@state.textContent = 'voice paused: allowance spent';} else if (voice.muted) {state.textContent = 'The voice is muted.';++ } else if (voice.mode === 'copilot') {++ state.textContent = `Co-pilot, in a ${voice.personality || 'calm'} voice: it reads the chat between steps.`;} else if (voice.mode === 'narrated') {state.textContent = `Narrated, in a ${voice.personality || 'calm'} voice.`;} else {@@@ -474,6 +525,7 @@ensureOverlay(next);renderStatus(next.stream, liveViewers);renderOwner(next);++ renderControls(next);schedulePoll(id);} catch { if (gen === generation) schedulePoll(id); }}, 20000);@@@ -554,6 +606,54 @@} catch { /* the desk still owns the curtain */ }});++ // S5: the moderators' bar, four controls, every press logged by the server.++ const pressControl = async (body) => {++ const id = idOf();++ if (!id) return;++ const view = state || {};++ try {++ const out = await api(`/streams/agent/${encodeURIComponent(id)}/control`, { method: 'POST', body });++ if (state) {++ state.paused = !!out.paused;++ state.paused_by = out.paused_by || null;++ if (state.stream && out.curtain) state.stream.curtain = out.curtain;++ if (state.stream && out.state) state.stream.state = out.state;++ }++ if (typeof out.muted === 'boolean' && voice) voice = { ...voice, muted: out.muted };++ renderControls(state || view);++ renderCurtain(!!(out.curtain && out.curtain.up));++ renderVoice();++ ensureOverlay(state || view);++ controlsLogFor = null;++ renderControls(state || view);++ } catch (err) {++ const hint = $('agent-controls-state');++ if (hint) hint.textContent = (err && err.message) || 'That control did not work.';++ }++ };++ $('agent-ctl-pause').addEventListener('click', () => pressControl({ control: 'pause', paused: !(state && state.paused) }));++ $('agent-ctl-curtain').addEventListener('click', () => {++ const up = !(state && state.stream && state.stream.curtain && state.stream.curtain.up);++ return pressControl({ control: 'curtain', up });++ });++ $('agent-ctl-mute').addEventListener('click', () => pressControl({ control: 'mute', muted: !(voice && voice.muted) }));++ // End is not a one-tap accident: the first press arms it, the second within five seconds ends it.++ let endArmed = false;++ let endTimer = null;++ $('agent-ctl-end').addEventListener('click', () => {++ const btn = $('agent-ctl-end');++ if (!endArmed) {++ endArmed = true;++ if (btn) btn.textContent = 'Press again to end';++ endTimer = setTimeout(() => { endArmed = false; if (btn) btn.textContent = 'End'; }, 5000);++ return undefined;++ }++ if (endTimer) clearTimeout(endTimer);++ endArmed = false;++ if (btn) btn.textContent = 'End';++ return pressControl({ control: 'end' });++ });++$('agent-end').addEventListener('click', () => {const confirm = $('agent-end-confirm');const end = $('agent-end');@@@ -68,6 +68,20 @@ async function load(req, res) {return row;}++/** The channel's Chat moderation (the moderators the owner set), cached by the client; [] on any failure. */++async function channelModeration(channelId) {++ try { return await require('../chat/moderation-client').getChannelModeration(Number(channelId) || 0); }++ catch { return { moderator_ids: [] }; }++}++++/** The person's role on this stream: the owner, one of its channel's moderators, or null. */++async function roleOn(row, user) {++ if (!user) return null;++ if (Number(user.id) === Number(row.user_id)) return 'owner';++ const moderation = await channelModeration(row.channel_id);++ return moderation.moderator_ids.includes(Number(user.id)) ? 'moderator' : null;++}++/*** The desk, or the person: a publish ticket for this stream (state, frame, curtain, end), else the owner's session.* A ticket names its stream, so a ticket for another stream is the same as no ticket.@@@ -131,12 +145,13 @@ router.post('/', auth.publisher, async (req, res) => {const task = cleanTask(body.task);const visibility = service.VISIBILITIES.includes(body.visibility) ? body.visibility : null;const allowSites = cleanSites(body.allow_sites);++ const copilot = body.copilot === true;if (!title) return fail(res, 400, 'invalid_title', 'A title of 1–200 characters is required');if (!task) return fail(res, 400, 'invalid_task', 'task needs source (codes, actor, desk, node), ref and title');if (!visibility) return fail(res, 400, 'invalid_visibility', 'visibility is private, unlisted or public');if (allowSites === null) return fail(res, 400, 'invalid_allow_sites', 'allow_sites is a list of at most 50 http(s) addresses');-- const row = await service.create({ user: req.publisher.user, title, task, visibility, allowSites });++ const row = await service.create({ user: req.publisher.user, title, task, visibility, allowSites, copilotReady: copilot });const publish = await tickets.issuePublish(row.managed_stream_id);const stream = service.publicView(row);if (row.unlisted_secret) stream.page_url = `${stream.page_url}?k=${encodeURIComponent(row.unlisted_secret)}`;@@@ -203,10 +218,17 @@ router.get('/:id', auth.viewer, async (req, res) => {via: isOwner ? 'owner' : (row.visibility === 'unlisted' ? 'link' : 'share'),}): null;++ const moderation = await channelModeration(row.channel_id);++ const isMod = !isOwner && !!(req.viewer && moderation.moderator_ids.includes(Number(req.viewer.id)));res.json({stream,live_stream_id: row.live_session_id ? Number(row.live_session_id) : null,can_manage: isOwner,++ can_moderate: isOwner || isMod,++ paused: !!Number(row.paused),++ paused_by: row.paused_by || null,++ paused_by_role: row.paused_by_role || null,++ copilot_ready: !!Number(row.copilot_ready),// The curtain's reason is the owner's alone; the resource itself never carries it....(isOwner && Number(row.curtain_up) && row.curtain_reason ? { curtain_reason: row.curtain_reason } : {}),...(view ? { view_ticket: view.ticket, view_ticket_expires_at: view.expires_at } : {}),@@@ -294,6 +316,10 @@ router.get('/:id/state', deskOrOwner, async (req, res) => {res.json({viewers: service.viewersNow(row.live_session_id),encode,++ mode: row.mode,++ paused: !!Number(row.paused),++ paused_by: row.paused_by || null,++ paused_by_role: row.paused_by_role || null,curtain: {up: !!Number(row.curtain_up),since: Number(row.curtain_up) ? service.iso(row.curtain_since) : null,@@@ -370,6 +396,96 @@ router.post('/:id/end', deskOrOwner, async (req, res) => {}});++// ── The four controls (S5) ────────────────────────────────────++++/**++ * POST /:id/control — pause (paused true/false), curtain (up true/false), mute (muted true/false) or end, pressed by++ * the owner or the channel's moderators. A moderator may not resume a pause the owner set, and every press is logged++ * with who pressed it. `end` ends the stream and stops the room's Co-pilot reading; the task keeps running off++ * camera.++ */++router.post('/:id/control', auth.viewer, async (req, res) => {++ try {++ const row = await load(req, res);++ if (!row) return;++ if (req.viewer && req.viewer.is_banned) return fail(res, 403, 'user_banned', 'A banned account cannot press a control');++ const role = await roleOn(row, req.viewer);++ if (!role) return fail(res, 403, 'not_moderator', 'Only the owner or a channel moderator may press this');++ const body = req.body || {};++ const control = service.CONTROLS.includes(body.control) ? body.control : null;++ if (!control) return fail(res, 400, 'invalid_control', 'control is pause, curtain, mute or end');++ const handle = String(req.viewer.username || '').toLowerCase().slice(0, 32);++ if (!/^[a-z0-9_]{2,32}$/.test(handle)) return fail(res, 409, 'no_handle', 'This account has no handle to log');++ const by = { handle, role };++ let next = row;++ let paused = null;++ if (control === 'pause') {++ const want = body.paused === undefined ? !Number(row.paused) : !!body.paused;++ const r = await service.setPaused(row, { paused: want, by });++ if (!r.allowed) return fail(res, 403, 'owner_pause', 'A moderator cannot resume a pause the owner set');++ next = r.row;++ paused = !!Number(next.paused);++ } else if (control === 'curtain') {++ const up = body.up === undefined ? !Number(row.curtain_up) : !!body.up;++ next = await service.setCurtain(row, up, 'manual');++ try { require('../streaming/curtains').apply(next.live_session_id); } catch { /* no SFU in a test */ }++ } else if (control === 'mute') {++ const muted = body.muted === undefined ? !Number(row.voice_muted) : !!body.muted;++ next = await service.update(row, { set: { voice_muted: !!muted } });++ } else if (control === 'end') {++ next = await service.end(row, { reason: `ended_by_${role}` });++ }++ await require('../db/database').run(++ `INSERT INTO agent_stream_controls (managed_stream_id, control, by_handle, by_role, paused, paused_by)++ VALUES (?, ?, ?, ?, ?, ?)`,++ [row.managed_stream_id, control, handle, role, control === 'pause' ? paused : null,++ control === 'pause' ? (paused ? handle : null) : (Number(next.paused) ? next.paused_by : null)]++ );++ res.set('Cache-Control', 'no-store');++ res.json({++ control: { control, by, at: new Date().toISOString(), stream: row.public_id, paused_by: Number(next.paused) ? next.paused_by : null },++ paused: !!Number(next.paused), paused_by: next.paused_by || null,++ curtain: { up: !!Number(next.curtain_up) },++ muted: !!Number(next.voice_muted),++ state: next.state,++ });++ } catch (err) {++ console.error('[AgentStream] control failed:', err.message);++ fail(res, 500, 'control_failed', 'The control could not be pressed');++ }++});++++/**++ * GET /:id/controls — what the owner (and a moderator) pressed, newest first. `live` is the pause in force, so the++ * page and a task's service can read it without the log.++ */++router.get('/:id/controls', auth.viewer, async (req, res) => {++ try {++ const row = await load(req, res);++ if (!row) return;++ if (req.viewer && req.viewer.is_banned) return fail(res, 403, 'user_banned', 'A banned account cannot read the controls');++ const role = await roleOn(row, req.viewer);++ if (!role) return fail(res, 403, 'not_moderator', 'Only the owner or a channel moderator may read this');++ const rows = await require('../db/database').all(++ `SELECT control, by_handle, by_role, paused, paused_by, at FROM agent_stream_controls++ WHERE managed_stream_id = ? ORDER BY id DESC LIMIT 50`, [row.managed_stream_id]);++ res.set('Cache-Control', 'no-store');++ res.json({++ controls: rows.map((r) => ({++ control: r.control,++ by: { handle: r.by_handle, role: r.by_role },++ at: service.iso(r.at),++ stream: row.public_id,++ paused_by: r.paused_by || null,++ })),++ live: { paused: !!Number(row.paused), paused_by: row.paused_by || null, paused_by_role: row.paused_by_role || null },++ });++ } catch (err) {++ console.error('[AgentStream] controls read failed:', err.message);++ fail(res, 500, 'controls_failed', 'The controls could not be read');++ }++});++// ── Steps, chapters and one-tap clips (S2) ─────────────────────// The steps the overlay draws. `after=<id>` returns only what is newer (the page's live poll); otherwise the newest 20@@@ -483,6 +599,22 @@ router.post('/:id/view-ticket', auth.viewer, async (req, res) => {}});++/**++ * POST /:id/copilot — the task's service declares whether it can read the room's digest (Codes and Actor do). The++ * owner's mode switch then allows Co-pilot on a public or unlisted stream.++ */++router.post('/:id/copilot', deskOrOwner, async (req, res) => {++ try {++ const body = req.body || {};++ if (typeof body.ready !== 'boolean') return fail(res, 400, 'invalid_ready', 'ready is true or false');++ const next = await service.setCopilotReady(req.agent, body.ready);++ res.json({ copilot_ready: !!Number(next.copilot_ready), mode: next.mode });++ } catch (err) {++ console.error('[AgentStream] copilot ready failed:', err.message);++ fail(res, 500, 'copilot_failed', 'The Co-pilot state could not be set');++ }++});++// ── The agent's voice (S4) ───────────────────────────────────// The owner chooses the mode (Watch or Narrated), the voice, the personality and the mute; a channel moderator can@@@ -517,7 +649,12 @@ router.put('/:id/voice', auth.publisher, async (req, res) => {const body = req.body || {};const set = {};if (body.mode !== undefined) {-- if (!['watch', 'narrated'].includes(body.mode)) return fail(res, 400, 'invalid_mode', 'mode is watch or narrated');++ if (!['watch', 'narrated', 'copilot'].includes(body.mode)) return fail(res, 400, 'invalid_mode', 'mode is watch, narrated or copilot');++ if (body.mode === 'copilot') {++ // Co-pilot needs a public or unlisted stream and a task's service that can read a digest.++ if (!['public', 'unlisted'].includes(row.visibility)) return fail(res, 422, 'copilot_needs_public', 'Co-pilot needs a public or unlisted stream');++ if (!Number(row.copilot_ready)) return fail(res, 422, 'copilot_not_ready', "This task's service cannot read a digest yet");++ }set.mode = body.mode;}if (body.personality !== undefined) {@@@ -26,6 +26,7 @@ const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;const AGENT_ID_RE = /^ags_[0-9A-HJKMNP-TV-Z]{26}$/;const VISIBILITIES = ['private', 'unlisted', 'public'];const MODES = ['watch', 'narrated', 'copilot', 'chatplays'];++const CONTROLS = ['pause', 'curtain', 'mute', 'end'];const TASK_SOURCES = ['codes', 'actor', 'desk', 'node'];const CURTAIN_REASONS = ['password_field', 'secret_file', 'env_print', 'manual', 'site', 'detector'];const PUBLISH_TICKET_TTL_MS = 4 * 60 * 60 * 1000;@@@ -75,6 +76,7 @@ const SELECT = `a.task, a.visibility, a.mode, a.state, a.curtain_up, a.curtain_since, a.curtain_reason,a.started_at, a.ended_at, a.worked_seconds, a.last_frame_path, a.last_frame_at, a.vod_id, a.allow_sites,a.voice, a.personality, a.voice_muted, a.voice_own, a.voice_own_consent_at, a.voice_paused,++ a.copilot_ready, a.paused, a.paused_by, a.paused_by_role, a.paused_at,a.unlisted_secret_hash,(SELECT s.id FROM streams s WHERE s.managed_stream_id = ms.id AND s.is_live = 1 ORDER BY s.started_at DESC LIMIT 1) AS live_session_id,(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,@@@ -217,7 +219,7 @@ async function listForChannel(username, viewerId, limit = 60) {** `visibility` 'unlisted' gets a link secret here; only its sha256 is stored, and the raw secret is returned once.*/--async function create({ user, title, task, visibility = 'private', allowSites = [], mode = 'watch' }) {++async function create({ user, title, task, visibility = 'private', allowSites = [], mode = 'watch', copilotReady = false }) {const subject = await ownerSubject(user.id);if (!SUBJECT_RE.test(String(subject || ''))) {const err = new Error('this account has no OpenVibe subject yet; sign in again');@@@ -235,15 +237,16 @@ async function create({ user, title, task, visibility = 'private', allowSites =[user.id, channel.id, String(title).slice(0, 200), labels, publicId]);await db.run(-- `INSERT INTO agent_streams (managed_stream_id, task, visibility, mode, state, allow_sites, unlisted_secret_hash)-- VALUES (?, ?::jsonb, ?, ?, 'idle', ?::jsonb, ?)`,-- [slot.lastInsertRowid, JSON.stringify(task), visibility, mode, JSON.stringify(allowSites || []), secret ? sha256(secret) : null]++ `INSERT INTO agent_streams (managed_stream_id, task, visibility, mode, state, allow_sites, unlisted_secret_hash, copilot_ready)++ VALUES (?, ?::jsonb, ?, ?, 'idle', ?::jsonb, ?, ?)`,++ [slot.lastInsertRowid, JSON.stringify(task), visibility, mode, JSON.stringify(allowSites || []), secret ? sha256(secret) : null, !!copilotReady]);return slot.lastInsertRowid;});const row = await getByManagedStreamId(result);if (secret) row.unlisted_secret = secret; // the one time the raw link secret exists outside the pageawait sharing.register(row);++ syncChatRoom(row);return row;}@@@ -301,7 +304,7 @@ async function end(row, { reason = 'ended' } = {}) {const worked = started ? Math.max(0, Math.round((now.getTime() - Date.parse(started)) / 1000)) : workedSeconds(row);const current = await getByManagedStreamId(row.managed_stream_id);const next = await update(current, {-- set: { state: 'ended', ended_at: now, worked_seconds: worked, curtain_up: 0, curtain_since: null, curtain_reason: null },++ set: { state: 'ended', ended_at: now, worked_seconds: worked, curtain_up: 0, curtain_since: null, curtain_reason: null, paused: 0, paused_by: null, paused_by_role: null, paused_at: null },event: { kind: 'ended', extra: { reason } },});await require('./tickets').revokeAll(row.managed_stream_id);@@@ -313,6 +316,7 @@ async function end(row, { reason = 'ended' } = {}) {} catch (err) { console.warn(`[AgentStream] ending session ${session.id} failed: ${err.message}`); }}try { require('../streaming/curtains').lower(session ? session.id : null, { notify: false }); } catch { /* */ }++ closeChatRoom(next);// Agent Streams S3: the stream part of the session report. The report the tool wrote for this task gets the// stream added to it; when no tool finished the session, a stream-only report is written instead. A report that// cannot be written never changes the stream: it ended either way.@@@ -324,6 +328,33 @@ async function end(row, { reason = 'ended' } = {}) {return next;}++/**++ * Co-pilot: the task's service says it can read the room's digest (Codes and Actor do; a desk does not). The owner++ * can then choose the mode, if the stream is public or unlisted.++ */++async function setCopilotReady(row, ready) {++ return await update(row, { set: { copilot_ready: !!ready } });++}++++/**++ * Pause or resume. A moderator may not resume a pause the owner set: `by.role` says who is asking. Returns the++ * updated row and whether the press was allowed.++ */++async function setPaused(row, { paused, by }) {++ if (!paused && Number(row.paused) && row.paused_by_role === 'owner' && by.role !== 'owner') {++ return { row, allowed: false, code: 'owner_pause' };++ }++ const next = await update(row, {++ set: {++ paused: !!paused,++ paused_by: paused ? by.handle : null,++ paused_by_role: paused ? by.role : null,++ paused_at: paused ? new Date() : null,++ },++ });++ return { row: next, allowed: true };++}++async function setVisibility(row, visibility) {if (!VISIBILITIES.includes(visibility)) return null;const secret = visibility === 'unlisted' && !row.unlisted_secret_hash ? linkSecret() : null;@@@ -332,10 +363,35 @@ async function setVisibility(row, visibility) {event: { kind: 'visibility', extra: { visibility } },});await sharing.setAudience(row, visibility);++ syncChatRoom(next);if (secret) next.unlisted_secret = secret;return next;}++/**++ * The Co-pilot room of a public or unlisted stream (Agent Streams S5, §1): Live makes it on OpenVibe.Chat when the++ * stream becomes visible. A private stream has none. Fire-and-forget: Chat being down never breaks the stream.++ */++function syncChatRoom(row) {++ if (!row || !['public', 'unlisted'].includes(row.visibility)) return;++ try {++ const chat = require('../chat/chat-client');++ if (!chat.available()) return;++ chat.agentRoom({ agent_stream_id: row.public_id, owner_subject: row.owner_subject || null, name: row.title })++ .catch(() => { /* logged by the client */ });++ } catch { /* Chat is optional for a stream */ }++}++++/** The stream ended: the room stays readable, but its Co-pilot reading is over. */++function closeChatRoom(row) {++ if (!row || !['public', 'unlisted'].includes(row.visibility)) return;++ try {++ const chat = require('../chat/chat-client');++ if (!chat.available()) return;++ chat.agentRoom({ agent_stream_id: row.public_id, close: true }).catch(() => { /* */ });++ } catch { /* */ }++}++/** A fresh unlisted link secret: the old links stop working at once. */async function rotateLink(row) {const secret = linkSecret();@@@ -407,11 +463,12 @@ function secureEqual(a, b) {}module.exports = {++ CONTROLS,AGENT_ID_RE, VISIBILITIES, MODES, TASK_SOURCES, CURTAIN_REASONS, SHARE_KIND,PUBLISH_TICKET_TTL_MS, VIEW_TICKET_TTL_MS, MAX_FRAME_BYTES,sha256, token, agentId, iso, pageUrl, whipUrl, shareRef,getByPublicId, getByManagedStreamId, liveSession, listByUser, countLiveByUser, liveLimit, ownerSubject,viewersNow, workedSeconds, publicView, cardView, listPublic, listForChannel,-- create, update, markPublishing, end, setVisibility, rotateLink, setCurtain, setFrame, setVod,-- mayView,++ create, update, markPublishing, end, setVisibility, rotateLink, setCurtain, setFrame, setVod, setCopilotReady, setPaused,++ mayView, syncChatRoom, closeChatRoom,};@@@ -57,7 +57,7 @@ function personalityOf(row) {return PERSONALITIES.includes(p) ? p : 'calm';}--function narrated(row) { return !!row && row.mode === 'narrated'; }++function narrated(row) { return !!row && (row.mode === 'narrated' || row.mode === 'copilot'); }function muted(row) { return !!row && !!Number(row.voice_muted); }/** Doxxing: an email address, a phone number, a street address or a handle. */@@@ -211,6 +211,14 @@ async function soundAsset(body) {return null;}++/**++ * Make (or close) a Co-pilot room on Chat for an agent stream: Live owns the room's creation (Agent Streams S5).++ * Chat answers with the room; a failure is logged in Chat's keys and returns null (the stream still works).++ */++async function agentRoom(body) {++ return await send('agent-room', body, '/internal/chat/agent-room');++}++module.exports = {CHAT_URL,available,@@@ -223,6 +231,6 @@ module.exports = {invalidate: async (body) => await send('invalidate', body),presence,readStats, readMessages, readTimeline, readFirstChat, readPendingIp, readRelayUsers, readRelayUser, readTtsOverride,-- readSounds, readSoundByCommand, readDmBlockState, soundAsset,++ readSounds, readSoundByCommand, readDmBlockState, soundAsset, agentRoom,_setRetryMs(ms) { retryMs = ms || RETRY_MS; },};nnew file mode 100644@@@ -0,0 +1,169 @@++'use strict';++/**++ * Agent streams S5, Live's part (design 2026-11-06): Co-pilot.++ *++ * - the mode copilot needs a public or unlisted stream and a task's service that says it can read a digest;++ * - the four controls (pause, curtain, mute, end) work for the owner and the channel's moderators;++ * - a moderator cannot resume a pause the owner set, and every press is logged;++ * - the state carries the pause the task's service reads (and the pause ends when the owner resumes);++ * - a public or unlisted stream gets its Chat room, and the end closes the reading.++ */++const assert = require('assert');++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 = (...a) => process.stderr.write(a.join(' ') + '\n');++++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, extra = {}) => await raw.prepare(++ `INSERT INTO users (id, username, display_name, email, password_hash, role, is_owner) OVERRIDING SYSTEM VALUE++ VALUES (?, ?, ?, ?, 'x', ?, ?) ON CONFLICT (id) DO NOTHING`).run(id, username, username, `${username}@x`, role, extra.is_owner ? 1 : 0);++ 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', ?) ON CONFLICT DO NOTHING").run(SID.ann);++ await raw.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (2, 'network', '102', ?) ON CONFLICT DO NOTHING").run(SID.bob);++++ const config = require('../server/config');++ config.baseUrl = 'https://openvibe.live';++ const service = require('../server/agent-streams/service');++ require('../server/agent-streams/sharing')._setNetwork({ json: async () => ({}) });++ const ann = await db.getUserById(1);++ const bob = await db.getUserById(2);++++ // Chat is another service: record what Live asks it, and make bob a moderator of ann's channel.++ const roomCalls = [];++ const chat = require('../server/chat/chat-client');++ chat.available = () => true;++ chat.agentRoom = async (body) => { roomCalls.push(body); return { room: 'agent:' + body.agent_stream_id }; };++ const moderation = require('../server/chat/moderation-client');++ moderation.getChannelModeration = async () => ({ settings: {}, moderator_ids: [2] });++++ const make = (visibility, { ready = true } = {}) => service.create({++ user: ann,++ title: 'Fixing the checkout tests ' + visibility,++ task: { source: 'codes', ref: 'tsk_1', title: 'Run the checkout tests', url: 'https://openvibe.codes/tasks/1' },++ visibility, copilotReady: ready,++ });++ const publicStream = await make('public');++ const privateStream = await make('private');++ const deskStream = await make('unlisted', { ready: false });++++ 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 } = {}) => {++ const res = await fetch(`${base}${url}`, {++ method, headers: { 'Content-Type': 'application/json', 'x-test-user': String(as) },++ 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 };++ };++ const control = (stream, body, as = 1) => call('POST', `/api/streams/agent/${stream.public_id}/control`, { as, body });++++ quiet('agent streams S5: Co-pilot');++++ await check('a public or unlisted stream with a ready service can be Co-pilot; a private one cannot', async () => {++ let r = await call('PUT', `/api/streams/agent/${publicStream.public_id}/voice`, { body: { mode: 'copilot' } });++ assert.strictEqual(r.status, 200, JSON.stringify(r.data));++ assert.strictEqual(r.data.voice.mode, 'copilot');++ r = await call('PUT', `/api/streams/agent/${privateStream.public_id}/voice`, { body: { mode: 'copilot' } });++ assert.strictEqual(r.status, 422, JSON.stringify(r.data));++ assert.strictEqual(r.data.code, 'copilot_needs_public');++ r = await call('PUT', `/api/streams/agent/${deskStream.public_id}/voice`, { body: { mode: 'copilot' } });++ assert.strictEqual(r.status, 422, JSON.stringify(r.data));++ assert.strictEqual(r.data.code, 'copilot_not_ready');++ r = await call('PUT', `/api/streams/agent/${deskStream.public_id}/voice`, { body: { mode: 'narrated' } });++ assert.strictEqual(r.status, 200, 'Narrated still works without a digest');++ r = await call('POST', `/api/streams/agent/${deskStream.public_id}/copilot`, { body: { ready: true } });++ assert.strictEqual(r.status, 200, JSON.stringify(r.data));++ r = await call('PUT', `/api/streams/agent/${deskStream.public_id}/voice`, { body: { mode: 'copilot' } });++ assert.strictEqual(r.status, 200, JSON.stringify(r.data));++ });++++ await check('the room is made for a public or unlisted stream, never a private one', async () => {++ assert.ok(roomCalls.some((c) => c.agent_stream_id === publicStream.public_id));++ assert.ok(roomCalls.some((c) => c.agent_stream_id === deskStream.public_id));++ assert.ok(!roomCalls.some((c) => c.agent_stream_id === privateStream.public_id));++ assert.strictEqual(roomCalls.find((c) => c.agent_stream_id === publicStream.public_id).owner_subject, SID.ann);++ });++++ await check('the owner pauses and resumes, and the state carries it', async () => {++ let r = await control(publicStream, { control: 'pause', paused: true });++ assert.strictEqual(r.status, 200, JSON.stringify(r.data));++ assert.strictEqual(r.data.paused, true);++ assert.strictEqual(r.data.paused_by, 'ann');++ r = await call('GET', `/api/streams/agent/${publicStream.public_id}/state`);++ assert.strictEqual(r.status, 200, JSON.stringify(r.data));++ assert.strictEqual(r.data.paused, true);++ assert.strictEqual(r.data.paused_by, 'ann');++ r = await control(publicStream, { control: 'pause', paused: false });++ assert.strictEqual(r.status, 200);++ assert.strictEqual(r.data.paused, false);++ });++++ await check("a moderator cannot resume the owner's pause, but can pause and resume their own", async () => {++ await control(publicStream, { control: 'pause', paused: true }, 1);++ let r = await control(publicStream, { control: 'pause', paused: false }, 2);++ assert.strictEqual(r.status, 403, JSON.stringify(r.data));++ assert.strictEqual(r.data.code, 'owner_pause');++ await control(publicStream, { control: 'pause', paused: false }, 1);++ r = await control(publicStream, { control: 'pause', paused: true }, 2);++ assert.strictEqual(r.status, 200, JSON.stringify(r.data));++ assert.strictEqual(r.data.paused_by, 'bob');++ r = await control(publicStream, { control: 'pause', paused: false }, 2);++ assert.strictEqual(r.status, 200, 'the moderator resumes their own pause');++ });++++ await check('a stranger may not press a control', async () => {++ const r = await control(publicStream, { control: 'pause', paused: true }, 0);++ assert.strictEqual(r.status, 403, JSON.stringify(r.data));++ });++++ await check('curtain and mute act, and every press is in the log', async () => {++ let r = await control(publicStream, { control: 'curtain', up: true }, 2);++ assert.strictEqual(r.status, 200, JSON.stringify(r.data));++ assert.strictEqual(r.data.curtain.up, true);++ r = await control(publicStream, { control: 'mute', muted: true }, 1);++ assert.strictEqual(r.status, 200, JSON.stringify(r.data));++ assert.strictEqual(r.data.muted, true);++ r = await call('GET', `/api/streams/agent/${publicStream.public_id}/controls`);++ assert.strictEqual(r.status, 200, JSON.stringify(r.data));++ const controls = r.data.controls.map((c) => c.control);++ for (const name of ['pause', 'curtain', 'mute']) assert.ok(controls.includes(name), `${name} is logged`);++ assert.ok(r.data.controls.some((c) => c.by.handle === 'bob' && c.by.role === 'moderator'));++ });++++ await check('end ends the stream and closes the Co-pilot reading', async () => {++ const r = await control(publicStream, { control: 'end' }, 2);++ assert.strictEqual(r.status, 200, JSON.stringify(r.data));++ assert.strictEqual(r.data.state, 'ended');++ assert.ok(roomCalls.some((c) => c.agent_stream_id === publicStream.public_id && c.close === true), 'the room is closed');++ const after = await control(publicStream, { control: 'pause', paused: true }, 1);++ assert.strictEqual(after.status, 200, 'an ended stream still logs the press; pause has no effect on the desk');++ });++++ const left = await db.get('SELECT COUNT(*) AS n FROM agent_stream_controls');++ assert.ok(Number(left.n) >= 8, 'every press is stored');++ server.close();++ quiet(failures ? `agent streams S5: ${failures} failed` : 'agent streams S5: Co-pilot mode, the four controls and the log');++ process.exit(failures ? 1 : 0);++})().catch((e) => { console.error(e); process.exit(1); });
@@ -163,3 +163,35 @@ counted "More (N)" after twelve.
@@ -163,3 +163,35 @@ counted "More (N)" after twelve.
**Not in S2:** the agent's own chat room (`agent:<ags_id>`), which needs a Chat-side room kind, and the voice work of
**Not in S2:** the agent's own chat room (`agent:<ags_id>`), which needs a Chat-side room kind, and the voice work of
S4. Chat is for people; the agent reading it is S5.
S4. Chat is for people; the agent reading it is S5.
+
+## S5: Co-pilot (Contracts 0.303.0)
+
+A stream can be in **Co-pilot** mode, which implies Narrated and turns on the agent's reading of its chat room. The
+owner chooses it with the voice controls (`PUT /api/streams/agent/<ags_id>/voice { "mode": "copilot" }`); it needs a
+public or unlisted stream and a task's service that says it can read a digest:
+
+```
+POST /api/streams/agent/<ags_id>/copilot { "ready": true } (the task's service, with its publish token)
+```
+
+Live makes the Chat room `agent:<ags_id>` when the stream becomes public or unlisted; a private stream has none.
+Codes and Actor read Chat's ten-item, 1,500-character digest between steps, always inside the fixed untrusted frame,
+and post an answer line, a poll or a credit ("Suggested by @mira") through Chat's `chat.agent-digest@1` and
+`chat.agent-poll@1`.
+
+**The four controls.** The owner and the channel's moderators have Pause, Curtain, Mute and End on the stream page,
+in the Dock and on the phone:
+
+```
+POST /api/streams/agent/<ags_id>/control { "control": "pause", "paused": true }
+ { "control": "curtain", "up": true }
+ { "control": "mute", "muted": true }
+ { "control": "end" }
+GET /api/streams/agent/<ags_id>/controls the log (newest first), for the owner and moderators
+```
+
+Pause is relayed to the task's service in the state it polls (`paused`, `paused_by`); it stops at the next safe point.
+A moderator may not resume a pause the owner set. The overlay says "Paused by @mod". Every press is stored in
+`agent_stream_controls` with who pressed it, which control and when; End also stops the room's Co-pilot reading while
+the task keeps running off camera. Curtain is S1's curtain, mute is S4's voice mute, and all four act within their
+design limits (2 s for pause and end, 1 s for curtain and mute).
new file mode 100644
new file mode 100644
@@ -0,0 +1,31 @@
@@ -0,0 +1,31 @@
+-- phase: expand
+-- Agent streams, S5 (design 2026-11-06, plan-state-2026-10-22/AGENT-STREAMS-S5-COPILOT-DESIGN-2026-11-06.md):
+-- Co-pilot. The stream's mode can be `copilot` (Contracts 0.303.0, live.agent-stream@1), which implies Narrated and
+-- turns on the chat digest reads and polls on OpenVibe.Chat. The owner and the channel's moderators have four
+-- controls (pause, curtain, mute, end); every press is logged with who pressed it.
+--
+-- The pause is the task's service reading it and stopping at the next safe point: `paused`, `paused_by` and
+-- `paused_by_role` travel in the state the desk polls. A moderator may not resume a pause the owner set.
+--
+-- The task's service declares it can read a digest (`copilot_ready`): Codes and Actor do, a desk does not. Co-pilot
+-- needs a public or unlisted stream, a ready service and the narrated voice.
+--
+-- ADD-only: an older release ignores the columns and the table.
+ALTER TABLE agent_streams ADD COLUMN IF NOT EXISTS copilot_ready boolean NOT NULL DEFAULT false;
+ALTER TABLE agent_streams ADD COLUMN IF NOT EXISTS paused boolean NOT NULL DEFAULT false;
+ALTER TABLE agent_streams ADD COLUMN IF NOT EXISTS paused_by text;
+ALTER TABLE agent_streams ADD COLUMN IF NOT EXISTS paused_by_role text CHECK (paused_by_role IN ('owner', 'moderator'));
+ALTER TABLE agent_streams ADD COLUMN IF NOT EXISTS paused_at timestamptz;
+
+-- Every control the owner and the channel's moderators pressed over a Co-pilot stream, and who pressed it.
+CREATE TABLE IF NOT EXISTS agent_stream_controls (
+ id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
+ managed_stream_id bigint NOT NULL REFERENCES managed_streams(id) ON DELETE CASCADE,
+ control text NOT NULL CHECK (control IN ('pause', 'curtain', 'mute', 'end')),
+ by_handle text NOT NULL,
+ by_role text NOT NULL CHECK (by_role IN ('owner', 'moderator')),
+ paused boolean, -- for a pause: true = paused, false = resumed
+ paused_by text, -- whose pause is in force after this press
+ at timestamptz NOT NULL DEFAULT now()
+);
+CREATE INDEX IF NOT EXISTS agent_stream_controls_stream ON agent_stream_controls (managed_stream_id, id DESC);
@@ -159,3 +159,15 @@
@@ -159,3 +159,15 @@
.home-agents-title { margin: 0; font-size: 1.05rem; font-weight: 800; }
.home-agents-title { margin: 0; font-size: 1.05rem; font-weight: 800; }
.home-agents-note { color: var(--text-muted); font-size: 0.78rem; }
.home-agents-note { color: var(--text-muted); font-size: 0.78rem; }
.home-agents .agent-cards { grid-template-columns: repeat(auto-fill, minmax(200px, 1fr)); }
.home-agents .agent-cards { grid-template-columns: repeat(auto-fill, minmax(200px, 1fr)); }
+
+/* S5: the moderators' bar and the paused line (Agent Streams S5). */
+.agent-controls { margin: 0 0 12px; padding: 10px 12px; border: 1px solid var(--border, rgba(255,255,255,.12)); border-radius: 12px;
+ background: var(--bg-card, rgba(18,24,38,.72)); }
+.agent-controls-title { margin: 0 0 8px; font-size: .8rem; letter-spacing: .06em; text-transform: uppercase; color: var(--text-secondary, #96a7c2); }
+.agent-controls-row { display: flex; flex-wrap: wrap; gap: 8px; }
+.agent-controls-row .btn { min-height: 44px; }
+.agent-controls-log { margin-top: 8px; }
+.agent-controls-log summary { cursor: pointer; color: var(--text-secondary, #96a7c2); font-size: .85rem; }
+.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); } }
@@ -4,6 +4,20 @@
@@ -4,6 +4,20 @@
<h1 id="agent-title" class="agent-title">Agent stream</h1>
<h1 id="agent-title" class="agent-title">Agent stream</h1>
<p id="agent-task" class="agent-task"></p>
<p id="agent-task" class="agent-task"></p>
</header>
</header>
+ <section class="agent-controls" id="agent-controls" aria-labelledby="agent-controls-title" hidden>
+ <h2 class="agent-controls-title" id="agent-controls-title">Stream controls</h2>
+ <div class="agent-controls-row">
+ <button type="button" class="btn btn-outline" id="agent-ctl-pause">Pause the agent</button>
+ <button type="button" class="btn btn-outline" id="agent-ctl-curtain">Curtain</button>
+ <button type="button" class="btn btn-outline" id="agent-ctl-mute">Mute voice</button>
+ <button type="button" class="btn btn-danger" id="agent-ctl-end">End</button>
+ </div>
+ <p class="agent-hint" id="agent-controls-state"></p>
+ <details class="agent-controls-log" id="agent-controls-log" hidden>
+ <summary>Control log</summary>
+ <ol class="agent-controls-log-list" id="agent-controls-log-list"></ol>
+ </details>
+ </section>
<div class="agent-layout">
<div class="agent-layout">
<div class="agent-main">
<div class="agent-main">
<div class="agent-stage" id="agent-stage">
<div class="agent-stage" id="agent-stage">
@@ -46,7 +60,9 @@
@@ -46,7 +60,9 @@
<select id="agent-voice-mode" class="agent-select">
<select id="agent-voice-mode" class="agent-select">
<option value="watch">Watch — the agent stays quiet</option>
<option value="watch">Watch — the agent stays quiet</option>
<option value="narrated">Narrated — the agent says what it is doing</option>
<option value="narrated">Narrated — the agent says what it is doing</option>
+ <option value="copilot">Co-pilot — it also reads the chat and may answer</option>
</select>
</select>
+ <p class="agent-note" id="agent-copilot-note" hidden>Co-pilot needs a public or unlisted stream and a task's service that can read a digest.</p>
</div>
</div>
<div class="agent-field">
<div class="agent-field">
<label class="agent-field-label" for="agent-voice-personality">How it says it</label>
<label class="agent-field-label" for="agent-voice-personality">How it says it</label>
@@ -44,7 +44,7 @@
@@ -44,7 +44,7 @@
const host = $('agent-embed-overlay');
const host = $('agent-embed-overlay');
if (!host || !window.AgentOverlay) return;
if (!host || !window.AgentOverlay) return;
if (!overlay) overlay = window.AgentOverlay.create({ host, title: stream.title, initial: steps });
if (!overlay) overlay = window.AgentOverlay.create({ host, title: stream.title, initial: steps });
- overlay.update({ steps, task: stream.task, workedSeconds: stream.worked_seconds, state: stream.state });
+ overlay.update({ steps, task: stream.task, workedSeconds: stream.worked_seconds, state: stream.state, pausedBy: streamView && streamView.__pausedBy });
}
}
function pollSteps() {
function pollSteps() {
@@ -71,6 +71,7 @@
@@ -71,6 +71,7 @@
ticket = data.view_ticket || ticket;
ticket = data.view_ticket || ticket;
streamState = stream.state;
streamState = stream.state;
streamView = stream;
streamView = stream;
+ streamView.__pausedBy = data.paused_by || null;
const title = $('agent-embed-title');
const title = $('agent-embed-title');
if (title) title.textContent = stream.title || '';
if (title) title.textContent = stream.title || '';
@@ -82,6 +82,9 @@
@@ -82,6 +82,9 @@
const dot = el('span', 'agent-overlay-dot');
const dot = el('span', 'agent-overlay-dot');
const stepTitle = el('span', 'agent-overlay-step-title');
const stepTitle = el('span', 'agent-overlay-step-title');
stepLine.append(dot, stepTitle);
stepLine.append(dot, stepTitle);
+ // S5: whose pause is in force ("Paused by @mod"), so a viewer knows why the work stopped.
+ const pausedLine = el('p', 'agent-overlay-paused');
+ pausedLine.hidden = true;
const ticks = el('div', 'agent-overlay-ticks');
const ticks = el('div', 'agent-overlay-ticks');
ticks.setAttribute('role', 'list');
ticks.setAttribute('role', 'list');
@@ -100,7 +103,7 @@
@@ -100,7 +103,7 @@
const sheetList = el('ol', 'agent-overlay-sheet-list');
const sheetList = el('ol', 'agent-overlay-sheet-list');
sheet.appendChild(sheetList);
sheet.appendChild(sheetList);
- body.append(taskLine, stepLine, captionLine, ticks, foot);
+ body.append(taskLine, stepLine, pausedLine, captionLine, ticks, foot);
root.append(foldBtn, body, sheet);
root.append(foldBtn, body, sheet);
host.appendChild(root);
host.appendChild(root);
@@ -178,6 +181,9 @@
@@ -178,6 +181,9 @@
voiceLine.textContent = '';
voiceLine.textContent = '';
voiceLine.hidden = true;
voiceLine.hidden = true;
}
}
+ const pausedBy = data.pausedBy ? String(data.pausedBy).replace(/^@/, '') : '';
+ pausedLine.textContent = pausedBy ? `Paused by @${pausedBy}` : '';
+ pausedLine.hidden = !pausedBy;
const cur = current(steps);
const cur = current(steps);
const task = data.task && data.task.title ? data.task.title : (options.title || 'An AI agent is working');
const task = data.task && data.task.title ? data.task.title : (options.title || 'An AI agent is working');
taskLine.textContent = task;
taskLine.textContent = task;
@@ -26,6 +26,7 @@
@@ -26,6 +26,7 @@
let linesTimer = null;
let linesTimer = null;
let overlay = null;
let overlay = null;
let chaptersFor = null; // the stream id whose chapters are rendered (they never change while the page lives)
let chaptersFor = null; // the stream id whose chapters are rendered (they never change while the page lives)
+ let controlsLogFor = null; // the stream id whose control log is rendered
let generation = 0; // bumped on every page leave, so a slow request from the last page changes nothing
let generation = 0; // bumped on every page leave, so a slow request from the last page changes nothing
function $(id) { return document.getElementById(id); }
function $(id) { return document.getElementById(id); }
@@ -45,6 +46,7 @@
@@ -45,6 +46,7 @@
voice = null;
voice = null;
voicesLoaded = false;
voicesLoaded = false;
chaptersFor = null;
chaptersFor = null;
+ controlsLogFor = null;
state = null;
state = null;
ticket = null;
ticket = null;
}
}
@@ -145,6 +147,7 @@
@@ -145,6 +147,7 @@
state: data.stream.state,
state: data.stream.state,
caption: captionText(),
caption: captionText(),
voice,
voice,
+ pausedBy: data.paused_by || null,
});
});
}
}
@@ -331,6 +334,7 @@
@@ -331,6 +334,7 @@
mountPlays(data);
mountPlays(data);
ensureOverlay(data);
ensureOverlay(data);
renderOwner(data);
renderOwner(data);
+ renderControls(data);
renderVoice();
renderVoice();
loadVoices();
loadVoices();
scheduleRenew(data);
scheduleRenew(data);
@@ -341,6 +345,51 @@
@@ -341,6 +345,51 @@
if (stream.state === 'ended') renderRecording(data);
if (stream.state === 'ended') renderRecording(data);
}
}
+ /** S5: the moderators' bar. The owner and the channel's moderators see it; the owner also sees the log. */
+ function renderControls(data) {
+ const bar = $('agent-controls');
+ if (!bar) return;
+ const allowed = !!(data && (data.can_manage || data.can_moderate));
+ bar.hidden = !allowed;
+ if (!allowed) return;
+ const paused = !!(data && data.paused);
+ const pauseBtn = $('agent-ctl-pause');
+ if (pauseBtn) {
+ pauseBtn.textContent = paused ? 'Resume the agent' : 'Pause the agent';
+ // A moderator cannot resume a pause the owner set (the server refuses it too).
+ pauseBtn.disabled = paused && data.paused_by_role === 'owner' && !data.can_manage;
+ }
+ const curtain = data && data.stream ? data.stream.curtain : null;
+ const curtainBtn = $('agent-ctl-curtain');
+ if (curtainBtn) curtainBtn.textContent = curtain && curtain.up ? 'Lift the curtain' : 'Curtain';
+ const muteBtn = $('agent-ctl-mute');
+ if (muteBtn) muteBtn.textContent = voice && voice.muted ? 'Unmute voice' : 'Mute voice';
+ const stateEl = $('agent-controls-state');
+ if (stateEl) {
+ if (paused && data.paused_by) stateEl.textContent = `Paused by @${data.paused_by}`;
+ else if (curtain && curtain.up) stateEl.textContent = 'The curtain is up: viewers see a private screen.';
+ else if (voice && voice.muted) stateEl.textContent = 'The voice is muted.';
+ else stateEl.textContent = '';
+ }
+ const log = $('agent-controls-log');
+ if (log && data.can_manage && data.stream && controlsLogFor !== data.stream.id) {
+ controlsLogFor = data.stream.id;
+ log.hidden = false;
+ const list = $('agent-controls-log-list');
+ if (list) list.replaceChildren();
+ api(`/streams/agent/${encodeURIComponent(data.stream.id)}/controls`).then((out) => {
+ if (!out || !Array.isArray(out.controls) || !list) return;
+ if (!out.controls.length) { list.appendChild(Object.assign(document.createElement('li'), { textContent: 'No control has been pressed.' })); return; }
+ for (const entry of out.controls) {
+ const li = document.createElement('li');
+ const when = new Date(entry.at);
+ li.textContent = `${entry.control} by @${entry.by.handle} (${entry.by.role}) · ${when.toLocaleString()}`;
+ list.appendChild(li);
+ }
+ }).catch(() => { controlsLogFor = null; });
+ } else if (log && !data.can_manage) log.hidden = true;
+ }
+
function renderOwner(data) {
function renderOwner(data) {
const panel = $('agent-owner');
const panel = $('agent-owner');
if (!panel) return;
if (!panel) return;
@@ -385,7 +434,7 @@
@@ -385,7 +434,7 @@
const panel = $('agent-owner');
const panel = $('agent-owner');
if (!panel || panel.hidden || !voice) return;
if (!panel || panel.hidden || !voice) return;
const mode = $('agent-voice-mode');
const mode = $('agent-voice-mode');
- if (mode) mode.value = voice.mode === 'narrated' ? 'narrated' : 'watch';
+ if (mode) mode.value = ['narrated', 'copilot'].includes(voice.mode) ? voice.mode : 'watch';
const personality = $('agent-voice-personality');
const personality = $('agent-voice-personality');
if (personality) personality.value = voice.personality || 'calm';
if (personality) personality.value = voice.personality || 'calm';
const own = $('agent-voice-own');
const own = $('agent-voice-own');
@@ -404,6 +453,8 @@
@@ -404,6 +453,8 @@
state.textContent = 'voice paused: allowance spent';
state.textContent = 'voice paused: allowance spent';
} else if (voice.muted) {
} else if (voice.muted) {
state.textContent = 'The voice is muted.';
state.textContent = 'The voice is muted.';
+ } else if (voice.mode === 'copilot') {
+ state.textContent = `Co-pilot, in a ${voice.personality || 'calm'} voice: it reads the chat between steps.`;
} else if (voice.mode === 'narrated') {
} else if (voice.mode === 'narrated') {
state.textContent = `Narrated, in a ${voice.personality || 'calm'} voice.`;
state.textContent = `Narrated, in a ${voice.personality || 'calm'} voice.`;
} else {
} else {
@@ -474,6 +525,7 @@
@@ -474,6 +525,7 @@
ensureOverlay(next);
ensureOverlay(next);
renderStatus(next.stream, liveViewers);
renderStatus(next.stream, liveViewers);
renderOwner(next);
renderOwner(next);
+ renderControls(next);
schedulePoll(id);
schedulePoll(id);
} catch { if (gen === generation) schedulePoll(id); }
} catch { if (gen === generation) schedulePoll(id); }
}, 20000);
}, 20000);
@@ -554,6 +606,54 @@
@@ -554,6 +606,54 @@
} catch { /* the desk still owns the curtain */ }
} catch { /* the desk still owns the curtain */ }
});
});
+ // S5: the moderators' bar, four controls, every press logged by the server.
+ const pressControl = async (body) => {
+ const id = idOf();
+ if (!id) return;
+ const view = state || {};
+ try {
+ const out = await api(`/streams/agent/${encodeURIComponent(id)}/control`, { method: 'POST', body });
+ if (state) {
+ state.paused = !!out.paused;
+ state.paused_by = out.paused_by || null;
+ if (state.stream && out.curtain) state.stream.curtain = out.curtain;
+ if (state.stream && out.state) state.stream.state = out.state;
+ }
+ if (typeof out.muted === 'boolean' && voice) voice = { ...voice, muted: out.muted };
+ renderControls(state || view);
+ renderCurtain(!!(out.curtain && out.curtain.up));
+ renderVoice();
+ ensureOverlay(state || view);
+ controlsLogFor = null;
+ renderControls(state || view);
+ } catch (err) {
+ const hint = $('agent-controls-state');
+ if (hint) hint.textContent = (err && err.message) || 'That control did not work.';
+ }
+ };
+ $('agent-ctl-pause').addEventListener('click', () => pressControl({ control: 'pause', paused: !(state && state.paused) }));
+ $('agent-ctl-curtain').addEventListener('click', () => {
+ const up = !(state && state.stream && state.stream.curtain && state.stream.curtain.up);
+ return pressControl({ control: 'curtain', up });
+ });
+ $('agent-ctl-mute').addEventListener('click', () => pressControl({ control: 'mute', muted: !(voice && voice.muted) }));
+ // End is not a one-tap accident: the first press arms it, the second within five seconds ends it.
+ let endArmed = false;
+ let endTimer = null;
+ $('agent-ctl-end').addEventListener('click', () => {
+ const btn = $('agent-ctl-end');
+ if (!endArmed) {
+ endArmed = true;
+ if (btn) btn.textContent = 'Press again to end';
+ endTimer = setTimeout(() => { endArmed = false; if (btn) btn.textContent = 'End'; }, 5000);
+ return undefined;
+ }
+ if (endTimer) clearTimeout(endTimer);
+ endArmed = false;
+ if (btn) btn.textContent = 'End';
+ return pressControl({ control: 'end' });
+ });
+
$('agent-end').addEventListener('click', () => {
$('agent-end').addEventListener('click', () => {
const confirm = $('agent-end-confirm');
const confirm = $('agent-end-confirm');
const end = $('agent-end');
const end = $('agent-end');
@@ -68,6 +68,20 @@ async function load(req, res) {
@@ -68,6 +68,20 @@ async function load(req, res) {
return row;
return row;
}
}
+/** The channel's Chat moderation (the moderators the owner set), cached by the client; [] on any failure. */
+async function channelModeration(channelId) {
+ try { return await require('../chat/moderation-client').getChannelModeration(Number(channelId) || 0); }
+ catch { return { moderator_ids: [] }; }
+}
+
+/** The person's role on this stream: the owner, one of its channel's moderators, or null. */
+async function roleOn(row, user) {
+ if (!user) return null;
+ if (Number(user.id) === Number(row.user_id)) return 'owner';
+ const moderation = await channelModeration(row.channel_id);
+ return moderation.moderator_ids.includes(Number(user.id)) ? 'moderator' : null;
+}
+
/**
/**
* The desk, or the person: a publish ticket for this stream (state, frame, curtain, end), else the owner's session.
* The desk, or the person: a publish ticket for this stream (state, frame, curtain, end), else the owner's session.
* A ticket names its stream, so a ticket for another stream is the same as no ticket.
* A ticket names its stream, so a ticket for another stream is the same as no ticket.
@@ -131,12 +145,13 @@ router.post('/', auth.publisher, async (req, res) => {
@@ -131,12 +145,13 @@ router.post('/', auth.publisher, async (req, res) => {
const task = cleanTask(body.task);
const task = cleanTask(body.task);
const visibility = service.VISIBILITIES.includes(body.visibility) ? body.visibility : null;
const visibility = service.VISIBILITIES.includes(body.visibility) ? body.visibility : null;
const allowSites = cleanSites(body.allow_sites);
const allowSites = cleanSites(body.allow_sites);
+ const copilot = body.copilot === true;
if (!title) return fail(res, 400, 'invalid_title', 'A title of 1–200 characters is required');
if (!title) return fail(res, 400, 'invalid_title', 'A title of 1–200 characters is required');
if (!task) return fail(res, 400, 'invalid_task', 'task needs source (codes, actor, desk, node), ref and title');
if (!task) return fail(res, 400, 'invalid_task', 'task needs source (codes, actor, desk, node), ref and title');
if (!visibility) return fail(res, 400, 'invalid_visibility', 'visibility is private, unlisted or public');
if (!visibility) return fail(res, 400, 'invalid_visibility', 'visibility is private, unlisted or public');
if (allowSites === null) return fail(res, 400, 'invalid_allow_sites', 'allow_sites is a list of at most 50 http(s) addresses');
if (allowSites === null) return fail(res, 400, 'invalid_allow_sites', 'allow_sites is a list of at most 50 http(s) addresses');
- const row = await service.create({ user: req.publisher.user, title, task, visibility, allowSites });
+ const row = await service.create({ user: req.publisher.user, title, task, visibility, allowSites, copilotReady: copilot });
const publish = await tickets.issuePublish(row.managed_stream_id);
const publish = await tickets.issuePublish(row.managed_stream_id);
const stream = service.publicView(row);
const stream = service.publicView(row);
if (row.unlisted_secret) stream.page_url = `${stream.page_url}?k=${encodeURIComponent(row.unlisted_secret)}`;
if (row.unlisted_secret) stream.page_url = `${stream.page_url}?k=${encodeURIComponent(row.unlisted_secret)}`;
@@ -203,10 +218,17 @@ router.get('/:id', auth.viewer, async (req, res) => {
@@ -203,10 +218,17 @@ router.get('/:id', auth.viewer, async (req, res) => {
via: isOwner ? 'owner' : (row.visibility === 'unlisted' ? 'link' : 'share'),
via: isOwner ? 'owner' : (row.visibility === 'unlisted' ? 'link' : 'share'),
})
})
: null;
: null;
+ const moderation = await channelModeration(row.channel_id);
+ const isMod = !isOwner && !!(req.viewer && moderation.moderator_ids.includes(Number(req.viewer.id)));
res.json({
res.json({
stream,
stream,
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,
+ paused: !!Number(row.paused),
+ paused_by: row.paused_by || null,
+ paused_by_role: row.paused_by_role || null,
+ copilot_ready: !!Number(row.copilot_ready),
// The curtain's reason is the owner's alone; the resource itself never carries it.
// The curtain's reason is the owner's alone; the resource itself never carries it.
...(isOwner && Number(row.curtain_up) && row.curtain_reason ? { curtain_reason: row.curtain_reason } : {}),
...(isOwner && Number(row.curtain_up) && row.curtain_reason ? { curtain_reason: row.curtain_reason } : {}),
...(view ? { view_ticket: view.ticket, view_ticket_expires_at: view.expires_at } : {}),
...(view ? { view_ticket: view.ticket, view_ticket_expires_at: view.expires_at } : {}),
@@ -294,6 +316,10 @@ router.get('/:id/state', deskOrOwner, async (req, res) => {
@@ -294,6 +316,10 @@ router.get('/:id/state', deskOrOwner, async (req, res) => {
res.json({
res.json({
viewers: service.viewersNow(row.live_session_id),
viewers: service.viewersNow(row.live_session_id),
encode,
encode,
+ mode: row.mode,
+ paused: !!Number(row.paused),
+ paused_by: row.paused_by || null,
+ paused_by_role: row.paused_by_role || null,
curtain: {
curtain: {
up: !!Number(row.curtain_up),
up: !!Number(row.curtain_up),
since: Number(row.curtain_up) ? service.iso(row.curtain_since) : null,
since: Number(row.curtain_up) ? service.iso(row.curtain_since) : null,
@@ -370,6 +396,96 @@ router.post('/:id/end', deskOrOwner, async (req, res) => {
@@ -370,6 +396,96 @@ router.post('/:id/end', deskOrOwner, async (req, res) => {
}
}
});
});
+// ── The four controls (S5) ────────────────────────────────────
+
+/**
+ * POST /:id/control — pause (paused true/false), curtain (up true/false), mute (muted true/false) or end, pressed by
+ * the owner or the channel's moderators. A moderator may not resume a pause the owner set, and every press is logged
+ * with who pressed it. `end` ends the stream and stops the room's Co-pilot reading; the task keeps running off
+ * camera.
+ */
+router.post('/:id/control', auth.viewer, async (req, res) => {
+ try {
+ const row = await load(req, res);
+ if (!row) return;
+ if (req.viewer && req.viewer.is_banned) return fail(res, 403, 'user_banned', 'A banned account cannot press a control');
+ const role = await roleOn(row, req.viewer);
+ if (!role) return fail(res, 403, 'not_moderator', 'Only the owner or a channel moderator may press this');
+ const body = req.body || {};
+ const control = service.CONTROLS.includes(body.control) ? body.control : null;
+ if (!control) return fail(res, 400, 'invalid_control', 'control is pause, curtain, mute or end');
+ const handle = String(req.viewer.username || '').toLowerCase().slice(0, 32);
+ if (!/^[a-z0-9_]{2,32}$/.test(handle)) return fail(res, 409, 'no_handle', 'This account has no handle to log');
+ const by = { handle, role };
+ let next = row;
+ let paused = null;
+ if (control === 'pause') {
+ const want = body.paused === undefined ? !Number(row.paused) : !!body.paused;
+ const r = await service.setPaused(row, { paused: want, by });
+ if (!r.allowed) return fail(res, 403, 'owner_pause', 'A moderator cannot resume a pause the owner set');
+ next = r.row;
+ paused = !!Number(next.paused);
+ } else if (control === 'curtain') {
+ const up = body.up === undefined ? !Number(row.curtain_up) : !!body.up;
+ next = await service.setCurtain(row, up, 'manual');
+ try { require('../streaming/curtains').apply(next.live_session_id); } catch { /* no SFU in a test */ }
+ } else if (control === 'mute') {
+ const muted = body.muted === undefined ? !Number(row.voice_muted) : !!body.muted;
+ next = await service.update(row, { set: { voice_muted: !!muted } });
+ } else if (control === 'end') {
+ next = await service.end(row, { reason: `ended_by_${role}` });
+ }
+ await require('../db/database').run(
+ `INSERT INTO agent_stream_controls (managed_stream_id, control, by_handle, by_role, paused, paused_by)
+ VALUES (?, ?, ?, ?, ?, ?)`,
+ [row.managed_stream_id, control, handle, role, control === 'pause' ? paused : null,
+ control === 'pause' ? (paused ? handle : null) : (Number(next.paused) ? next.paused_by : null)]
+ );
+ res.set('Cache-Control', 'no-store');
+ res.json({
+ control: { control, by, at: new Date().toISOString(), stream: row.public_id, paused_by: Number(next.paused) ? next.paused_by : null },
+ paused: !!Number(next.paused), paused_by: next.paused_by || null,
+ curtain: { up: !!Number(next.curtain_up) },
+ muted: !!Number(next.voice_muted),
+ state: next.state,
+ });
+ } catch (err) {
+ console.error('[AgentStream] control failed:', err.message);
+ fail(res, 500, 'control_failed', 'The control could not be pressed');
+ }
+});
+
+/**
+ * GET /:id/controls — what the owner (and a moderator) pressed, newest first. `live` is the pause in force, so the
+ * page and a task's service can read it without the log.
+ */
+router.get('/:id/controls', auth.viewer, async (req, res) => {
+ try {
+ const row = await load(req, res);
+ if (!row) return;
+ if (req.viewer && req.viewer.is_banned) return fail(res, 403, 'user_banned', 'A banned account cannot read the controls');
+ const role = await roleOn(row, req.viewer);
+ if (!role) return fail(res, 403, 'not_moderator', 'Only the owner or a channel moderator may read this');
+ const rows = await require('../db/database').all(
+ `SELECT control, by_handle, by_role, paused, paused_by, at FROM agent_stream_controls
+ WHERE managed_stream_id = ? ORDER BY id DESC LIMIT 50`, [row.managed_stream_id]);
+ res.set('Cache-Control', 'no-store');
+ res.json({
+ controls: rows.map((r) => ({
+ control: r.control,
+ by: { handle: r.by_handle, role: r.by_role },
+ at: service.iso(r.at),
+ stream: row.public_id,
+ paused_by: r.paused_by || null,
+ })),
+ live: { paused: !!Number(row.paused), paused_by: row.paused_by || null, paused_by_role: row.paused_by_role || null },
+ });
+ } catch (err) {
+ console.error('[AgentStream] controls read failed:', err.message);
+ fail(res, 500, 'controls_failed', 'The controls could not be read');
+ }
+});
+
// ── Steps, chapters and one-tap clips (S2) ─────────────────────
// ── Steps, chapters and one-tap clips (S2) ─────────────────────
// The steps the overlay draws. `after=<id>` returns only what is newer (the page's live poll); otherwise the newest 20
// The steps the overlay draws. `after=<id>` returns only what is newer (the page's live poll); otherwise the newest 20
@@ -483,6 +599,22 @@ router.post('/:id/view-ticket', auth.viewer, async (req, res) => {
@@ -483,6 +599,22 @@ router.post('/:id/view-ticket', auth.viewer, async (req, res) => {
}
}
});
});
+/**
+ * POST /:id/copilot — the task's service declares whether it can read the room's digest (Codes and Actor do). The
+ * owner's mode switch then allows Co-pilot on a public or unlisted stream.
+ */
+router.post('/:id/copilot', deskOrOwner, async (req, res) => {
+ try {
+ const body = req.body || {};
+ if (typeof body.ready !== 'boolean') return fail(res, 400, 'invalid_ready', 'ready is true or false');
+ const next = await service.setCopilotReady(req.agent, body.ready);
+ res.json({ copilot_ready: !!Number(next.copilot_ready), mode: next.mode });
+ } catch (err) {
+ console.error('[AgentStream] copilot ready failed:', err.message);
+ fail(res, 500, 'copilot_failed', 'The Co-pilot state could not be set');
+ }
+});
+
// ── The agent's voice (S4) ───────────────────────────────────
// ── The agent's voice (S4) ───────────────────────────────────
// The owner chooses the mode (Watch or Narrated), the voice, the personality and the mute; a channel moderator can
// The owner chooses the mode (Watch or Narrated), the voice, the personality and the mute; a channel moderator can
@@ -517,7 +649,12 @@ router.put('/:id/voice', auth.publisher, async (req, res) => {
@@ -517,7 +649,12 @@ router.put('/:id/voice', auth.publisher, async (req, res) => {
const body = req.body || {};
const body = req.body || {};
const set = {};
const set = {};
if (body.mode !== undefined) {
if (body.mode !== undefined) {
- if (!['watch', 'narrated'].includes(body.mode)) return fail(res, 400, 'invalid_mode', 'mode is watch or narrated');
+ if (!['watch', 'narrated', 'copilot'].includes(body.mode)) return fail(res, 400, 'invalid_mode', 'mode is watch, narrated or copilot');
+ if (body.mode === 'copilot') {
+ // Co-pilot needs a public or unlisted stream and a task's service that can read a digest.
+ if (!['public', 'unlisted'].includes(row.visibility)) return fail(res, 422, 'copilot_needs_public', 'Co-pilot needs a public or unlisted stream');
+ if (!Number(row.copilot_ready)) return fail(res, 422, 'copilot_not_ready', "This task's service cannot read a digest yet");
+ }
set.mode = body.mode;
set.mode = body.mode;
}
}
if (body.personality !== undefined) {
if (body.personality !== undefined) {
@@ -26,6 +26,7 @@ const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
@@ -26,6 +26,7 @@ const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
const AGENT_ID_RE = /^ags_[0-9A-HJKMNP-TV-Z]{26}$/;
const AGENT_ID_RE = /^ags_[0-9A-HJKMNP-TV-Z]{26}$/;
const VISIBILITIES = ['private', 'unlisted', 'public'];
const VISIBILITIES = ['private', 'unlisted', 'public'];
const MODES = ['watch', 'narrated', 'copilot', 'chatplays'];
const MODES = ['watch', 'narrated', 'copilot', 'chatplays'];
+const CONTROLS = ['pause', 'curtain', 'mute', 'end'];
const TASK_SOURCES = ['codes', 'actor', 'desk', 'node'];
const TASK_SOURCES = ['codes', 'actor', 'desk', 'node'];
const CURTAIN_REASONS = ['password_field', 'secret_file', 'env_print', 'manual', 'site', 'detector'];
const CURTAIN_REASONS = ['password_field', 'secret_file', 'env_print', 'manual', 'site', 'detector'];
const PUBLISH_TICKET_TTL_MS = 4 * 60 * 60 * 1000;
const PUBLISH_TICKET_TTL_MS = 4 * 60 * 60 * 1000;
@@ -75,6 +76,7 @@ const SELECT = `
@@ -75,6 +76,7 @@ const SELECT = `
a.task, a.visibility, a.mode, a.state, a.curtain_up, a.curtain_since, a.curtain_reason,
a.task, a.visibility, a.mode, a.state, a.curtain_up, a.curtain_since, a.curtain_reason,
a.started_at, a.ended_at, a.worked_seconds, a.last_frame_path, a.last_frame_at, a.vod_id, a.allow_sites,
a.started_at, a.ended_at, a.worked_seconds, a.last_frame_path, a.last_frame_at, a.vod_id, a.allow_sites,
a.voice, a.personality, a.voice_muted, a.voice_own, a.voice_own_consent_at, a.voice_paused,
a.voice, a.personality, a.voice_muted, a.voice_own, a.voice_own_consent_at, a.voice_paused,
+ a.copilot_ready, a.paused, a.paused_by, a.paused_by_role, a.paused_at,
a.unlisted_secret_hash,
a.unlisted_secret_hash,
(SELECT s.id FROM streams s WHERE s.managed_stream_id = ms.id AND s.is_live = 1 ORDER BY s.started_at DESC LIMIT 1) AS live_session_id,
(SELECT s.id FROM streams s WHERE s.managed_stream_id = ms.id AND s.is_live = 1 ORDER BY s.started_at DESC LIMIT 1) AS live_session_id,
(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,
@@ -217,7 +219,7 @@ async function listForChannel(username, viewerId, limit = 60) {
@@ -217,7 +219,7 @@ async function listForChannel(username, viewerId, limit = 60) {
*
*
* `visibility` 'unlisted' gets a link secret here; only its sha256 is stored, and the raw secret is returned once.
* `visibility` 'unlisted' gets a link secret here; only its sha256 is stored, and the raw secret is returned once.
*/
*/
-async function create({ user, title, task, visibility = 'private', allowSites = [], mode = 'watch' }) {
+async function create({ user, title, task, visibility = 'private', allowSites = [], mode = 'watch', copilotReady = false }) {
const subject = await ownerSubject(user.id);
const subject = await ownerSubject(user.id);
if (!SUBJECT_RE.test(String(subject || ''))) {
if (!SUBJECT_RE.test(String(subject || ''))) {
const err = new Error('this account has no OpenVibe subject yet; sign in again');
const err = new Error('this account has no OpenVibe subject yet; sign in again');
@@ -235,15 +237,16 @@ async function create({ user, title, task, visibility = 'private', allowSites =
@@ -235,15 +237,16 @@ async function create({ user, title, task, visibility = 'private', allowSites =
[user.id, channel.id, String(title).slice(0, 200), labels, publicId]
[user.id, channel.id, String(title).slice(0, 200), labels, publicId]
);
);
await db.run(
await db.run(
- `INSERT INTO agent_streams (managed_stream_id, task, visibility, mode, state, allow_sites, unlisted_secret_hash)
- VALUES (?, ?::jsonb, ?, ?, 'idle', ?::jsonb, ?)`,
- [slot.lastInsertRowid, JSON.stringify(task), visibility, mode, JSON.stringify(allowSites || []), secret ? sha256(secret) : null]
+ `INSERT INTO agent_streams (managed_stream_id, task, visibility, mode, state, allow_sites, unlisted_secret_hash, copilot_ready)
+ VALUES (?, ?::jsonb, ?, ?, 'idle', ?::jsonb, ?, ?)`,
+ [slot.lastInsertRowid, JSON.stringify(task), visibility, mode, JSON.stringify(allowSites || []), secret ? sha256(secret) : null, !!copilotReady]
);
);
return slot.lastInsertRowid;
return slot.lastInsertRowid;
});
});
const row = await getByManagedStreamId(result);
const row = await getByManagedStreamId(result);
if (secret) row.unlisted_secret = secret; // the one time the raw link secret exists outside the page
if (secret) row.unlisted_secret = secret; // the one time the raw link secret exists outside the page
await sharing.register(row);
await sharing.register(row);
+ syncChatRoom(row);
return row;
return row;
}
}
@@ -301,7 +304,7 @@ async function end(row, { reason = 'ended' } = {}) {
@@ -301,7 +304,7 @@ async function end(row, { reason = 'ended' } = {}) {
const worked = started ? Math.max(0, Math.round((now.getTime() - Date.parse(started)) / 1000)) : workedSeconds(row);
const worked = started ? Math.max(0, Math.round((now.getTime() - Date.parse(started)) / 1000)) : workedSeconds(row);
const current = await getByManagedStreamId(row.managed_stream_id);
const current = await getByManagedStreamId(row.managed_stream_id);
const next = await update(current, {
const next = await update(current, {
- set: { state: 'ended', ended_at: now, worked_seconds: worked, curtain_up: 0, curtain_since: null, curtain_reason: null },
+ set: { state: 'ended', ended_at: now, worked_seconds: worked, curtain_up: 0, curtain_since: null, curtain_reason: null, paused: 0, paused_by: null, paused_by_role: null, paused_at: null },
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);
@@ -313,6 +316,7 @@ async function end(row, { reason = 'ended' } = {}) {
@@ -313,6 +316,7 @@ async function end(row, { reason = 'ended' } = {}) {
} catch (err) { console.warn(`[AgentStream] ending session ${session.id} failed: ${err.message}`); }
} catch (err) { console.warn(`[AgentStream] ending session ${session.id} failed: ${err.message}`); }
}
}
try { require('../streaming/curtains').lower(session ? session.id : null, { notify: false }); } catch { /* */ }
try { require('../streaming/curtains').lower(session ? session.id : null, { notify: false }); } catch { /* */ }
+ closeChatRoom(next);
// Agent Streams S3: the stream part of the session report. The report the tool wrote for this task gets the
// Agent Streams S3: the stream part of the session report. The report the tool wrote for this task gets the
// stream added to it; when no tool finished the session, a stream-only report is written instead. A report that
// stream added to it; when no tool finished the session, a stream-only report is written instead. A report that
// cannot be written never changes the stream: it ended either way.
// cannot be written never changes the stream: it ended either way.
@@ -324,6 +328,33 @@ async function end(row, { reason = 'ended' } = {}) {
@@ -324,6 +328,33 @@ async function end(row, { reason = 'ended' } = {}) {
return next;
return next;
}
}
+/**
+ * Co-pilot: the task's service says it can read the room's digest (Codes and Actor do; a desk does not). The owner
+ * can then choose the mode, if the stream is public or unlisted.
+ */
+async function setCopilotReady(row, ready) {
+ return await update(row, { set: { copilot_ready: !!ready } });
+}
+
+/**
+ * Pause or resume. A moderator may not resume a pause the owner set: `by.role` says who is asking. Returns the
+ * updated row and whether the press was allowed.
+ */
+async function setPaused(row, { paused, by }) {
+ if (!paused && Number(row.paused) && row.paused_by_role === 'owner' && by.role !== 'owner') {
+ return { row, allowed: false, code: 'owner_pause' };
+ }
+ const next = await update(row, {
+ set: {
+ paused: !!paused,
+ paused_by: paused ? by.handle : null,
+ paused_by_role: paused ? by.role : null,
+ paused_at: paused ? new Date() : null,
+ },
+ });
+ return { row: next, allowed: true };
+}
+
async function setVisibility(row, visibility) {
async function setVisibility(row, visibility) {
if (!VISIBILITIES.includes(visibility)) return null;
if (!VISIBILITIES.includes(visibility)) return null;
const secret = visibility === 'unlisted' && !row.unlisted_secret_hash ? linkSecret() : null;
const secret = visibility === 'unlisted' && !row.unlisted_secret_hash ? linkSecret() : null;
@@ -332,10 +363,35 @@ async function setVisibility(row, visibility) {
@@ -332,10 +363,35 @@ async function setVisibility(row, visibility) {
event: { kind: 'visibility', extra: { visibility } },
event: { kind: 'visibility', extra: { visibility } },
});
});
await sharing.setAudience(row, visibility);
await sharing.setAudience(row, visibility);
+ syncChatRoom(next);
if (secret) next.unlisted_secret = secret;
if (secret) next.unlisted_secret = secret;
return next;
return next;
}
}
+/**
+ * The Co-pilot room of a public or unlisted stream (Agent Streams S5, §1): Live makes it on OpenVibe.Chat when the
+ * stream becomes visible. A private stream has none. Fire-and-forget: Chat being down never breaks the stream.
+ */
+function syncChatRoom(row) {
+ if (!row || !['public', 'unlisted'].includes(row.visibility)) return;
+ try {
+ const chat = require('../chat/chat-client');
+ if (!chat.available()) return;
+ chat.agentRoom({ agent_stream_id: row.public_id, owner_subject: row.owner_subject || null, name: row.title })
+ .catch(() => { /* logged by the client */ });
+ } catch { /* Chat is optional for a stream */ }
+}
+
+/** The stream ended: the room stays readable, but its Co-pilot reading is over. */
+function closeChatRoom(row) {
+ if (!row || !['public', 'unlisted'].includes(row.visibility)) return;
+ try {
+ const chat = require('../chat/chat-client');
+ if (!chat.available()) return;
+ chat.agentRoom({ agent_stream_id: row.public_id, close: true }).catch(() => { /* */ });
+ } catch { /* */ }
+}
+
/** A fresh unlisted link secret: the old links stop working at once. */
/** A fresh unlisted link secret: the old links stop working at once. */
async function rotateLink(row) {
async function rotateLink(row) {
const secret = linkSecret();
const secret = linkSecret();
@@ -407,11 +463,12 @@ function secureEqual(a, b) {
@@ -407,11 +463,12 @@ function secureEqual(a, b) {
}
}
module.exports = {
module.exports = {
+ CONTROLS,
AGENT_ID_RE, VISIBILITIES, MODES, TASK_SOURCES, CURTAIN_REASONS, SHARE_KIND,
AGENT_ID_RE, VISIBILITIES, MODES, TASK_SOURCES, CURTAIN_REASONS, SHARE_KIND,
PUBLISH_TICKET_TTL_MS, VIEW_TICKET_TTL_MS, MAX_FRAME_BYTES,
PUBLISH_TICKET_TTL_MS, VIEW_TICKET_TTL_MS, MAX_FRAME_BYTES,
sha256, token, agentId, iso, pageUrl, whipUrl, shareRef,
sha256, token, agentId, iso, pageUrl, whipUrl, shareRef,
getByPublicId, getByManagedStreamId, liveSession, listByUser, countLiveByUser, liveLimit, ownerSubject,
getByPublicId, getByManagedStreamId, liveSession, listByUser, countLiveByUser, liveLimit, ownerSubject,
viewersNow, workedSeconds, publicView, cardView, listPublic, listForChannel,
viewersNow, workedSeconds, publicView, cardView, listPublic, listForChannel,
- create, update, markPublishing, end, setVisibility, rotateLink, setCurtain, setFrame, setVod,
- mayView,
+ create, update, markPublishing, end, setVisibility, rotateLink, setCurtain, setFrame, setVod, setCopilotReady, setPaused,
+ mayView, syncChatRoom, closeChatRoom,
};
};
@@ -57,7 +57,7 @@ function personalityOf(row) {
@@ -57,7 +57,7 @@ function personalityOf(row) {
return PERSONALITIES.includes(p) ? p : 'calm';
return PERSONALITIES.includes(p) ? p : 'calm';
}
}
-function narrated(row) { return !!row && row.mode === 'narrated'; }
+function narrated(row) { return !!row && (row.mode === 'narrated' || row.mode === 'copilot'); }
function muted(row) { return !!row && !!Number(row.voice_muted); }
function muted(row) { return !!row && !!Number(row.voice_muted); }
/** Doxxing: an email address, a phone number, a street address or a handle. */
/** Doxxing: an email address, a phone number, a street address or a handle. */
@@ -211,6 +211,14 @@ async function soundAsset(body) {
@@ -211,6 +211,14 @@ async function soundAsset(body) {
return null;
return null;
}
}
+/**
+ * Make (or close) a Co-pilot room on Chat for an agent stream: Live owns the room's creation (Agent Streams S5).
+ * Chat answers with the room; a failure is logged in Chat's keys and returns null (the stream still works).
+ */
+async function agentRoom(body) {
+ return await send('agent-room', body, '/internal/chat/agent-room');
+}
+
module.exports = {
module.exports = {
CHAT_URL,
CHAT_URL,
available,
available,
@@ -223,6 +231,6 @@ module.exports = {
@@ -223,6 +231,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,
+ readSounds, readSoundByCommand, readDmBlockState, soundAsset, agentRoom,
_setRetryMs(ms) { retryMs = ms || RETRY_MS; },
_setRetryMs(ms) { retryMs = ms || RETRY_MS; },
};
};
new file mode 100644
new file mode 100644
@@ -0,0 +1,169 @@
@@ -0,0 +1,169 @@
+'use strict';
+/**
+ * Agent streams S5, Live's part (design 2026-11-06): Co-pilot.
+ *
+ * - the mode copilot needs a public or unlisted stream and a task's service that says it can read a digest;
+ * - the four controls (pause, curtain, mute, end) work for the owner and the channel's moderators;
+ * - a moderator cannot resume a pause the owner set, and every press is logged;
+ * - the state carries the pause the task's service reads (and the pause ends when the owner resumes);
+ * - a public or unlisted stream gets its Chat room, and the end closes the reading.
+ */
+const assert = require('assert');
+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 = (...a) => process.stderr.write(a.join(' ') + '\n');
+
+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, extra = {}) => await raw.prepare(
+ `INSERT INTO users (id, username, display_name, email, password_hash, role, is_owner) OVERRIDING SYSTEM VALUE
+ VALUES (?, ?, ?, ?, 'x', ?, ?) ON CONFLICT (id) DO NOTHING`).run(id, username, username, `${username}@x`, role, extra.is_owner ? 1 : 0);
+ 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', ?) ON CONFLICT DO NOTHING").run(SID.ann);
+ await raw.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (2, 'network', '102', ?) ON CONFLICT DO NOTHING").run(SID.bob);
+
+ const config = require('../server/config');
+ config.baseUrl = 'https://openvibe.live';
+ const service = require('../server/agent-streams/service');
+ require('../server/agent-streams/sharing')._setNetwork({ json: async () => ({}) });
+ const ann = await db.getUserById(1);
+ const bob = await db.getUserById(2);
+
+ // Chat is another service: record what Live asks it, and make bob a moderator of ann's channel.
+ const roomCalls = [];
+ const chat = require('../server/chat/chat-client');
+ chat.available = () => true;
+ chat.agentRoom = async (body) => { roomCalls.push(body); return { room: 'agent:' + body.agent_stream_id }; };
+ const moderation = require('../server/chat/moderation-client');
+ moderation.getChannelModeration = async () => ({ settings: {}, moderator_ids: [2] });
+
+ const make = (visibility, { ready = true } = {}) => service.create({
+ user: ann,
+ title: 'Fixing the checkout tests ' + visibility,
+ task: { source: 'codes', ref: 'tsk_1', title: 'Run the checkout tests', url: 'https://openvibe.codes/tasks/1' },
+ visibility, copilotReady: ready,
+ });
+ const publicStream = await make('public');
+ const privateStream = await make('private');
+ const deskStream = await make('unlisted', { ready: false });
+
+ 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 } = {}) => {
+ const res = await fetch(`${base}${url}`, {
+ method, headers: { 'Content-Type': 'application/json', 'x-test-user': String(as) },
+ 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 };
+ };
+ const control = (stream, body, as = 1) => call('POST', `/api/streams/agent/${stream.public_id}/control`, { as, body });
+
+ quiet('agent streams S5: Co-pilot');
+
+ await check('a public or unlisted stream with a ready service can be Co-pilot; a private one cannot', async () => {
+ let r = await call('PUT', `/api/streams/agent/${publicStream.public_id}/voice`, { body: { mode: 'copilot' } });
+ assert.strictEqual(r.status, 200, JSON.stringify(r.data));
+ assert.strictEqual(r.data.voice.mode, 'copilot');
+ r = await call('PUT', `/api/streams/agent/${privateStream.public_id}/voice`, { body: { mode: 'copilot' } });
+ assert.strictEqual(r.status, 422, JSON.stringify(r.data));
+ assert.strictEqual(r.data.code, 'copilot_needs_public');
+ r = await call('PUT', `/api/streams/agent/${deskStream.public_id}/voice`, { body: { mode: 'copilot' } });
+ assert.strictEqual(r.status, 422, JSON.stringify(r.data));
+ assert.strictEqual(r.data.code, 'copilot_not_ready');
+ r = await call('PUT', `/api/streams/agent/${deskStream.public_id}/voice`, { body: { mode: 'narrated' } });
+ assert.strictEqual(r.status, 200, 'Narrated still works without a digest');
+ r = await call('POST', `/api/streams/agent/${deskStream.public_id}/copilot`, { body: { ready: true } });
+ assert.strictEqual(r.status, 200, JSON.stringify(r.data));
+ r = await call('PUT', `/api/streams/agent/${deskStream.public_id}/voice`, { body: { mode: 'copilot' } });
+ assert.strictEqual(r.status, 200, JSON.stringify(r.data));
+ });
+
+ await check('the room is made for a public or unlisted stream, never a private one', async () => {
+ assert.ok(roomCalls.some((c) => c.agent_stream_id === publicStream.public_id));
+ assert.ok(roomCalls.some((c) => c.agent_stream_id === deskStream.public_id));
+ assert.ok(!roomCalls.some((c) => c.agent_stream_id === privateStream.public_id));
+ assert.strictEqual(roomCalls.find((c) => c.agent_stream_id === publicStream.public_id).owner_subject, SID.ann);
+ });
+
+ await check('the owner pauses and resumes, and the state carries it', async () => {
+ let r = await control(publicStream, { control: 'pause', paused: true });
+ assert.strictEqual(r.status, 200, JSON.stringify(r.data));
+ assert.strictEqual(r.data.paused, true);
+ assert.strictEqual(r.data.paused_by, 'ann');
+ r = await call('GET', `/api/streams/agent/${publicStream.public_id}/state`);
+ assert.strictEqual(r.status, 200, JSON.stringify(r.data));
+ assert.strictEqual(r.data.paused, true);
+ assert.strictEqual(r.data.paused_by, 'ann');
+ r = await control(publicStream, { control: 'pause', paused: false });
+ assert.strictEqual(r.status, 200);
+ assert.strictEqual(r.data.paused, false);
+ });
+
+ await check("a moderator cannot resume the owner's pause, but can pause and resume their own", async () => {
+ await control(publicStream, { control: 'pause', paused: true }, 1);
+ let r = await control(publicStream, { control: 'pause', paused: false }, 2);
+ assert.strictEqual(r.status, 403, JSON.stringify(r.data));
+ assert.strictEqual(r.data.code, 'owner_pause');
+ await control(publicStream, { control: 'pause', paused: false }, 1);
+ r = await control(publicStream, { control: 'pause', paused: true }, 2);
+ assert.strictEqual(r.status, 200, JSON.stringify(r.data));
+ assert.strictEqual(r.data.paused_by, 'bob');
+ r = await control(publicStream, { control: 'pause', paused: false }, 2);
+ assert.strictEqual(r.status, 200, 'the moderator resumes their own pause');
+ });
+
+ await check('a stranger may not press a control', async () => {
+ const r = await control(publicStream, { control: 'pause', paused: true }, 0);
+ assert.strictEqual(r.status, 403, JSON.stringify(r.data));
+ });
+
+ await check('curtain and mute act, and every press is in the log', async () => {
+ let r = await control(publicStream, { control: 'curtain', up: true }, 2);
+ assert.strictEqual(r.status, 200, JSON.stringify(r.data));
+ assert.strictEqual(r.data.curtain.up, true);
+ r = await control(publicStream, { control: 'mute', muted: true }, 1);
+ assert.strictEqual(r.status, 200, JSON.stringify(r.data));
+ assert.strictEqual(r.data.muted, true);
+ r = await call('GET', `/api/streams/agent/${publicStream.public_id}/controls`);
+ assert.strictEqual(r.status, 200, JSON.stringify(r.data));
+ const controls = r.data.controls.map((c) => c.control);
+ for (const name of ['pause', 'curtain', 'mute']) assert.ok(controls.includes(name), `${name} is logged`);
+ assert.ok(r.data.controls.some((c) => c.by.handle === 'bob' && c.by.role === 'moderator'));
+ });
+
+ await check('end ends the stream and closes the Co-pilot reading', async () => {
+ const r = await control(publicStream, { control: 'end' }, 2);
+ assert.strictEqual(r.status, 200, JSON.stringify(r.data));
+ assert.strictEqual(r.data.state, 'ended');
+ assert.ok(roomCalls.some((c) => c.agent_stream_id === publicStream.public_id && c.close === true), 'the room is closed');
+ const after = await control(publicStream, { control: 'pause', paused: true }, 1);
+ assert.strictEqual(after.status, 200, 'an ended stream still logs the press; pause has no effect on the desk');
+ });
+
+ const left = await db.get('SELECT COUNT(*) AS n FROM agent_stream_controls');
+ assert.ok(Number(left.n) >= 8, 'every press is stored');
+ server.close();
+ quiet(failures ? `agent streams S5: ${failures} failed` : 'agent streams S5: Co-pilot mode, the four controls and the log');
+ process.exit(failures ? 1 : 0);
+})().catch((e) => { console.error(e); process.exit(1); });