Ask 273: Vibes gain/loss API with cursor, read state, balance.changed event (#260)

ddiff --git a/docs/service-manifest-proposal.json b/docs/service-manifest-proposal.json
iindex 8c98832..3e69b42 100644
---- a/docs/service-manifest-proposal.json
++++ b/docs/service-manifest-proposal.json
@@@ -40,7 +40,8 @@
      "billing.receipt.external",
      "billing.credit.settled",
      "billing.vibes.charge.paid",
--    "billing.vibes.charge.refunded"
++    "billing.vibes.charge.refunded",
++    "billing.balance.changed"
    ],
    "eventsConsumed": [],
    "namespacesOwned": [
ddiff --git a/migrations/0018_balance_reads.sql b/migrations/0018_balance_reads.sql
nnew file mode 100644
iindex 0000000..f9529b7
---- /dev/null
++++ b/migrations/0018_balance_reads.sql
@@@ -0,0 +1,10 @@
++-- phase: expand
++-- Ask 273: the person's own read state for "what changed in my balances since I last looked".
++-- One row per person per currency: the transaction id cursor the bar has already shown.
++CREATE TABLE balance_reads (
++    subject text COLLATE "C" NOT NULL,
++    currency text COLLATE "C" NOT NULL CHECK (currency IN ('vibes')),
++    last_seen_txn text COLLATE "C" NOT NULL REFERENCES transactions (id),
++    updated_at text COLLATE "C" NOT NULL,
++    PRIMARY KEY (subject, currency)
++);
ddiff --git a/server/account-data.js b/server/account-data.js
iindex 460b28d..624c9e8 100644
---- a/server/account-data.js
++++ b/server/account-data.js
@@@ -51,6 +51,9 @@ const TABLES = [
      // charge they never paid has no payer and is not their data; the app's own settings are not people's records.
      { table: 'vibes_charges', subject: 'payer_subject', file: 'vibes-pay-charges.json', columns: ['id', 'app_id', 'project_id', 'amount', 'description', 'reference', 'status', 'sandbox', 'paid_at', 'refunded_at', 'created_at'], erase: { keep: ledger } },
      { table: 'vibes_payers', subject: 'subject', file: 'vibes-pay-payers.json', columns: ['app_id', 'payer_ref', 'created_at'], erase: { keep: ledger } },
++    // Ask 273: the "since I last looked" cursor is ephemeral UI state, not the person's data: no export
++    // file, erased with the account like staff sessions.
++    { table: 'balance_reads', subject: 'subject', file: null },
  ];
  
  /** Accounts with their balances (a balance is the account's cached sum of ledger entries). */
ddiff --git a/server/api/actor-limits.js b/server/api/actor-limits.js
iindex 1c133e4..07a9cb7 100644
---- a/server/api/actor-limits.js
++++ b/server/api/actor-limits.js
@@@ -35,6 +35,8 @@ const BUDGETS = {
      'billing.credits.topup.create': { minute: 5, hour: 30 },
      // Moving Balance to a project (POST /api/me/balance/transfers) is a wallet write, not a checkout.
      'billing.balance.transfer': { minute: 10, hour: 100 },
++    // Marking balances read (PUT /api/me/balances/read, ask 273) only stores a cursor: generous.
++    'billing.balance.read-mark': { minute: 60, hour: 1200 },
      // Getting Vibes with Balance (POST /api/me/vibes/from-balance): a few buys at a time.
      'billing.vibes.from_balance': { minute: 5, hour: 30 },
      // A tip, donation or paid interaction from a person's credit: one a second at most, above Tips' own
ddiff --git a/server/api/me.js b/server/api/me.js
iindex 3885151..e4601bc 100644
---- a/server/api/me.js
++++ b/server/api/me.js
@@@ -14,6 +14,12 @@
   *   GET /api/me/balance    the person's Balance plus the Balance of every project they administer
   *   GET /api/me/credits    the same answer (the old wire path stays an alias)
   *   GET /api/me/activity   the person's own Balance and Vibes movements, newest first (before=<txn id>)
++ *   GET /api/me/balances/changes?currency=vibes&since=<txn id>&limit=
++ *     → 200 { currency, balance, moves, parts: { earned, spent, received, sent }, net, cursor, ... }:
++ *           what changed in the person's Vibes since the cursor (or since they last looked:
++ *           an omitted since falls back to their stored read state), oldest first
++ *   PUT /api/me/balances/read { currency, cursor }
++ *     → 200 { currency, read_cursor }: store "I looked" (monotonic: looking backwards is refused)
   *   POST /api/me/balance/transfers  { to_project_id, amount_nano_usd, idempotency_key }
   *   POST /api/me/vibes/from-balance { vibes, idempotency_key }
   *
@@@ -31,6 +37,7 @@ const topups = require('../ops/topups');
  const balances = require('../ops/balance');
  const vibesPay = require('../ops/vibes');
  const activity = require('../ops/activity');
++const gainLoss = require('../ops/balance-changes');
  
  const HOST_RE = /^(?=.{1,253}$)([a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z]{2,63}$/;
  let origins = null;
@@@ -55,7 +62,7 @@ function meRouter({ ctx, config, keys, limits, adapters }) {
          const origin = req.get('origin');
          res.set({ 'Cache-Control': 'private, no-store', Vary: 'Origin' });
          if (origin && firstPartyOrigins().has(origin)) {
--            res.set({ 'Access-Control-Allow-Origin': origin, 'Access-Control-Allow-Headers': 'Authorization, Content-Type', 'Access-Control-Allow-Methods': 'GET, POST', 'Access-Control-Max-Age': '600' });
++            res.set({ 'Access-Control-Allow-Origin': origin, 'Access-Control-Allow-Headers': 'Authorization, Content-Type', 'Access-Control-Allow-Methods': 'GET, POST, PUT', 'Access-Control-Max-Age': '600' });
          }
          if (req.method === 'OPTIONS') return res.status(204).end();
          return next();
@@@ -120,6 +127,30 @@ function meRouter({ ctx, config, keys, limits, adapters }) {
          }
      });
  
++    // Ask 273: what changed in the person's balances since they last looked — the bar's gain or loss
++    // with its reason, per currency, with the cursor and the person's own read state. Their token,
++    // their accounts only: there is no subject parameter, so another person's balance is unreachable.
++    r.get('/balances/changes', person, limits.selfRead, async (req, res) => {
++        if (config.authority !== 'billing') return problem(res, 409, 'billing.not_authoritative', 'OpenVibe.Live keeps Vibes balances until the switch to OpenVibe.Billing');
++        try {
++            return res.json(await gainLoss.changes(ctx, req.person.subject, {
++                currency: req.query.currency || 'vibes', since: req.query.since, limit: req.query.limit,
++            }));
++        } catch (e) {
++            return failure(req, res, e, 'the balance changes are unavailable');
++        }
++    });
++
++    r.put('/balances/read', express.json({ limit: '16kb' }), person,
++        limits.budget('billing.balance.read-mark', (req) => req.person && req.person.subject), async (req, res) => {
++            const body = req.body || {};
++            try {
++                return res.json(await gainLoss.markRead(ctx, req.person.subject, { currency: body.currency || 'vibes', cursor: body.cursor }));
++            } catch (e) {
++                return failure(req, res, e, 'the read state is unavailable');
++            }
++        });
++
      // Move money from the person's Balance to a project they administer (D12): atomic, idempotent, never below zero.
      r.post('/balance/transfers', express.json({ limit: '16kb' }), person, body('billing.balance-transfer-request@1'),
          limits.budget('billing.balance.transfer', (req) => req.person && req.person.subject), async (req, res) => {
ddiff --git a/server/ledger.js b/server/ledger.js
iindex 0a84c9f..f7e616e 100644
---- a/server/ledger.js
++++ b/server/ledger.js
@@@ -23,6 +23,7 @@
   * the movement it guards are one atomic unit.
   */
  const { ids } = require('openvibe-contracts');
++const { enqueue } = require('./outbox');
  
  const NANO_USD = 'nano-usd';
  const isBigCurrency = (currency) => currency === NANO_USD;
@@@ -162,6 +163,32 @@ async function post(ctx, spec) {
              await insEntry.run(id, acct, m.amount);
              await bump.run(m.amount, nowIso, acct);
          }
++        // Ask 273: one subject-visible message per person whose Vibes moved, in the same transaction
++        // as the movement, so an open page updates without a reload. Non-Vibes and non-person entries
++        // (provider clearing, revenue, project accounts) announce nothing here.
++        if (ctx.outbox) {
++            const perUser = new Map();
++            for (const m of merged.values()) {
++                if (m.amount == 0 || m.currency !== 'vibes-bits') continue;
++                if (m.kind !== 'user_credit' && m.kind !== 'creator_payable') continue;
++                if (!ids.isSubjectId('user', m.owner)) continue;
++                perUser.set(m.owner, (perUser.get(m.owner) || 0) + m.amount);
++            }
++            for (const [owner, delta] of perUser) {
++                if (delta == 0) continue;
++                await enqueue(ctx, {
++                    event_type: 'billing.balance.changed',
++                    subject: { type: 'user', id: owner },
++                    visibility: 'subject',
++                    payload: {
++                        currency: 'vibes', delta,
++                        credit: await balance(db, { kind: 'user_credit', owner, currency: 'vibes-bits' }),
++                        payable: await balance(db, { kind: 'creator_payable', owner, currency: 'vibes-bits' }),
++                        txn: id, type: spec.type,
++                    },
++                });
++            }
++        }
          return { txn: await getTxn(db, id), replay: false };
      };
      return db.inTransaction() ? await run() : await money(db, run);
ddiff --git a/server/ops/balance-changes.js b/server/ops/balance-changes.js
nnew file mode 100644
iindex 0000000..0558f10
---- /dev/null
++++ b/server/ops/balance-changes.js
@@@ -0,0 +1,171 @@
++'use strict';
++
++/**
++ * Ask 273: "what changed in my balances since I last looked", per currency.
++ *
++ *   changes(ctx, subject, { currency, since, limit })
++ *     → { currency, balance, since, moves, parts, net, cursor, next_since, read_cursor }
++ *
++ * One move per journal transaction that touched the person's accounts of the currency, oldest
++ * first: { id, at, delta, part, label, detail, reason_key }. `parts` sums the window's moves into
++ * earned / spent / received / sent and `net` is their total — what the bar needs for a gain or a
++ * loss with its reason. `since` is a transaction id from an earlier answer; omitted, the person's
++ * own stored read state answers ("since I last looked"); with neither, the window starts at the
++ * person's first move. `cursor` is the newest move seen (or the head when the window is empty) and
++ * `next_since` pages the moves when they overflow the limit; parts and net always cover the whole
++ * window. markRead() stores the read state (monotonic: looking backwards is refused).
++ *
++ * Billing is the only money ledger: every number here is read from the journal, never a second ledger.
++ */
++const { fail, userSubject, A } = require('./common');
++const { balance } = require('../ledger');
++const { labelFor } = require('./activity');
++
++const DEFAULT_LIMIT = 100;
++const MAX_LIMIT = 500;
++const WINDOW_TXN_CAP = 5000;
++const TXN_RE = /^txn_[0-9A-HJKMNP-TV-Z]{26}$/;
++
++const TOUCH_VIBES = `((a.kind = 'user_credit' AND a.currency = 'vibes-bits' AND a.owner_subject = ?)
++    OR (a.kind = 'creator_payable' AND a.currency = 'vibes-bits' AND a.owner_subject = ?))`;
++
++/**
++ * Which part of the change a move is. Incoming creator earnings are earned; anything else arriving
++ * (bought Vibes, refunds back to credit) is received. Money leaving for another person or out of
++ * the system (a tip, a cashout, a refunded tip) is sent; money consumed (a subscription, an app or
++ * store purchase) is spent.
++ */
++function partOf({ net, type, kinds }) {
++    if (net > 0) return kinds.has('creator_payable') ? 'earned' : 'received';
++    if (type === 'donation' || type === 'cashout_request' || type === 'cashout_paid' || type === 'refund') return 'sent';
++    return 'spent';
++}
++
++async function txnById(db, id) {
++    if (!TXN_RE.test(String(id || ''))) fail(422, 'billing.invalid_cursor', 'since must be a transaction id from this list');
++    const row = await db.prepare('SELECT id, created_at FROM transactions WHERE id = ?').get(String(id));
++    if (!row) fail(422, 'billing.invalid_cursor', 'since names no transaction');
++    return row;
++}
++
++async function readCursor(db, subject, currency) {
++    const row = await db.prepare('SELECT last_seen_txn AS cursor FROM balance_reads WHERE subject = ? AND currency = ?').get(subject, currency);
++    return row ? row.cursor : null;
++}
++
++/** The person's own movements since a cursor, oldest first, with gain/loss parts over the window. */
++async function changes(ctx, rawSubject, { currency = 'vibes', since = null, limit = DEFAULT_LIMIT } = {}) {
++    const db = ctx.db;
++    const subject = userSubject(rawSubject, 'subject');
++    if (currency !== 'vibes') fail(422, 'billing.invalid_currency', 'currency is vibes');
++    const n = Number(limit);
++    const take = Number.isSafeInteger(n) && n > 0 ? Math.min(n, MAX_LIMIT) : DEFAULT_LIMIT;
++
++    let start = null;
++    if (since != null && since !== '') {
++        start = await txnById(db, since);
++    } else {
++        const stored = await readCursor(db, subject, currency);
++        if (stored) start = await txnById(db, stored);
++    }
++
++    let where = `WHERE ${TOUCH_VIBES}`;
++    const args = [subject, subject];
++    if (start) {
++        where += ' AND (t.created_at::timestamptz, t.id) > (?::timestamptz, ?)';
++        args.push(start.created_at, start.id);
++    }
++    const ids = await db.prepare(`SELECT DISTINCT t.id FROM transactions t
++        JOIN ledger_entries e ON e.txn_id = t.id JOIN accounts a ON a.id = e.account_id
++        ${where} ORDER BY t.id ASC LIMIT ?`).all(...args, WINDOW_TXN_CAP + 1);
++    const truncated = ids.length > WINDOW_TXN_CAP;
++    const windowIds = (truncated ? ids.slice(0, WINDOW_TXN_CAP) : ids).map((r) => r.id);
++    if (!windowIds.length) {
++        const head = await db.prepare(`SELECT t.id FROM transactions t
++            JOIN ledger_entries e ON e.txn_id = t.id JOIN accounts a ON a.id = e.account_id
++            WHERE ${TOUCH_VIBES} ORDER BY t.created_at DESC, t.id DESC LIMIT 1`).get(subject, subject);
++        return {
++            currency, balance: await current(ctx, subject),
++            since: start ? start.id : null, moves: [],
++            parts: { earned: 0, spent: 0, received: 0, sent: 0 }, net: 0,
++            cursor: head ? head.id : null, next_since: null, read_cursor: await readCursor(db, subject, currency),
++        };
++    }
++
++    const holes = windowIds.map(() => '?').join(',');
++    const rows = await db.prepare(`SELECT t.id, t.type, t.created_at, t.metadata, t.to_subject,
++            a.kind AS account_kind, e.amount::text AS amount
++        FROM transactions t JOIN ledger_entries e ON e.txn_id = t.id JOIN accounts a ON a.id = e.account_id
++        WHERE t.id IN (${holes}) AND ${TOUCH_VIBES}
++        ORDER BY t.created_at ASC, t.id ASC, a.kind`).all(...windowIds, subject, subject);
++
++    const donationTargets = [...new Set(rows.filter((r) => r.type === 'donation' && r.to_subject).map((r) => r.to_subject))];
++    const names = new Map();
++    if (donationTargets.length && ctx.network && typeof ctx.network.profiles === 'function') {
++        try { for (const [k, v] of await ctx.network.profiles(donationTargets)) names.set(k, v); } catch { /* fall back to "a creator" */ }
++    }
++
++    const byTxn = new Map();
++    for (const r of rows) {
++        const cur = byTxn.get(r.id) || { txn: r, vibes: 0, kinds: new Set() };
++        cur.vibes += Number(r.amount);
++        cur.kinds.add(r.account_kind);
++        byTxn.set(r.id, cur);
++    }
++
++    const parts = { earned: 0, spent: 0, received: 0, sent: 0 };
++    const moves = [];
++    for (const id of windowIds) {
++        const cur = byTxn.get(id);
++        if (!cur || cur.vibes === 0) continue;
++        const creator = names.get(cur.txn.to_subject);
++        const d = labelFor(cur.txn, { settlement: undefined, projectName: null, creatorName: creator && creator.username ? creator.username : null });
++        const part = partOf({ net: cur.vibes, type: cur.txn.type, kinds: cur.kinds });
++        parts[part] += cur.vibes;
++        moves.push({
++            id, at: cur.txn.created_at, delta: cur.vibes, part,
++            label: d.label, detail: d.detail, subject: d.subject, reason_key: d.reason_key,
++        });
++    }
++    const net = parts.earned + parts.spent + parts.received + parts.sent;
++    const page = moves.slice(0, take);
++    return {
++        currency,
++        balance: await current(ctx, subject),
++        since: start ? start.id : null,
++        moves: page,
++        parts, net,
++        cursor: moves.length ? moves[moves.length - 1].id : (start ? start.id : null),
++        next_since: moves.length > take ? page[page.length - 1].id : null,
++        read_cursor: await readCursor(db, subject, currency),
++        ...(truncated ? { truncated: true } : {}),
++    };
++}
++
++async function current(ctx, subject) {
++    return {
++        credit: await balance(ctx.db, A.credit(subject)),
++        payable: await balance(ctx.db, A.payable(subject)),
++    };
++}
++
++/** Store "I looked": the cursor must name a real transaction and may only move forward. */
++async function markRead(ctx, rawSubject, { currency = 'vibes', cursor } = {}) {
++    const db = ctx.db;
++    const subject = userSubject(rawSubject, 'subject');
++    if (currency !== 'vibes') fail(422, 'billing.invalid_currency', 'currency is vibes');
++    const at = await txnById(db, cursor);
++    const stored = await db.prepare('SELECT last_seen_txn AS cursor FROM balance_reads WHERE subject = ? AND currency = ?').get(subject, currency);
++    if (stored) {
++        const prev = await txnById(db, stored.cursor);
++        const back = at.created_at < prev.created_at || (at.created_at === prev.created_at && at.id < prev.id);
++        if (back) fail(409, 'billing.stale_read', 'the read cursor only moves forward');
++    }
++    const now = new Date(ctx.now()).toISOString();
++    await db.prepare(`INSERT INTO balance_reads (subject, currency, last_seen_txn, updated_at) VALUES (?, ?, ?, ?)
++        ON CONFLICT (subject, currency) DO UPDATE SET last_seen_txn = EXCLUDED.last_seen_txn, updated_at = EXCLUDED.updated_at`)
++        .run(subject, currency, at.id, now);
++    return { currency, read_cursor: at.id };
++}
++
++module.exports = { changes, markRead };
ddiff --git a/server/outbox.js b/server/outbox.js
iindex 9301730..d0957b2 100644
---- a/server/outbox.js
++++ b/server/outbox.js
@@@ -9,8 +9,13 @@ const { ids, validate } = require('openvibe-contracts');
  
  const ACTOR = { type: 'service', id: 'billing' };
  
--/** actor defaults to the service itself; staff actions pass the person ({ type: 'user', id }). */
--async function enqueue(ctx, { event_type, subject, payload, priority = 'important', traceId, actor }) {
++/**
++ * actor defaults to the service itself; staff actions pass the person ({ type: 'user', id }).
++ * visibility is 'internal' (service principals only) unless the caller names 'subject': then only
++ * the user the subject names sees it over Events realtime — how an open page learns its own
++ * balance changed without a reload (ask 273: billing.balance.changed).
++ */
++async function enqueue(ctx, { event_type, subject, payload, priority = 'important', traceId, actor, visibility = 'internal' }) {
      const ms = ctx.now();
      const env = {
          event_id: ids.newId('event', ms),
@@@ -20,7 +25,7 @@ async function enqueue(ctx, { event_type, subject, payload, priority = 'importan
          actor: actor || ACTOR,
          timestamp: new Date(ms).toISOString(),
          priority,
--        visibility: 'internal',
++        visibility,
          subject,
          payload: payload || {},
      };
ddiff --git a/test/auth-outbox.test.js b/test/auth-outbox.test.js
iindex 1740222..6f4fec3 100644
---- a/test/auth-outbox.test.js
++++ b/test/auth-outbox.test.js
@@@ -50,7 +50,8 @@ const { startEvents } = require('./helpers/stubs');
              const v = validate('events.event-envelope@1', e);
              assert.ok(v.valid, JSON.stringify(v.errors));
              assert.deepStrictEqual([e.source, e.actor.type, e.actor.id], ['billing', 'service', 'billing']);
--            assert.ok(['transaction', 'entitlement', 'cashout', 'subscription'].includes(e.subject.type));
++            assert.ok(['transaction', 'entitlement', 'cashout', 'subscription', 'user'].includes(e.subject.type),
++                `subject type ${e.subject.type} of ${e.event_type} is one the outbox sends`);
          }
          const txnCount = (await t.db.prepare('SELECT COUNT(*) AS n FROM transactions').get()).n;
          const refused = await t.call('POST', '/api/v1/transfers', { body: { from: a, to: b, amount: 999999 } });
ddiff --git a/test/me-balances-changes.test.js b/test/me-balances-changes.test.js
nnew file mode 100644
iindex 0000000..d43f57a
---- /dev/null
++++ b/test/me-balances-changes.test.js
@@@ -0,0 +1,122 @@
++'use strict';
++/**
++ * Ask 273, Billing part: GET /api/me/balances/changes (a gain, a loss, a cursor, another person's
++ * balance refused) and PUT /api/me/balances/read (the person's own read state), plus the
++ * billing.balance.changed subject-visible message a balance move emits for an open page.
++ */
++const assert = require('assert');
++const crypto = require('crypto');
++const { boot, fund, check, done } = require('./helpers/app');
++const transfers = require('../server/ops/transfers');
++
++const key = (p) => `${p}:${crypto.randomBytes(8).toString('hex')}`;
++
++(async () => {
++    console.log('balances changes');
++    const t = await boot({ env: { BILLING_AUTHORITY: 'billing' } });
++    const bob = t.user(102, 'bob');
++    const ann = t.user(101, 'ann');
++    const bobToken = t.network.signUser({ subject_id: bob.id, username: 'bob' });
++    const annToken = t.network.signUser({ subject_id: ann.id, username: 'ann' });
++
++    const get = async (token, q = '') => {
++        const r = await fetch(`${t.base}/api/me/balances/changes${q}`, { headers: token ? { Authorization: `Bearer ${token}` } : {} });
++        return { status: r.status, body: await r.json() };
++    };
++    const putRead = async (token, body) => {
++        const r = await fetch(`${t.base}/api/me/balances/read`, {
++            method: 'PUT', headers: { ...(token ? { Authorization: `Bearer ${token}` } : {}), 'Content-Type': 'application/json' },
++            body: JSON.stringify(body),
++        });
++        return { status: r.status, body: await r.json() };
++    };
++
++    const funded = await fund(t, bob.id, 500);
++    const tip = await transfers.create(t.ctx, {
++        from: bob.id, to: ann.id, amount: 100, idempotencyKey: key('tip'), actor: { principal: `user:${bob.id}` },
++    });
++
++    await check('a gain: bought Vibes arrive received with their reason', async () => {
++        const { status, body } = await get(bobToken);
++        assert.strictEqual(status, 200);
++        assert.strictEqual(body.currency, 'vibes');
++        assert.deepStrictEqual(body.balance, { credit: 400, payable: 0 });
++        assert.strictEqual(body.moves.length, 2);
++        assert.strictEqual(body.moves[0].id, funded.id);
++        assert.strictEqual(body.moves[0].delta, 500);
++        assert.strictEqual(body.moves[0].part, 'received');
++        assert.ok(body.moves[0].label, 'a plain-words reason rides along');
++        assert.ok(body.moves[0].reason_key, 'and the ledger key for support');
++        assert.deepStrictEqual(body.parts, { earned: 0, spent: 0, received: 500, sent: -100 });
++        assert.strictEqual(body.net, 400);
++        assert.strictEqual(body.cursor, tip.txn.id);
++        assert.strictEqual(body.next_since, null);
++    });
++
++    await check("a loss: the tip leaves as sent, and the creator's side is earned", async () => {
++        const { body } = await get(bobToken);
++        assert.strictEqual(body.moves[1].delta, -100);
++        assert.strictEqual(body.moves[1].part, 'sent');
++        const annChanges = (await get(annToken)).body;
++        assert.strictEqual(annChanges.moves.length, 1);
++        assert.strictEqual(annChanges.moves[0].delta, 100);
++        assert.strictEqual(annChanges.moves[0].part, 'earned');
++        assert.deepStrictEqual(annChanges.parts, { earned: 100, spent: 0, received: 0, sent: 0 });
++        assert.deepStrictEqual(annChanges.balance, { credit: 0, payable: 100 });
++    });
++
++    await check('a cursor: since replays only what came after', async () => {
++        const { status, body } = await get(bobToken, `?since=${funded.id}`);
++        assert.strictEqual(status, 200);
++        assert.strictEqual(body.since, funded.id);
++        assert.strictEqual(body.moves.length, 1);
++        assert.strictEqual(body.moves[0].id, tip.txn.id);
++        assert.deepStrictEqual(body.parts, { earned: 0, spent: 0, received: 0, sent: -100 });
++        assert.strictEqual(body.net, -100);
++        assert.strictEqual((await get(bobToken, '?since=nope')).status, 422, 'a made-up cursor is refused');
++    });
++
++    await check("another person's balance is refused: no subject parameter, tokens only open their own", async () => {
++        assert.strictEqual((await get(null)).status, 401);
++        assert.strictEqual((await get(t.network.signService({ cap: ['billing.*'] }))).status, 401, 'a service token is not a person');
++        assert.strictEqual((await get(`${bobToken.slice(0, -4)}AAAA`)).status, 401);
++        const annSeenByBob = (await get(bobToken)).body.moves.filter((m) => m.delta === 100 && m.part === 'earned');
++        assert.strictEqual(annSeenByBob.length, 0, 'bob never sees ann’s earned move');
++        assert.strictEqual((await get(bobToken, '?currency=coins')).status, 422);
++    });
++
++    await check('read state: marking read empties "since I last looked", and it only moves forward', async () => {
++        const marked = await putRead(bobToken, { currency: 'vibes', cursor: tip.txn.id });
++        assert.strictEqual(marked.status, 200);
++        assert.strictEqual(marked.body.read_cursor, tip.txn.id);
++        const after = await get(bobToken);
++        assert.strictEqual(after.body.moves.length, 0, 'nothing since the look');
++        assert.strictEqual(after.body.cursor, tip.txn.id);
++        assert.strictEqual(after.body.read_cursor, tip.txn.id);
++        assert.strictEqual((await putRead(bobToken, { currency: 'vibes', cursor: funded.id })).status, 409, 'looking backwards is refused');
++        assert.strictEqual((await putRead(bobToken, { currency: 'vibes', cursor: 'txn_00000000000000000000000000' })).status, 422);
++        assert.strictEqual((await putRead(null, { currency: 'vibes', cursor: tip.txn.id })).status, 401);
++    });
++
++    await check('a balance move emits billing.balance.changed, subject-visible, per person', async () => {
++        const rows = await t.db.prepare(`SELECT envelope AS event FROM service_outbox
++            WHERE (envelope #>> '{event_type}') = 'billing.balance.changed' ORDER BY id`).all();
++        const events = rows.map((r) => (typeof r.event === 'string' ? JSON.parse(r.event) : r.event));
++        assert.strictEqual(events.length, 3, `fund + tip tip both sides, got ${events.length}`);
++        for (const e of events) {
++            assert.strictEqual(e.visibility, 'subject', 'only the person sees their own over realtime');
++            assert.strictEqual(e.subject.type, 'user');
++            assert.strictEqual(e.payload.currency, 'vibes');
++        }
++        const bobEvents = events.filter((e) => e.subject.id === bob.id);
++        assert.deepStrictEqual(bobEvents.map((e) => e.payload.delta), [500, -100], 'one message per move, in order');
++        assert.deepStrictEqual(
++            { credit: bobEvents[1].payload.credit, payable: bobEvents[1].payload.payable }, { credit: 400, payable: 0 });
++        const annEvents = events.filter((e) => e.subject.id === ann.id);
++        assert.strictEqual(annEvents.length, 1);
++        assert.strictEqual(annEvents[0].payload.delta, 100);
++    });
++
++    await t.close();
++    done();
++})();