Wait for a Watch alert, a channel going live or an OpenVibe release (A2b) (#268)

ddiff --git a/server/runtime/tools/wait_for.js b/server/runtime/tools/wait_for.js
iindex dba223a..5018c28 100644
---- a/server/runtime/tools/wait_for.js
++++ b/server/runtime/tools/wait_for.js
@@@ -16,17 +16,20 @@
   */
  const { personOf } = require('./_openvibe');
  
--const SOURCES = ['codes', 'run', 'time'];
++const SOURCES = ['codes', 'run', 'watch', 'live', 'release', 'time'];
  
  module.exports = {
      name: 'wait_for',
--    description: 'Wait for something to happen without spending anything while you wait. Use it when the answer depends on something that is still running (a coding task on the person\'s own machine, a job in the cloud) or when you should simply look again later. The task keeps its place, its budget and its plan; no model runs until the event arrives or the deadline passes. Say the source, what to wait for, how long at most, and the short line the person should read.',
++    description: 'Wait for something to happen without spending anything while you wait. Use it when the answer depends on something that is still running (a coding task on the person\'s own machine, a job in the cloud), on something that has not happened yet (one of the person\'s own Watch alerts firing, a channel going live, a new release of an OpenVibe service) or when you should simply look again later. The task keeps its place, its budget and its plan; no model runs until the event arrives or the deadline passes. Say the source, what to wait for, how long at most, and the short line the person should read.',
      input: {
          type: 'object',
          properties: {
--            source: { type: 'string', enum: SOURCES, description: 'codes: a coding task on a repository; run: a job in the cloud; time: just some time' },
++            source: { type: 'string', enum: SOURCES, description: 'codes: a coding task on a repository; run: a job in the cloud; watch: one of the person\'s OpenVibe.Watch alerts fires; live: a channel goes live on OpenVibe.Live; release: an OpenVibe service ships a new release; time: just some time' },
              repo: { type: 'string', description: 'owner/name of the repository, for source "codes"' },
              job_id: { type: 'string', description: 'the job id (job_…), for source "run"' },
++            watch_id: { type: 'string', description: 'for source "watch": one watch\'s id; leave it out to wait for any of the person\'s watches' },
++            channel: { type: 'string', description: 'for source "live": the channel\'s username on OpenVibe.Live' },
++            service: { type: 'string', description: 'for source "release": the service id, e.g. live, chat, media' },
              seconds: { type: 'number', description: 'how long to wait at most, in seconds (the deadline)' },
              on_timeout: { type: 'string', enum: ['ask', 'retry', 'fail'], description: 'what to do when nothing happened: ask the person, try that step again once, or fail the task' },
              line: { type: 'string', description: 'the short line the person reads, e.g. "waiting for your Codes task on owner/repo to finish"' },
@@@ -47,7 +50,9 @@ module.exports = {
          if (!SOURCES.includes(source)) {
              return { text: JSON.stringify({ error: `source must be one of ${SOURCES.join(', ')}` }), isError: true };
          }
--        const target = source === 'codes' ? String(input.repo || '') : source === 'run' ? String(input.job_id || '') : null;
++        const target = source === 'codes' ? String(input.repo || '') : source === 'run' ? String(input.job_id || '')
++            : source === 'watch' ? String(input.watch_id || '') : source === 'live' ? String(input.channel || '')
++                : source === 'release' ? String(input.service || '') : null;
          const seconds = Number(input.seconds);
          if (!Number.isFinite(seconds) || seconds <= 0) {
              return { text: JSON.stringify({ error: 'seconds must be how long to wait (a positive number of seconds)' }), isError: true };
ddiff --git a/server/waits/index.js b/server/waits/index.js
iindex 8fd8b01..2611dd2 100644
---- a/server/waits/index.js
++++ b/server/waits/index.js
@@@ -21,15 +21,28 @@ const { guard } = require('./guard');
  const { topicMatches, personId, isValidPattern } = require('./match');
  const { limits } = require('./limits');
  
--/** The sources a wait may name in A2a: a Codes task, a Run job, or just time. */
++/**
++ * The sources a wait may name. A2a: a Codes task, a Run job, or just time. A2b (11-02):
++ *   watch    one of the person's own OpenVibe.Watch alerts firing (watch.watch.triggered; only their own: the
++ *            recipient is the person, and a watch id narrows it to one)
++ *   live     a channel going live on OpenVibe.Live (live.stream.started, public: any channel by its name)
++ *   release  a new release of an OpenVibe service (host.release.published, public: the service id, as live or chat)
++ * A Chat source waits for Chat's payload contract and a room id a person can name (its rooms are stream-keyed objects).
++ */
  const SOURCES = {
      codes: { topic: 'codes.task.*', states: ['succeeded', 'failed', 'cancelled'] },
      run: { topic: 'run.job.*', states: ['succeeded', 'failed', 'cancelled', 'expired'] },
++    watch: { topic: 'watch.watch.triggered', states: null },
++    live: { topic: 'live.stream.started', states: null },
++    release: { topic: 'host.release.published', states: null },
      time: { topic: null, states: [] },
  };
  
  const REPO_RE = /^[^/\s]+\/[^/\s]+$/;
  const JOB_RE = /^job_[0-9A-HJKMNP-TV-Z]{26}$/;
++const WATCH_RE = /^[A-Za-z0-9_-]{4,64}$/;
++const CHANNEL_RE = /^[A-Za-z0-9_-]{2,40}$/;
++const SERVICE_RE = /^[a-z][a-z0-9-]{1,40}$/;
  
  function createWaits({ s, config, log = console, now = () => Date.now(), events = null, engine = null }) {
      const L = limits(config);
@@@ -78,7 +91,7 @@ function createWaits({ s, config, log = console, now = () => Date.now(), events
          if (src.topic) {
              until.topic = src.topic;
              until.from = iso(now());
--            filter['payload.state'] = src.states;
++            if (src.states) filter['payload.state'] = src.states;
          } else {
              until.at = deadlineAt;
          }
@@@ -92,6 +105,23 @@ function createWaits({ s, config, log = console, now = () => Date.now(), events
              if (!JOB_RE.test(job)) throw Object.assign(new Error('a Run wait names the job id'), { code: 'actor.wait.target' });
              filter['payload.job_id'] = job;
              filter['payload.requester.id'] = me;
++        } else if (source === 'watch') {
++            // Only the person's own alerts: the event names who is notified. A watch id narrows it to one of them.
++            filter['payload.recipient'] = me;
++            const id = String(target || '').trim();
++            if (id) {
++                if (!WATCH_RE.test(id)) throw Object.assign(new Error('a Watch wait names the watch by its id'), { code: 'actor.wait.target' });
++                filter['payload.watch_id'] = id;
++            }
++        } else if (source === 'live') {
++            const name = String(target || '').trim().replace(/^@/, '');
++            if (!CHANNEL_RE.test(name)) throw Object.assign(new Error('a Live wait names the channel by its username'), { code: 'actor.wait.target' });
++            // Live's usernames are matched as Live writes them; the two common spellings are both accepted.
++            filter['payload.channel.username'] = [...new Set([name, name.toLowerCase()])];
++        } else if (source === 'release') {
++            const service = String(target || '').trim().toLowerCase();
++            if (!SERVICE_RE.test(service)) throw Object.assign(new Error('a release wait names the service, for example live or chat'), { code: 'actor.wait.target' });
++            filter['payload.service'] = service;
          }
          // The bounds: a person may not keep a hundred subscriptions, and a task may not wait on a hundred things.
          const openForPerson = await store.countOpen(s, requester);
@@@ -116,6 +146,9 @@ function createWaits({ s, config, log = console, now = () => Date.now(), events
      function defaultLine(source, target) {
          if (source === 'codes') return `waiting for your Codes task on ${target} to finish`;
          if (source === 'run') return `waiting for your run ${target} to finish`;
++        if (source === 'watch') return target ? `waiting for your watch ${target} to fire` : 'waiting for one of your watches to fire';
++        if (source === 'live') return `waiting for ${String(target || '').replace(/^@/, '')} to go live`;
++        if (source === 'release') return `waiting for a new release of OpenVibe ${target}`;
          return 'waiting for a while';
      }
  
ddiff --git a/test/wait-sources.test.js b/test/wait-sources.test.js
nnew file mode 100644
iindex 0000000..5f6956d
---- /dev/null
++++ b/test/wait-sources.test.js
@@@ -0,0 +1,112 @@
++'use strict';
++/**
++ * The wait sources of plan A2b: one of the person's own Watch alerts firing, a channel going live on OpenVibe.Live,
++ * a new release of an OpenVibe service. Each source's filter matches its own payload and not another's; a Watch
++ * alert for another person never wakes the wait (the payload names who is notified); a stream on another channel or a
++ * release of another service does not either. The wait_for tool passes each source's target through.
++ *
++ *   node test/wait-sources.test.js
++ */
++const assert = require('assert');
++const { testDb } = require('./helpers/db');
++const { createStore } = require('../server/db');
++const { createWaits } = require('../server/waits');
++const waitFor = require('../server/runtime/tools/wait_for');
++
++const A = 'usr_01JZ0000000000000000000AAA';
++const B = 'usr_01JZ0000000000000000000BBB';
++let clock = Date.now();
++const at = () => new Date(clock + 1000).toISOString();
++let n = 0;
++const eventId = () => `evt_01JZ00000000000000000S${String(++n).padStart(4, '0')}`;
++
++const watchEvent = (recipient, watchId = 'wch_lisbon_flights') => ({ event_id: eventId(), event_type: 'watch.watch.triggered', source: 'watch', at: at(), subject: { type: 'watch', id: watchId },
++    payload: { watch_id: watchId, recipient, project_id: null, name: 'Flights to Lisbon', condition: { below: 120 }, observation: { id: 'obs_1', observed_at: at(), value: 99, previous_value: 140 }, fired_at: at() } });
++const liveEvent = (username) => ({ event_id: eventId(), event_type: 'live.stream.started', source: 'live', at: at(), subject: { type: 'stream', id: '2722' },
++    payload: { stream_id: 2722, channel: { username, display_name: username, url: `https://openvibe.live/${username}`, subject: null }, title: 'Building', category: null, protocol: 'webrtc', is_nsfw: false, started_at: at() } });
++const releaseEvent = (service) => ({ event_id: eventId(), event_type: 'host.release.published', source: 'host', at: at(), subject: { type: 'release', id: `${service}-r1` },
++    payload: { service, release: 'rel_20261102T160000Z_abc123', commit: 'abc123', origin: `https://openvibe.${service}`, deployed_at: at() } });
++
++(async () => {
++    const d = await testDb();
++    const s = createStore(d.db, { now: () => Date.now() });
++    const resumed = [];
++    const engine = {
++        resumeWithEvent: async (taskId, payload) => { resumed.push({ taskId, payload }); return { id: taskId }; },
++        resumeWithNotice: async () => null, failTask: async () => null, notifyWaitTimeout: async () => null,
++    };
++    const config = { waits: { perPerson: 20, perTask: 1, perTaskLifetime: 5, deadlineMs: 6 * 60 * 60_000, filterClauses: 4 } };
++    const waits = createWaits({ s, config, log: { warn() {}, log() {}, error() {} }, now: () => clock, engine, events: null });
++    let t = 0;
++    const mkTask = async (requester = `user:${A}`) => {
++        const id = `tsk_01JZ000000000000000000S${String(++t).padStart(3, '0')}`;
++        await s.db.query("INSERT INTO tasks (id, requester, task, mode, budget_task, budget_day, state, created_at) VALUES ($1, $2, 'wait', 'balanced', 1, 1, 'running', $3)", [id, requester, new Date(clock).toISOString()]);
++        return id;
++    };
++    const wait = async (source, target, requester = `user:${A}`) => waits.create({ taskId: await mkTask(requester), requester, source, target, seconds: 900 });
++    const failed = [];
++    const ok = async (name, fn) => {
++        try { await fn(); process.stdout.write('  ok ' + name + '\n'); } catch (e) { failed.push(name); process.stdout.write('  FAIL ' + name + '\n    ' + (e.stack || String(e)).split('\n').slice(0, 6).join('\n    ') + '\n'); }
++    };
++
++    try {
++        await ok('watch: only the person\'s own alert wakes it; a watch id narrows it to that one', async () => {
++            const any = await wait('watch', '');
++            assert.strictEqual(any.until.topic, 'watch.watch.triggered');
++            assert.strictEqual(any.filter['payload.recipient'], A);
++            assert.ok(!('payload.state' in any.filter), 'a Watch alert has no state to filter on');
++            assert.strictEqual(any.line, 'waiting for one of your watches to fire');
++            assert.strictEqual(await waits.onEvent(watchEvent(B)), 'ignored:no-match', 'another person\'s alert');
++            assert.strictEqual(await waits.onEvent(watchEvent(A)), 'woken:1');
++            assert.strictEqual(resumed.at(-1).payload.event.payload.recipient, A, 'the event reaches the run as data');
++            const one = await wait('watch', 'wch_rent_ceiling');
++            assert.strictEqual(one.filter['payload.watch_id'], 'wch_rent_ceiling');
++            assert.strictEqual(await waits.onEvent(watchEvent(A, 'wch_other')), 'ignored:no-match', 'another of their watches');
++            assert.strictEqual(await waits.onEvent(watchEvent(A, 'wch_rent_ceiling')), 'woken:1');
++            await assert.rejects(wait('watch', 'no spaces allowed'), (e) => e.code === 'actor.wait.target');
++        });
++
++        await ok('live: the channel by its username, either spelling; another channel does not wake it', async () => {
++            const w = await wait('live', '@Maticus');
++            assert.strictEqual(w.until.topic, 'live.stream.started');
++            assert.deepStrictEqual(w.filter['payload.channel.username'], ['Maticus', 'maticus']);
++            assert.strictEqual(w.line, 'waiting for Maticus to go live');
++            assert.strictEqual(await waits.onEvent(liveEvent('somebody')), 'ignored:no-match');
++            assert.strictEqual(await waits.onEvent(liveEvent('maticus')), 'woken:1');
++            await assert.rejects(wait('live', ''), (e) => e.code === 'actor.wait.target', 'a channel is required');
++        });
++
++        await ok('release: a new release of the named service; another service does not wake it', async () => {
++            const w = await wait('release', 'Live');
++            assert.strictEqual(w.until.topic, 'host.release.published');
++            assert.strictEqual(w.filter['payload.service'], 'live');
++            assert.strictEqual(w.line, 'waiting for a new release of OpenVibe Live');
++            assert.strictEqual(await waits.onEvent(releaseEvent('chat')), 'ignored:no-match');
++            assert.strictEqual(await waits.onEvent(releaseEvent('live')), 'woken:1');
++            await assert.rejects(wait('release', 'not a service!'), (e) => e.code === 'actor.wait.target');
++        });
++
++        await ok('an event from before the wait never wakes it', async () => {
++            await wait('live', 'latecomer');
++            const old = liveEvent('latecomer');
++            old.at = new Date(clock - 60_000).toISOString();
++            assert.strictEqual(await waits.onEvent(old), 'ignored:no-match');
++        });
++
++        await ok('the tool: each source\'s target goes through; the model never names the person', async () => {
++            const asked = [];
++            const ctx = { requester: `user:${A}`, taskId: await mkTask(), waits: { create: async (o) => { asked.push(o); return { id: 'wai_x', line: o.line || 'x', deadline_at: new Date(clock + 900_000).toISOString(), on_timeout: 'ask' }; } } };
++            for (const [input, target] of [[{ source: 'watch', watch_id: 'wch_1', seconds: 600 }, 'wch_1'], [{ source: 'live', channel: 'maticus', seconds: 600 }, 'maticus'], [{ source: 'release', service: 'chat', seconds: 600 }, 'chat']]) {
++                const r = await waitFor.run(input, ctx);
++                assert.ok(r.park && r.park.wait_id === 'wai_x', JSON.stringify(r));
++                assert.strictEqual(asked.at(-1).target, target);
++                assert.strictEqual(asked.at(-1).requester, `user:${A}`);
++            }
++            assert.deepStrictEqual(waitFor.input.properties.source.enum, ['codes', 'run', 'watch', 'live', 'release', 'time']);
++        });
++    } finally {
++        await d.close();
++    }
++    if (failed.length) { console.error(`wait-sources: ${failed.length} failed`); process.exit(1); }
++    console.log('wait-sources: all checks passed');
++})().catch((err) => { console.error(err); process.exit(1); });