Pulse shows a stream once, not once per reconnect (#221)

ddiff --git a/server/pulse/consumer.js b/server/pulse/consumer.js
iindex ccc95e5..94c0f37 100644
---- a/server/pulse/consumer.js
++++ b/server/pulse/consumer.js
@@@ -189,7 +189,9 @@ function createPulseConsumer({ db, secrets = [], vipCache = null, revocations =
          const item = itemFor(event);
          if (typeof item === 'string') { stats.ignored++; return res.json({ event_id: event.event_id, duplicate: false, outcome: item }); }
          try {
--            const r = await inbox.once(CONSUMER, event.event_id, async () => ({ outcome: (await store.upsertItem(db, item)).created ? 'pulse:created' : 'pulse:updated' }));
++            // A reconnect is a new Live session and a new event; the stream itself is one broadcast.
++            const put = item.source_service === 'live' && item.source_type === 'stream' ? store.upsertLiveItem : store.upsertItem;
++            const r = await inbox.once(CONSUMER, event.event_id, async () => ({ outcome: (await put(db, item)).created ? 'pulse:created' : 'pulse:updated' }));
              if (r.duplicate) stats.duplicates++; else stats.applied++;
              return res.json({ event_id: event.event_id, duplicate: r.duplicate, outcome: r.duplicate ? null : r.result.outcome });
          } catch (err) {
ddiff --git a/server/pulse/store.js b/server/pulse/store.js
iindex 872f026..d51479d 100644
---- a/server/pulse/store.js
++++ b/server/pulse/store.js
@@@ -6,7 +6,52 @@
   * An item is keyed by its source (service, type, id). Re-recording the same source refreshes
   * its title and link but never its provenance: origin, actor and occurred_at are what the first
   * record said.
++ *
++ * A live stream is the exception: a dropped connection starts a new Live session (a new subject
++ * id), but it is the same broadcast. upsertLiveItem keeps one item for the same person (or the
++ * same channel URL) and title within six hours and moves its time to the newest start. Existing
++ * duplicates are collapsed the same way when the feed is read, so the page shows one line.
   */
++const LIVE_WINDOW_MS = 6 * 60 * 60 * 1000;
++
++/** A live.stream.started item: one row per broadcast, not per reconnect. → { item, created } */
++async function upsertLiveItem(db, it) {
++    const find = db.prepare('SELECT * FROM pulse_items WHERE source_service = ? AND source_type = ? AND source_id = ?');
++    return await db.tx(async () => {
++        const key = [it.source_service, it.source_type, String(it.source_id)];
++        const existing = await find.get(...key);
++        if (existing) {
++            await db.prepare('UPDATE pulse_items SET title = ?, url = ? WHERE source_service = ? AND source_type = ? AND source_id = ?').run(it.title, it.url, ...key);
++            return { item: await find.get(...key), created: false };
++        }
++        const at = Date.parse(String(it.occurred_at || ''));
++        const actor = it.actor_subject || null;
++        if (Number.isFinite(at)) {
++            // The same person (or channel URL) and title within six hours of this start is the same
++            // broadcast. Older sessions only: a delayed event must not move a newer item back in time.
++            const candidates = actor
++                ? await db.prepare(`SELECT * FROM pulse_items
++                        WHERE source_service = 'live' AND source_type = 'stream' AND title = ?
++                          AND ((actor_subject IS NOT NULL AND actor_subject = ?) OR url = ?)
++                        ORDER BY occurred_at DESC, id DESC`).all(it.title, actor, it.url)
++                : await db.prepare(`SELECT * FROM pulse_items
++                        WHERE source_service = 'live' AND source_type = 'stream' AND title = ? AND url = ?
++                        ORDER BY occurred_at DESC, id DESC`).all(it.title, it.url);
++            let prior = null, priorAt = -Infinity;
++            for (const c of candidates) {
++                const t = Date.parse(c.occurred_at);
++                if (Number.isFinite(t) && t <= at && at - t <= LIVE_WINDOW_MS && t > priorAt) { prior = c; priorAt = t; }
++            }
++            if (prior) {
++                if (at > priorAt) await db.prepare('UPDATE pulse_items SET occurred_at = ? WHERE id = ?').run(it.occurred_at, prior.id);
++                return { item: await db.prepare('SELECT * FROM pulse_items WHERE id = ?').get(prior.id), created: false };
++            }
++        }
++        await db.prepare(`INSERT INTO pulse_items (source_service, source_type, source_id, title, url, actor_subject, origin, visibility, occurred_at)
++                    VALUES (?, ?, ?, ?, ?, ?, ?, 'public', ?)`).run(...key, it.title, it.url, actor, it.origin, it.occurred_at);
++        return { item: await find.get(...key), created: true };
++    });
++}
  
  /** → { item, created } */
  async function upsertItem(db, it) {
@@@ -31,7 +76,9 @@ async function removeItem(db, service, type, id) {
   * Newest first, keyset-paginated by (occurred_at, id). Community's own items are re-checked
   * against their source at read time — a paste made private, a deleted post, a submission no longer
   * accepted or a thread in a space that stopped being public (or became members-only) never shows,
-- * even if nothing removed its item.
++ * even if nothing removed its item. Live streams recorded before upsertLiveItem existed are
++ * collapsed here too: an older session is hidden when the same person (or URL) and title started
++ * again within six hours, so one line and its newest time remain.
   *   opts.origin   'user' | 'ai' | 'system'
   *   opts.before   [occurred_at, id] of the last item already seen
   * → { rows, hasMore }
@@@ -41,6 +88,13 @@ async function listItems(db, { origin = null, before = null, limit = 30 } = {})
          SELECT i.* FROM pulse_items i
          WHERE (@origin::text IS NULL OR i.origin = @origin)
            AND (@at::text IS NULL OR i.occurred_at < @at OR (i.occurred_at = @at AND i.id < @id::bigint))
++          AND NOT (i.source_service = 'live' AND i.source_type = 'stream' AND EXISTS (
++                   SELECT 1 FROM pulse_items j
++                   WHERE j.source_service = 'live' AND j.source_type = 'stream'
++                     AND j.title = i.title
++                     AND ((j.actor_subject IS NOT NULL AND j.actor_subject = i.actor_subject) OR j.url = i.url)
++                     AND (j.occurred_at > i.occurred_at OR (j.occurred_at = i.occurred_at AND j.id > i.id))
++                     AND j.occurred_at::timestamptz <= i.occurred_at::timestamptz + INTERVAL '6 hours'))
            AND (i.source_service <> 'community'
                 OR (i.source_type = 'paste' AND EXISTS (SELECT 1 FROM pastes p WHERE p.slug = i.source_id AND p.deleted_at IS NULL AND p.visibility = 'public'))
                 OR (i.source_type = 'thread' AND EXISTS (SELECT 1 FROM threads t JOIN spaces s ON s.id = t.space_id
@@@ -57,4 +111,4 @@ async function listItems(db, { origin = null, before = null, limit = 30 } = {})
      return { rows, hasMore };
  }
  
--module.exports = { upsertItem, removeItem, listItems };
++module.exports = { upsertItem, upsertLiveItem, removeItem, listItems };
ddiff --git a/test/pulse-consumer.test.js b/test/pulse-consumer.test.js
iindex ea54544..49620f9 100644
---- a/test/pulse-consumer.test.js
++++ b/test/pulse-consumer.test.js
@@@ -86,6 +86,55 @@ let SECRET, db, app, vipDrops, USR, seq, evt;
      assert.strictEqual((await post(mod(20, { payload: { owner: { type: 'user', id: USR2 }, namespace: 'games.progress.summary', change: 'delete' } }))).json.outcome, 'ignored:change');
      assert.strictEqual((await post(mod(20, { payload: { owner: { type: 'user', id: USR2 }, namespace: 'games.progress.summary', change: 'updated', keys: ['level'] } }))).json.outcome, 'ignored:payload', 'no public level, nothing read');
      assert.strictEqual((await items()).length, 4);
++    // A dropped connection is a new Live session (a new subject id), but the stream is one broadcast.
++    const store = require('../server/pulse/store');
++    const T = Date.parse('2026-06-01T10:00:00.000Z');
++    const liveAt = (stream, at, title, { actor = USR, ...over } = {}) => evt({
++        subject: { type: 'stream', id: String(stream), revision: 1 },
++        payload: { stream_id: stream, channel: { username: 'goosely', display_name: 'Goosely', url: 'https://openvibe.live/@goosely', subject: { type: 'user', id: actor } }, title, started_at: new Date(at).toISOString() },
++        ...over,
++    });
++    const shows = async (title) => await db.prepare("SELECT id, source_id, occurred_at FROM pulse_items WHERE source_service = 'live' AND source_type = 'stream' AND title = ? ORDER BY id").all(title);
++    const REPEAT = 'Goosely went live: Repeat show';
++    r = await post(liveAt(9001, T, 'Repeat show'));
++    assert.strictEqual(r.json.outcome, 'pulse:created');
++    r = await post(liveAt(9002, T + 20 * 60_000, 'Repeat show'));
++    assert.strictEqual(r.json.outcome, 'pulse:updated', 'a reconnect within six hours is the same broadcast');
++    let shown = await shows(REPEAT);
++    assert.strictEqual(shown.length, 1, 'one item, not one per session');
++    assert.strictEqual(shown[0].source_id, '9001', 'the item keeps its id (and source)');
++    assert.strictEqual(shown[0].occurred_at, new Date(T + 20 * 60_000).toISOString(), 'its time is the later start');
++
++    r = await post(liveAt(9003, T + 30 * 60_000, 'New show'));
++    assert.strictEqual(r.json.outcome, 'pulse:created');
++    assert.strictEqual((await shows(REPEAT)).length, 1, 'the older broadcast is untouched');
++    assert.strictEqual((await shows('Goosely went live: New show')).length, 1, 'a different title is a new item');
++    assert.strictEqual((await items()).length, 6);
++
++    r = await post(liveAt(9004, T + 7 * 3_600_000, 'Repeat show'));
++    assert.strictEqual(r.json.outcome, 'pulse:created', 'more than six hours later is a new broadcast');
++    assert.strictEqual((await shows(REPEAT)).length, 2);
++    assert.strictEqual((await items()).length, 7);
++
++    // The channel URL is the identity too, when the event names another acting subject for it.
++    r = await post(liveAt(9005, T + 10 * 60_000, 'URL show'));
++    assert.strictEqual(r.json.outcome, 'pulse:created');
++    r = await post(liveAt(9006, T + 25 * 60_000, 'URL show', { actor: 'usr_01JAB2C3D4E5F6G7H8J9K0MNPS' }));
++    assert.strictEqual(r.json.outcome, 'pulse:updated');
++    assert.strictEqual((await shows('Goosely went live: URL show')).length, 1);
++    assert.strictEqual((await items()).length, 8);
++
++    // Duplicates already in the table: the feed read shows one line per channel+title per six hours,
++    // at the newest time, without deleting the rows.
++    const PRE = 'Goosely went live: Old repeat';
++    const pre = db.prepare(`INSERT INTO pulse_items (source_service, source_type, source_id, title, url, actor_subject, origin, occurred_at)
++                VALUES ('live', 'stream', ?, ?, 'https://openvibe.live/@old', ?, 'user', ?)`);
++    await pre.run('pre1', PRE, USR, '2026-06-02T10:00:00.000Z');
++    await pre.run('pre2', PRE, USR, '2026-06-02T10:20:00.000Z');
++    await pre.run('pre3', PRE, USR, '2026-06-02T17:00:00.000Z');
++    const feed = await store.listItems(db, { limit: 100 });
++    assert.deepStrictEqual(feed.rows.filter((x) => x.title === PRE).map((x) => x.source_id), ['pre3', 'pre2'], 'one line per six-hour window, at the newest time');
++    assert.strictEqual((await db.prepare('SELECT COUNT(*)::int AS n FROM pulse_items WHERE title = ?').get(PRE)).n, 3, 'the rows themselves stay');
      srv.close();
      console.log('pulse consumer: all checks passed');
      process.exit(0);