Agent Streams S1: publish grant for Codes, Actor and a node acting for its owner (Contracts 0.298.0) (#589)

ddiff --git a/package-lock.json b/package-lock.json
iindex b835444c..4a5c7663 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -20,7 +20,7 @@
          "iovalkey": "^0.4.0",
          "jsonwebtoken": "^9.0.0",
          "multer": "^1.4.5-lts.1",
--        "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.297.0",
++        "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.298.0",
          "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.90.0",
          "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.115.1",
          "pg": "^8.23.0",
@@@ -1722,9 +1722,9 @@
        }
      },
      "node_modules/openvibe-contracts": {
--      "version": "0.297.0",
--      "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.297.0",
--      "integrity": "sha512-FycRElRNzeFLEfzVUKIyE3ieKphUrQbAbsprnvs0Uu7ZdlBKJG6wQV/9gD77Neu1Etcsyezf9Boyq+MsjuraVA==",
++      "version": "0.298.0",
++      "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.298.0",
++      "integrity": "sha512-G6JmHxYQOKQ7kjjufsGBD1X9QeSgfHZT/UuDRbHPvJ8+FFHZ95nzlNLtbRkgBcW5AlgmmxBOq/1Tj80XbmlmAg==",
        "license": "AGPL-3.0-only",
        "dependencies": {
          "ajv": "^8.17.1",
ddiff --git a/package.json b/package.json
iindex 2d2871fe..a415b6e1 100644
---- a/package.json
++++ b/package.json
@@@ -24,7 +24,7 @@
      "iovalkey": "^0.4.0",
      "jsonwebtoken": "^9.0.0",
      "multer": "^1.4.5-lts.1",
--    "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.297.0",
++    "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.298.0",
      "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.90.0",
      "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.115.1",
      "pg": "^8.23.0",
ddiff --git a/server/identity/principals.js b/server/identity/principals.js
iindex 80c67f62..9859cca3 100644
---- a/server/identity/principals.js
++++ b/server/identity/principals.js
@@@ -795,6 +795,19 @@ const DEFAULT_GRANTS = [
      // cannot ask treats every purpose as off. None of these rows was ever granted or revoked before, so none needs a
      // RESTORED_GRANTS entry (that list is only for a default that was once revoked).
      ...['live', 'vision', 'ai', 'media-hub', 'cam', 'chat', 'actor'].map((svc) => [svc, 'network.consent.read', SELF_AUDIENCE, []]),
++    // The Agent Streams plan (S1, Contracts 0.298.0): Codes and Actor start and steer the person's agent stream on
++    // OpenVibe.Live (live.agent_stream.publish, the audience openvibe.live). They act for the person and for nobody
++    // else, so the grant is unscoped and the audience is Live's alone.
++    ['codes', 'live.agent_stream.publish', 'openvibe.live', []],
++    ['actor', 'live.agent_stream.publish', 'openvibe.live', []],
++];
++
++// What a node principal (a person's own machine, paired with its credential) may do for its owner outside Network's
++// own audience. The Agent Streams plan (S1, contracts 0.298.0): the desk publishes the person's agent stream on
++// OpenVibe.Live, so the machine's token carries live.agent_stream.publish for the audience openvibe.live and names
++// its owner (on_behalf_of). The machine acts for its owner, never for the person directly and never for anyone else.
++const NODE_GRANTS = [
++    ['openvibe.live', 'live.agent_stream.publish'],
  ];
  
  // Grants withdrawn by decision; applied at every boot so an old default can't come back.
@@@ -1097,4 +1110,4 @@ function guardAny(caps, opts) {
      };
  }
  
--module.exports = { ensureSchema, issueToken, guard, guardAny, grantsFor, recordDecision, DEFAULT_GRANTS, REVOKED_GRANTS, RESTORED_GRANTS, TOKEN_TTL_S, SELF_AUDIENCE };
++module.exports = { ensureSchema, issueToken, guard, guardAny, grantsFor, recordDecision, DEFAULT_GRANTS, NODE_GRANTS, REVOKED_GRANTS, RESTORED_GRANTS, TOKEN_TTL_S, SELF_AUDIENCE };
ddiff --git a/server/registry/node-principals.js b/server/registry/node-principals.js
iindex 7815a050..bfeefc77 100644
---- a/server/registry/node-principals.js
++++ b/server/registry/node-principals.js
@@@ -32,7 +32,7 @@ const crypto = require('crypto');
  const express = require('express');
  const { ids, http, serviceAuth, validate, assertValid } = require('openvibe-contracts');
  const { BOOTSTRAP_CELL, RegistryError } = require('./cells');
--const { TOKEN_TTL_S, SELF_AUDIENCE } = require('../identity/principals');
++const { TOKEN_TTL_S, SELF_AUDIENCE, NODE_GRANTS } = require('../identity/principals');
  const eventRelay = require('../developer/event-relay');
  
  const CODE_TTL_MS = 10 * 60 * 1000;
@@@ -393,13 +393,22 @@ async function issueNodeToken(db, { clientId, clientSecret, audience, privateKey
      // Connect (openvibe.website) binds a share to one machine and checks the node token at every tunnel hello (ADR-057
      // decision 4, CON-4). The token carries no capability: Connect authorises the node by its own share binding.
      const anyNode = p.owner_kind === 'user' && NODE_WIDE_AUDIENCES.includes(aud);
--    if (aud !== SELF_AUDIENCE && !anyNode && !(p.paired_by_service && aud === `openvibe.${p.paired_by_service}`)) {
--        return { status: 400, body: { error: 'invalid_scope', error_description: `a node token is for ${SELF_AUDIENCE}${p.paired_by_service ? ` or openvibe.${p.paired_by_service}` : ''}${p.owner_kind === 'user' ? `, or ${NODE_WIDE_AUDIENCES.join(', ')}` : ''}` } };
++    // The Agent Streams plan (S1, Contracts 0.298.0): the desk publishes the person's agent stream on OpenVibe.Live, so
++    // a person's machine may ask for a token for openvibe.live that carries live.agent_stream.publish and names its
++    // owner (on_behalf_of). It is the owner's stream; a machine without an owner (a platform one) gets nothing.
++    const nodeCaps = p.owner_kind === 'user' && p.owner_subject ? NODE_GRANTS.filter(([a]) => a === aud).map(([, c]) => c) : [];
++    if (aud !== SELF_AUDIENCE && !anyNode && !nodeCaps.length && !(p.paired_by_service && aud === `openvibe.${p.paired_by_service}`)) {
++        const audiences = [SELF_AUDIENCE, ...(p.paired_by_service ? [`openvibe.${p.paired_by_service}`] : []),
++            ...(p.owner_kind === 'user' ? [...NODE_WIDE_AUDIENCES, ...NODE_GRANTS.map(([a]) => a)] : [])];
++        return { status: 400, body: { error: 'invalid_scope', error_description: `a node token is for ${audiences.join(', ')}` } };
      }
      const iat = Math.floor(now / 1000);
      const claims = {
          iss: issuer, sub: `node:${p.id}`, actor_type: 'node', aud: [aud],
--        cap: aud === SELF_AUDIENCE ? ['network.node.self.manage'] : [],
++        cap: aud === SELF_AUDIENCE ? ['network.node.self.manage'] : nodeCaps,
++        // The person the machine acts for, on the audiences where it acts for them (the Agent Streams plan's Live
++        // token): Live reads the owner of an agent stream from it. Other audiences name nobody.
++        ...(nodeCaps.length && p.owner_subject ? { on_behalf_of: p.owner_subject } : {}),
          ...(p.owner_kind === 'project' && p.project_id ? { project_id: p.project_id } : {}),
          iat, exp: iat + TOKEN_TTL_S, jti: `tok_${crypto.randomBytes(12).toString('hex')}`,
      };
ddiff --git a/test/agent-stream-grants.test.js b/test/agent-stream-grants.test.js
nnew file mode 100644
iindex 00000000..ecd3d71c
---- /dev/null
++++ b/test/agent-stream-grants.test.js
@@@ -0,0 +1,110 @@
++'use strict';
++/**
++ * The Agent Streams plan (S1, Contracts 0.298.0): who may publish a person's agent stream on OpenVibe.Live.
++ *
++ * Codes and Actor hold live.agent_stream.publish for the audience openvibe.live as service principals: they act for the
++ * person and for nobody else. A node principal (a person's own machine, paired with its credential) holds the same
++ * capability for the same audience, but only as the owner's machine: the token names its owner (on_behalf_of), so Live
++ * knows whose stream it is. A machine with no owner (a platform one) gets nothing, and no audience but Live's carries
++ * the capability.
++ *
++ *   node test/agent-stream-grants.test.js
++ */
++const assert = require('assert');
++const crypto = require('crypto');
++const contracts = require('openvibe-contracts');
++const principals = require('../server/identity/principals');
++const nodePrincipals = require('../server/registry/node-principals');
++
++const CAP = 'live.agent_stream.publish';
++const ISSUER = 'https://openvibe.network';
++
++// The service principals: Codes and Actor publish the person's agent stream on Live.
++const grant = (client) => principals.DEFAULT_GRANTS.find((g) => g[0] === client && g[1] === CAP && g[2] === 'openvibe.live');
++for (const client of ['codes', 'actor']) {
++    const g = grant(client);
++    assert.ok(g, `${client} may publish the person's agent stream on Live`);
++    assert.deepStrictEqual(g[3], [], `${client} holds it for Live's audience alone`);
++}
++assert.ok(!principals.DEFAULT_GRANTS.some((g) => g[1] === CAP && g[2] !== 'openvibe.live'),
++    'the capability is granted for Live and for no other audience');
++assert.ok(!principals.DEFAULT_GRANTS.some((g) => g[0] !== 'codes' && g[0] !== 'actor' && g[1] === CAP),
++    'no other service principal holds it');
++
++// The node principals: a person's machine publishes for its owner.
++const nodeGrant = principals.NODE_GRANTS.find(([aud, cap]) => aud === 'openvibe.live' && cap === CAP);
++assert.ok(nodeGrant, "a person's machine may publish for its owner on Live");
++
++const OWNER = 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ';
++const CREDENTIAL = 'node-credential-for-the-agent-stream-test';
++const principal = {
++    id: 'nod_01JAB2C3D4E5F6G7H8J9K0MNPQ',
++    status: 'active',
++    owner_kind: 'user',
++    owner_subject: OWNER,
++    paired_by_service: 'bot',
++    credential_hash: crypto.createHash('sha256').update(CREDENTIAL).digest('hex'),
++};
++const dbFor = (row) => ({
++    prepare: (sql) => ({
++        get: async () => (sql.includes('platform_node_principals') ? row : null),
++        run: async () => ({ changes: 1 }),
++    }),
++});
++const keys = crypto.generateKeyPairSync('rsa', {
++    modulusLength: 2048,
++    publicKeyEncoding: { type: 'spki', format: 'pem' },
++    privateKeyEncoding: { type: 'pkcs8', format: 'pem' },
++});
++
++(async () => {
++    const issue = (row, audience) => nodePrincipals.issueNodeToken(dbFor(row), {
++        clientId: row.id, clientSecret: CREDENTIAL, audience, privateKey: keys.privateKey, issuer: ISSUER,
++    });
++
++    const r = await issue(principal, 'openvibe.live');
++    assert.strictEqual(r.status, 200, JSON.stringify(r.body));
++    assert.strictEqual(r.body.scope, CAP, 'the token carries the publish capability and nothing else');
++    const v = contracts.serviceAuth.verifyServiceToken(r.body.access_token, {
++        publicKey: keys.publicKey, issuer: ISSUER, audience: 'openvibe.live',
++    });
++    assert.ok(v.ok, v.reason);
++    assert.deepStrictEqual([v.claims.sub, v.claims.actor_type, v.claims.aud, v.claims.cap],
++        [`node:${principal.id}`, 'node', ['openvibe.live'], [CAP]]);
++    assert.strictEqual(v.claims.on_behalf_of, OWNER, 'the token names the owner it acts for');
++    assert.ok(contracts.validate('identity.service-token-claims@1', v.claims).valid,
++        'the claims fit identity.service-token-claims@1');
++
++    const other = await issue(principal, 'openvibe.chat');
++    assert.deepStrictEqual([other.status, other.body.error], [400, 'invalid_scope'], 'no other audience');
++
++    const website = await issue(principal, 'openvibe.website');
++    assert.strictEqual(website.status, 200, JSON.stringify(website.body));
++    const w = contracts.serviceAuth.verifyServiceToken(website.body.access_token, {
++        publicKey: keys.publicKey, issuer: ISSUER, audience: 'openvibe.website',
++    });
++    assert.ok(w.ok, w.reason);
++    assert.deepStrictEqual(w.claims.cap, [], 'the Connect audience carries no capability');
++    assert.strictEqual(w.claims.on_behalf_of, undefined, 'only the audience where the machine acts for its owner names the owner');
++
++    const self = await issue(principal, 'openvibe.network');
++    assert.strictEqual(self.status, 200, JSON.stringify(self.body));
++    const st = contracts.serviceAuth.verifyServiceToken(self.body.access_token, {
++        publicKey: keys.publicKey, issuer: ISSUER, audience: 'openvibe.network',
++    });
++    assert.ok(st.ok, st.reason);
++    assert.strictEqual(st.claims.on_behalf_of, undefined, 'self-management names nobody');
++
++    const platform = { ...principal, owner_kind: 'platform', owner_subject: null };
++    const none = await issue(platform, 'openvibe.live');
++    assert.deepStrictEqual([none.status, none.body.error], [400, 'invalid_scope'], 'a machine with no owner publishes for nobody');
++
++    const unbound = { ...principal, owner_subject: null };
++    const noOwner = await issue(unbound, 'openvibe.live');
++    assert.deepStrictEqual([noOwner.status, noOwner.body.error], [400, 'invalid_scope'], 'a user machine must name its owner');
++
++    const known = contracts.capabilities.get(CAP);
++    if (known) assert.strictEqual(known.status, 'active', `${CAP} is an active capability`);
++    else console.log(`note: the pinned contracts do not know ${CAP} yet (the pin round lands it)`);
++    console.log('agent stream grants: codes and actor publish for the person; a node principal publishes for its owner');
++})().catch((e) => { console.error(e); process.exit(1); });
ddiff --git a/test/registry-topics.test.js b/test/registry-topics.test.js
iindex ee2ed540..cc182557 100644
---- a/test/registry-topics.test.js
++++ b/test/registry-topics.test.js
@@@ -70,7 +70,7 @@ const rel = (contractsVersion, packages) => [200, { service: 'x', release: 'abcd
      let r = await get('/api/v1/registry/topics');
      assert.ok(r.body.topics.length >= produced.size, 'no longer empty');
      r = await get('/api/v1/registry/topics?producer=live');
--    assert.deepStrictEqual(r.body.topics.map(t => t.topic).sort(), ['live.index_document.deleted', 'live.index_document.upserted', 'live.moderation.action', 'live.release.deployed', 'live.setup.task_completed', 'live.stream.ended', 'live.stream.started']);
++    assert.deepStrictEqual(r.body.topics.map(t => t.topic).sort(), ['live.agent_stream.changed', 'live.index_document.deleted', 'live.index_document.upserted', 'live.moderation.action', 'live.release.deployed', 'live.setup.task_completed', 'live.stream.ended', 'live.stream.started']);
      r = await get('/api/v1/registry/topics?consumer=network');
      assert.deepStrictEqual(r.body.topics.map(t => t.topic).sort(), [...TOPICS].sort());
      r = await get('/api/v1/registry/topics?prefix=media.vod.');