openvibers/openvibe.actor Commits

Actor: subscribe to agent.step.* (the topic was nested, so Events never saw it) (#308)

OP OpenVibers committed 6 days ago · 5d418c6a10 · 2 files, +50 −3

The loop passed AGENT_TOPICS as one item, so Events was asked for
topic_pattern ["agent.step.*"] and refused it: the one timeline behind the Dock
could never receive a step. Spread the list, return the full topic set, and test
that every posted topic is a plain string.

Changes

ddiff --git a/server/events-consumer.js b/server/events-consumer.js
iindex 0f31709..8b556d7 100644
---- a/server/events-consumer.js
++++ b/server/events-consumer.js
@@@ -35,7 +35,7 @@ function startSubscriptions({ config, port, secret, eventsUrl = config.events.ur
          // The ADR-033 topics, chat.message.created while asking Actor from chat is on (chat/asks.js), and the
          // agent's own steps (design 2026-11-07): every mode publishes agent.step.* into the network's agent
          // namespace, and Actor keeps the one timeline behind the Dock.
--        for (const topic of [...TOPICS, AGENT_TOPICS, ...extra]) {
++        for (const topic of [...TOPICS, ...AGENT_TOPICS, ...extra]) {
              if (mine.some((s) => s.topic_pattern === topic)) continue;
              const r = await call('POST', '/api/v1/subscriptions', { topic_pattern: topic, endpoint, secret });
              if (!r.ok && r.status !== 409) throw new Error(`subscribing to ${topic}: ${r.status} ${r.body.code || ''}`);
@@@ -52,7 +52,7 @@ function startSubscriptions({ config, port, secret, eventsUrl = config.events.ur
      }); };
      timer = setTimeout(run, delays[0]); if (timer.unref) timer.unref();
      // Graceful stop: no further attempts (they run again at the next start).
--    return { topics: TOPICS, endpoint, stop() { stopped = true; if (timer) clearTimeout(timer); timer = null; } };
++    return { topics: [...TOPICS, ...AGENT_TOPICS, ...extra], endpoint, stop() { stopped = true; if (timer) clearTimeout(timer); timer = null; } };
  }
  
--module.exports = { startSubscriptions, TOPICS };
++module.exports = { startSubscriptions, TOPICS, AGENT_TOPICS };
ddiff --git a/test/events-consumer.test.js b/test/events-consumer.test.js
nnew file mode 100644
iindex 0000000..9da71a7
---- /dev/null
++++ b/test/events-consumer.test.js
@@@ -0,0 +1,47 @@
++'use strict';
++/**
++ * The agent's steps (design 2026-11-07): Actor subscribes to agent.step.* at OpenVibe.Events, beside the ADR-033
++ * account topics, as one plain string pattern. (The first version nested the list, so Events was asked for
++ * topic_pattern ["agent.step.*"] and refused it; no timeline row could ever arrive.)
++ *
++ *   node test/events-consumer.test.js
++ */
++const assert = require('assert');
++const { startSubscriptions } = require('../server/events-consumer');
++const { TOPICS } = require('../server/identity/account-data');
++
++const ENDPOINT = 'http://127.0.0.1:4950/internal/events';
++const CONFIG = {
++    events: { url: 'http://events.test', endpoint: ENDPOINT },
++    oauth: { clientId: 'actor', clientSecret: 'test-secret' },
++    networkInternalUrl: 'http://network.test',
++};
++
++const posted = [];
++const fetchImpl = async (url, init = {}) => {
++    const u = String(url);
++    const reply = (body, status = 200) => ({ ok: status < 400, status, json: async () => body });
++    if (u.endsWith('/oauth/token')) return reply({ access_token: 'tok', expires_in: 300 });
++    const method = init.method || 'GET';
++    if (u.endsWith('/api/v1/subscriptions') && method === 'GET') return reply({ subscriptions: [] });
++    if (u.endsWith('/api/v1/subscriptions') && method === 'POST') {
++        const body = JSON.parse(init.body);
++        posted.push(body.topic_pattern);
++        return reply({ id: `sub_${posted.length}` }, 201);
++    }
++    return reply({}, 404);
++};
++
++(async () => {
++    const s = startSubscriptions({ config: CONFIG, port: 4950, secret: 'events-secret', fetchImpl, log: { log() {}, warn() {} } });
++    assert.ok(s, 'it starts when the events URL and the OAuth secret are set');
++    await new Promise((r) => setTimeout(r, 50));
++    s.stop();
++
++    assert.deepStrictEqual(posted, [...TOPICS, 'agent.step.*'], 'the account topics first, then the agent namespace');
++    assert.ok(posted.includes('agent.step.*'), 'agent.step.* is one topic string');
++    assert.ok(posted.every((t) => typeof t === 'string' && t.length > 0), 'every topic is a string, never a list');
++    assert.deepStrictEqual(s.topics, [...TOPICS, 'agent.step.*'], 'the started consumer reports what it subscribes to');
++    console.log('events consumer: Actor subscribes to agent.step.* beside the account topics');
++    console.log('all passed');
++})().catch((e) => { console.error(e); process.exit(1); });