News: clusters match on their core terms, so one story cannot swallow the wire; resplit regroups the old ones (#51)

ddiff --git a/scripts/resplit-clusters.js b/scripts/resplit-clusters.js
nnew file mode 100644
iindex 0000000..57baacd
---- /dev/null
++++ b/scripts/resplit-clusters.js
@@@ -0,0 +1,40 @@
++'use strict';
++/**
++ * Regroup open clusters with today's rule (clusters.resplit): each cluster's reports are grouped on its core terms
++ * again, and every group but the first report's is split off as an audited, reversible split (actor svc:news).
++ * Clusters built before 2026-10-15 matched on the union of their reports' words and could take unrelated reports.
++ *
++ *   node scripts/resplit-clusters.js                 # dry run: what would split, nothing changes
++ *   node scripts/resplit-clusters.js --apply         # split
++ *   node scripts/resplit-clusters.js clu_… [--apply] # only these clusters
++ *
++ * The events (news.cluster.updated) wait in the outbox; the running service's relay sends them.
++ */
++require('dotenv').config();
++const { createServiceOutbox } = require('openvibe-sdk/events');
++const { load } = require('../server/config');
++const { openStore } = require('../server/db');
++const { createClusters } = require('../server/domain/clusters');
++
++const args = process.argv.slice(2);
++const apply = args.includes('--apply');
++const only = args.filter((a) => /^clu_/.test(a));
++const config = load();
++(async () => {
++    const store = await openStore(config);
++    const outbox = createServiceOutbox({ db: store.db, source: 'news', log: { log() {}, warn() {}, error: console.error }, now: store.now });
++    const clusters = createClusters({ store, config, outbox });
++    const ids = only.length ? only : (await clusters.recent({ limit: 1000 })).map((c) => c.id);
++    let split = 0;
++    for (const id of ids) {
++        const r = await clusters.resplit(null, id, { dryRun: !apply });
++        if (r.groups < 2) continue;
++        split += r.groups - 1;
++        const headline = async (itemId) => (await store.db.prepare('SELECT headline FROM news_source_items WHERE id = ?').get(itemId) || {}).headline;
++        console.log(`${id}: ${r.grouping.reduce((n, g) => n + g.length, 0)} reports → ${r.groups} groups${apply ? ` (split audits ${r.splits.map((a) => a.id).join(', ')})` : ''}`);
++        for (const g of r.grouping.slice(0, 12)) console.log(`   ${String(g.length).padStart(3)}  ${String(await headline(g[0])).slice(0, 100)}`);
++        if (r.grouping.length > 12) console.log(`   … ${r.grouping.length - 12} more groups`);
++    }
++    console.log(`${apply ? 'split off' : 'would split off'} ${split} group(s) from ${ids.length} open cluster(s)`);
++    await store.close();
++})().catch((err) => { console.error(`[resplit] ${err.message}`); process.exit(1); });
ddiff --git a/server/domain/clusters.js b/server/domain/clusters.js
iindex 522b36f..3fe4489 100644
---- a/server/domain/clusters.js
++++ b/server/domain/clusters.js
@@@ -4,7 +4,9 @@
   * Clusters: deterministic, explainable grouping of source items that report the same story.
   *
   *   assign(item)   an active item joins the open cluster it shares the most with inside the time
-- *                  window (NEWS_CLUSTER_WINDOW_HOURS, default 72 h):
++ *                  window (NEWS_CLUSTER_WINDOW_HOURS, default 72 h), matched on the cluster's core only: the
++ *                  entities and terms about a third of its reports carry (coreMin), so a big cluster cannot
++ *                  snowball on the union of everything it holds;
   *                    shared named entities count 2, shared terms count 1;
   *                    it joins when (≥ 1 shared entity AND ≥ 2 shared terms) OR ≥ 3 shared terms;
   *                    ties go to the oldest cluster. Otherwise it starts a new cluster.
@@@ -12,6 +14,8 @@
   *                  shared terms and score, so every membership can be explained and recomputed.
   *                  A duplicate joins its original's cluster (rule "duplicate_of").
   *   merge(a, b)    an editor merges b into a; split(a, items) moves items into a new cluster.
++ *   resplit(a)     regroups a cluster's reports with today's rule; every group but the first report's is split
++ *                  off (audited and reversible, like an editor's split).
   *                  Both are audited (news_cluster_audit, with exactly which items moved) and
   *                  reversible (reverse(auditId)); nothing is deleted.
   *
@@@ -37,15 +41,23 @@ function itemTime(item) {
      return Number.isFinite(p) ? p : item.first_seen_at;
  }
  
--/** Match an item's signature against a cluster's. → { sharedEntities, sharedTerms, score, joins } */
--function compare(sig, cluster) {
--    const ce = parse(cluster.entities, {});
--    const ct = parse(cluster.terms, {});
--    const sharedEntities = sig.entities.filter((e) => ce[e]);
--    const sharedTerms = sig.terms.filter((t) => ct[t]);
++/**
++ * How many of a cluster's n reports must carry an entity or term for it to be the story's own: about a third (one
++ * report of one or two, two of three to five, 66 of 196). Matching on the union instead let one cluster take 196
++ * unrelated reports from three outlets (2026-10-15: 1,950 terms, so nearly any report shared three).
++ */
++const coreMin = (n) => Math.ceil((Math.max(1, n) + 1) / 3);
++
++/** Match an item's signature against a cluster's core (n = its reports). → { sharedEntities, sharedTerms, score, joins, coreMin } */
++function compare(sig, cluster, n = 1) {
++    const ce = typeof cluster.entities === 'string' ? parse(cluster.entities, {}) : cluster.entities || {};
++    const ct = typeof cluster.terms === 'string' ? parse(cluster.terms, {}) : cluster.terms || {};
++    const min = coreMin(n);
++    const sharedEntities = sig.entities.filter((e) => (ce[e] || 0) >= min);
++    const sharedTerms = sig.terms.filter((t) => (ct[t] || 0) >= min);
      const score = 2 * sharedEntities.length + sharedTerms.length;
      const joins = (sharedEntities.length >= 1 && sharedTerms.length >= 2) || sharedTerms.length >= 3;
--    return { sharedEntities, sharedTerms, score, joins };
++    return { sharedEntities, sharedTerms, score, joins, coreMin: min };
  }
  
  function labelOf(entities, terms) {
@@@ -63,6 +75,7 @@ function createClusters({ store, config, outbox }) {
          insert: db.prepare(`INSERT INTO news_story_clusters (id, label, status, split_from, terms, entities, window_start, window_end, created_by, created_at, updated_at)
                              VALUES (@id, @label, 'open', @split_from, @terms, @entities, @window_start, @window_end, @created_by, @now, @now)`),
          members: db.prepare("SELECT * FROM news_source_items WHERE cluster_id = ? ORDER BY first_seen_at, seq"),
++        sizes: db.prepare("SELECT cluster_id, count(*) AS n FROM news_source_items WHERE cluster_id = ANY(?) AND status <> 'removed' GROUP BY cluster_id"),
          setItem: db.prepare('UPDATE news_source_items SET cluster_id = ?, cluster_reason = ?, updated_at = ? WHERE id = ?'),
          audit: db.prepare('SELECT * FROM news_cluster_audit WHERE id = ?'),
          auditFor: db.prepare('SELECT * FROM news_cluster_audit WHERE cluster_id = ? OR other_id = ? ORDER BY created_at, seq'),
@@@ -117,13 +130,15 @@ function createClusters({ store, config, outbox }) {
          const at = itemTime(item);
          const w = config.clustering.windowMs;
          let best = null;
--        for (const c of await q.open.all(at - w, at + w)) {
--            const m = compare(sig, c);
++        const open = await q.open.all(at - w, at + w);
++        const sizes = new Map(open.length ? (await q.sizes.all(open.map((c) => c.id))).map((r) => [r.cluster_id, Number(r.n)]) : []);
++        for (const c of open) {
++            const m = compare(sig, c, sizes.get(c.id) || 1);
              if (!m.joins) continue;
--            if (!best || m.score > best.m.score) best = { c, m };   // ORDER BY created_at, seq: the oldest wins a tie
++            if (!best || m.score > best.m.score) best = { c, m, n: sizes.get(c.id) || 1 };   // ORDER BY created_at, seq: the oldest wins a tie
          }
          if (best) {
--            const reason = { rule: 'shared_terms', cluster: best.c.id, shared_entities: best.m.sharedEntities, shared_terms: best.m.sharedTerms, score: best.m.score, window_hours: w / 3600000 };
++            const reason = { rule: 'shared_terms', cluster: best.c.id, shared_entities: best.m.sharedEntities, shared_terms: best.m.sharedTerms, score: best.m.score, core_min: best.m.coreMin, cluster_size: best.n, window_hours: w / 3600000 };
              await q.setItem.run(best.c.id, JSON.stringify(reason), store.now(), item.id);
              const cluster = await recompute(best.c.id);
              await emit(cluster, 'item_added', { item_id: item.id, rule: 'shared_terms', score: best.m.score });
@@@ -167,7 +182,7 @@ function createClusters({ store, config, outbox }) {
      }
  
      const api = {
--        signature, compare, itemTime, labelOf, assign, resolve, recompute,
++        signature, compare, coreMin, itemTime, labelOf, assign, resolve, recompute,
  
          get: async (id) => await q.get.get(String(id || '')) || null,
          members: async (id) => await q.members.all(id),
@@@ -236,6 +251,51 @@ function createClusters({ store, config, outbox }) {
              });
          },
  
++        /**
++         * Regroup one open cluster's reports with today's rule (coreMin), in the order they were published: each
++         * report joins the group whose core it shares the most with inside the window, a duplicate follows its
++         * original. The group holding the first report stays; every other group is split off as an editor's split
++         * would be (audited, reversible one by one). Removed reports stay where they are. dryRun answers the grouping
++         * and changes nothing. → { cluster, groups, splits, grouping: [[item id, …], …] }
++         */
++        async resplit(viewer, clusterId, { reason = null, dryRun = false } = {}) {
++            const from = await mustOpen(clusterId);
++            const w = config.clustering.windowMs;
++            const members = (await q.members.all(from.id)).filter((i) => i.status !== 'removed')
++                .sort((a, b) => (itemTime(a) - itemTime(b)) || (a.first_seen_at - b.first_seen_at) || (a.seq - b.seq));
++            const groups = [];
++            const add = (g, it) => {
++                const sig = signature(it);
++                for (const t of sig.terms) g.terms[t] = (g.terms[t] || 0) + 1;
++                for (const e of sig.entities) g.entities[e] = (g.entities[e] || 0) + 1;
++                const at = itemTime(it);
++                g.start = Math.min(g.start, at); g.end = Math.max(g.end, at);
++                g.ids.push(it.id);
++            };
++            for (const it of members) {
++                let into = it.status === 'duplicate' && it.duplicate_of ? groups.find((g) => g.ids.includes(it.duplicate_of)) : null;
++                if (!into) {
++                    const sig = signature(it);
++                    const at = itemTime(it);
++                    let best = null;
++                    for (const g of groups) {
++                        if (g.end < at - w || g.start > at + w) continue;
++                        const m = compare(sig, g, g.ids.length);
++                        if (m.joins && (!best || m.score > best.m.score)) best = { g, m };
++                    }
++                    into = best ? best.g : null;
++                }
++                if (!into) { into = { terms: {}, entities: {}, start: Infinity, end: -Infinity, ids: [] }; groups.push(into); }
++                add(into, it);
++            }
++            const grouping = groups.map((g) => g.ids);
++            if (dryRun) return { cluster: from, groups: groups.length, splits: [], grouping };
++            const splits = [];
++            const why = reason ? String(reason).slice(0, 500) : 'regrouped on the cluster\'s core terms';
++            for (const g of groups.slice(1)) splits.push((await api.split(viewer, from.id, g.ids, { reason: why })).audit);
++            return { cluster: await q.get.get(from.id), groups: groups.length, splits, grouping };
++        },
++
          /**
           * Undo a merge or a split: the items that moved (and are still where they were moved to) go
           * back; the absorbed cluster reopens / the split-off cluster is dissolved. Audited as well.
@@@ -287,4 +347,4 @@ function createClusters({ store, config, outbox }) {
      return api;
  }
  
--module.exports = { createClusters, signature, compare, itemTime, labelOf };
++module.exports = { createClusters, signature, compare, coreMin, itemTime, labelOf };
ddiff --git a/test/clusters.test.js b/test/clusters.test.js
iindex bf032fe..3ba7b1c 100644
---- a/test/clusters.test.js
++++ b/test/clusters.test.js
@@@ -111,6 +111,42 @@ const { boot, check, done, launchReports } = require('./helpers/boot');
          assert.deepStrictEqual(audit, ['merge', 'reverse_merge', 'split', 'reverse_split']);
      });
  
++    await check('a cluster is matched on its core, not the union of its reports: it cannot snowball', async () => {
++        const { compare, coreMin, signature } = require('../server/domain/clusters');
++        assert.deepStrictEqual([1, 2, 3, 5, 6, 196].map(coreMin), [1, 1, 2, 2, 3, 66]);
++        // Nine statements from one outlet: "statement", "ambassador" and "security" are in all nine; a railway report
++        // is in the union once. A new railway report shares three union terms but none of the core.
++        const terms = { statement: 9, ambassador: 9, security: 9, railway: 1, british: 1, scheme: 1, great: 1 };
++        const cluster = { terms: JSON.stringify(terms), entities: JSON.stringify({ un: 9, 'great british railways': 1 }) };
++        const railway = signature({ headline: 'Great British Railways confirms its apprenticeship scheme', summary: 'The railway scheme opens to young people.' });
++        assert.strictEqual(compare(railway, cluster, 1).joins, true, 'the union would take it');
++        const m = compare(railway, cluster, 9);
++        assert.strictEqual(m.joins, false, JSON.stringify(m));
++        assert.strictEqual(m.coreMin, 4);
++        const statement = signature({ headline: 'UN ambassador statement on security in Haiti', summary: 'The ambassador gave a statement to the security council.' });
++        assert.strictEqual(compare(statement, cluster, 9).joins, true, 'a report on the story itself still joins');
++    });
++
++    await check('resplit regroups a polluted cluster: the unrelated report splits off, audited and reversible', async () => {
++        const merged = await t.api(`/clusters/${main}/merge`, { json: { other: lone, reason: 'pollute it for the test' } });
++        assert.strictEqual(merged.status, 200, merged.text);
++        assert.strictEqual((await item(unrelated.id)).cluster_id, main);
++        const dry = await t.ctx.clusters.resplit(null, main, { dryRun: true });
++        assert.strictEqual(dry.groups, 2, JSON.stringify(dry.grouping));
++        assert.strictEqual((await item(unrelated.id)).cluster_id, main, 'a dry run changes nothing');
++        const r = await t.ctx.clusters.resplit(null, main);
++        assert.strictEqual(r.splits.length, 1);
++        assert.deepStrictEqual(r.splits[0].item_ids, [(await item(unrelated.id)).id], 'only the bakery report moves');
++        assert.strictEqual(r.splits[0].action, 'split');
++        assert.strictEqual(r.splits[0].actor, 'svc:news');
++        assert.notStrictEqual((await item(unrelated.id)).cluster_id, main);
++        assert.strictEqual((await item(reports.a.id)).cluster_id, main, 'the first report\'s group stays');
++        assert.strictEqual((await t.ctx.clusters.resplit(null, main, { dryRun: true })).groups, 1, 'a clean cluster stays whole');
++        const back = await t.api(`/clusters/audit/${r.splits[0].id}/reverse`, { json: { reason: 'undo the resplit' } });
++        assert.strictEqual(back.status, 200, back.text);
++        assert.strictEqual((await item(unrelated.id)).cluster_id, main);
++    });
++
      await t.close();
      done();
  })();