feat(relay): monotonic seq cursor + timing-safe token compare

docs/groupmind-local.md

@@ -31,7 +31,13 @@ If `IAK_RELAY_TOKEN` is set, every request must present it via `X-API-Key` or
 | `GET` | `/health` | — | always open; reports message count |
 | `POST` | `/api/v1/messages` | `{ room, body, from?, metadata? }` | create a message |
 | `POST` | `/api/v1/rooms/:room/messages` | `{ body, from?, metadata? }` | create (room in path) |
-| `GET` | `/api/v1/rooms/:room/messages` | `?limit=&since=` | newest-first, GroupMind shape |
+| `GET` | `/api/v1/rooms/:room/messages` | `?limit=&since=&after=` | see below |
+
+`GET` returns the recent view (**newest-first**) by default. Pass `after=<seq>`
+for an incremental **cursor feed**: messages with `seq > after`, **oldest-first**,
+capped by `limit` — the client advances its cursor and re-fetches, so nothing is
+skipped even when messages share a millisecond timestamp. Every message carries a
+monotonic `seq` (the exact cursor) alongside its millisecond `created_at`.
 
 An agent points its room client at `http://<mini-lan-ip>:18790/api/v1` when cloud
 GroupMind is unreachable — same paths, just a different base URL.

scripts/local-relay.mjs

@@ -35,7 +35,7 @@
 
 import http from 'node:http';
 import { appendFileSync, mkdirSync, readFileSync, existsSync } from 'node:fs';
-import { randomUUID } from 'node:crypto';
+import { randomUUID, timingSafeEqual, createHash } from 'node:crypto';
 import { homedir } from 'node:os';
 import { dirname, join } from 'node:path';
 
@@ -52,6 +52,10 @@ function readMessages(storePath) {
     if (!line.trim()) continue;
     try { out.push(JSON.parse(line)); } catch { /* skip a corrupt line, never crash a read */ }
   }
+  // `seq` is the 1-based append position, assigned HERE on read so it is present
+  // and correct even for rows written before seq existed (e.g. a store from the
+  // #37 MVP). This is the authoritative cursor value; any stored seq is ignored.
+  out.forEach((m, i) => { m.seq = i + 1; });
   return out;
 }
 
@@ -60,6 +64,14 @@ function appendMessage(storePath, msg) {
   appendFileSync(storePath, JSON.stringify(msg) + '\n');
 }
 
+// Constant-time token compare. Hash both sides to a fixed 32 bytes first so
+// timingSafeEqual never sees a length mismatch (and length isn't leaked).
+function safeTokenEqual(provided, expected) {
+  const a = createHash('sha256').update(String(provided ?? '')).digest();
+  const b = createHash('sha256').update(String(expected ?? '')).digest();
+  return timingSafeEqual(a, b);
+}
+
 function send(res, status, obj) {
   const payload = JSON.stringify(obj);
   res.writeHead(status, { 'Content-Type': 'application/json', 'Content-Length': Buffer.byteLength(payload) });
@@ -115,10 +127,18 @@ export function startRelay(opts = {}) {
 
   if (!token) log('WARNING: no IAK_RELAY_TOKEN set — relay is LAN-open (fine on a trusted home LAN only).');
 
+  // `seq` = 1-based append position (see readMessages). `created_at` is
+  // millisecond-resolution so it can't cursor exactly (a same-ms checkpoint
+  // would skip messages); `seq` is the precise cursor the failover + sync
+  // layers rely on. `count` is seeded from the (possibly pre-seq) store so old
+  // rows are counted, and it tracks the next position across appends in-process.
+  let count = readMessages(storePath).length;
+  const commit = (msg) => { msg.seq = ++count; appendMessage(storePath, msg); return msg; };
+
   function authorized(req) {
     if (!token) return true;
     const bearer = (req.headers['authorization'] || '').replace(/^Bearer\s+/i, '');
-    return req.headers['x-api-key'] === token || bearer === token;
+    return safeTokenEqual(req.headers['x-api-key'], token) || safeTokenEqual(bearer, token);
   }
 
   const server = http.createServer(async (req, res) => {
@@ -138,12 +158,24 @@ export function startRelay(opts = {}) {
         const room = decodeURIComponent(getMatch[1]);
         const limit = Math.min(Math.max(1, Number(url.searchParams.get('limit')) || DEFAULT_LIMIT), MAX_LIMIT);
         const since = url.searchParams.get('since');
+        const after = url.searchParams.get('after'); // exact seq cursor (authoritative)
+        const cursoring = after !== null && after !== '';
+        // readMessages returns JSONL append order = chronological, each with its
+        // positional seq.
         let msgs = readMessages(storePath).filter((m) => m.room === room);
-        if (since) msgs = msgs.filter((m) => m.created_at > since);
-        // JSONL append order IS chronological, so it orders even messages that
-        // share a millisecond timestamp. Take the last `limit` (newest) and
-        // reverse to newest-first, matching GroupMind's shape.
-        const messages = msgs.slice(-limit).reverse();
+        if (cursoring) {
+          // `after` (seq) is authoritative — never ALSO narrow by the
+          // millisecond `since`, which would re-drop the same-ms messages the
+          // cursor is meant to return (PR #38 review).
+          msgs = msgs.filter((m) => m.seq > Number(after));
+        } else if (since) {
+          msgs = msgs.filter((m) => m.created_at > since);
+        }
+        // Cursor feed (`after`): chronological forward from the cursor, capped —
+        // the client advances its cursor and re-fetches, so nothing is skipped
+        // even for same-millisecond messages. Recent view (no cursor):
+        // newest-first, matching GroupMind's shape.
+        const messages = cursoring ? msgs.slice(0, limit) : msgs.slice(-limit).reverse();
         return send(res, 200, { messages, count: messages.length });
       }
 
@@ -152,8 +184,7 @@ export function startRelay(opts = {}) {
         const b = await readJsonBody(req);
         const { msg, error } = buildMessage({ room: b.room, body: b.body, from: b.from, metadata: b.metadata });
         if (error) return send(res, 400, { error });
-        appendMessage(storePath, msg);
-        return send(res, 201, msg);
+        return send(res, 201, commit(msg));
       }
 
       // POST /api/v1/rooms/:room/messages  { body, ... }
@@ -163,8 +194,7 @@ export function startRelay(opts = {}) {
         const b = await readJsonBody(req);
         const { msg, error } = buildMessage({ room, body: b.body, from: b.from, metadata: b.metadata });
         if (error) return send(res, 400, { error });
-        appendMessage(storePath, msg);
-        return send(res, 201, msg);
+        return send(res, 201, commit(msg));
       }
 
       return send(res, 404, { error: 'not found' });

test/local-relay.test.mjs

@@ -3,7 +3,7 @@ import test from 'node:test';
 import { once } from 'node:events';
 import { tmpdir } from 'node:os';
 import { join } from 'node:path';
-import { rmSync } from 'node:fs';
+import { rmSync, writeFileSync } from 'node:fs';
 import { randomUUID } from 'node:crypto';
 import { startRelay } from '../scripts/local-relay.mjs';
 
@@ -126,3 +126,84 @@ test('unknown route is 404', async () => {
     assert.equal(r.status, 404);
   });
 });
+
+test('messages get a monotonic seq; after-cursor is a no-skip chronological feed', async () => {
+  await withRelay({}, async ({ base, headers }) => {
+    const posted = [];
+    for (const n of [1, 2, 3, 4]) {
+      const r = await fetch(`${base}/api/v1/messages`, {
+        method: 'POST', headers: j(headers), body: JSON.stringify({ room: 'c', body: `m${n}` }),
+      });
+      posted.push(await r.json());
+    }
+    // seq strictly increasing even when created_at collides in the same ms
+    for (let i = 1; i < posted.length; i++) assert.ok(posted[i].seq > posted[i - 1].seq, 'seq increases');
+
+    // cursor from 0: all four, oldest-first
+    const all = await (await fetch(`${base}/api/v1/rooms/c/messages?after=0`, { headers })).json();
+    assert.deepEqual(all.messages.map((m) => m.body), ['m1', 'm2', 'm3', 'm4']);
+
+    // cursor from m2's seq: exactly m3, m4 — the same-ms skip that `since` alone would cause
+    const after2 = await (await fetch(`${base}/api/v1/rooms/c/messages?after=${posted[1].seq}`, { headers })).json();
+    assert.deepEqual(after2.messages.map((m) => m.body), ['m3', 'm4']);
+
+    // cursor from the last seq: caught up, empty
+    const afterLast = await (await fetch(`${base}/api/v1/rooms/c/messages?after=${posted[3].seq}`, { headers })).json();
+    assert.equal(afterLast.count, 0);
+  });
+});
+
+test('after-cursor honors limit and stays chronological (forward feed pages)', async () => {
+  await withRelay({}, async ({ base, headers }) => {
+    for (const n of [1, 2, 3, 4, 5]) {
+      await fetch(`${base}/api/v1/messages`, {
+        method: 'POST', headers: j(headers), body: JSON.stringify({ room: 'c', body: `m${n}` }),
+      });
+    }
+    const page = await (await fetch(`${base}/api/v1/rooms/c/messages?after=0&limit=2`, { headers })).json();
+    assert.deepEqual(page.messages.map((m) => m.body), ['m1', 'm2'], 'oldest-first page from the cursor');
+  });
+});
+
+test('after is authoritative over since (since must not re-drop same-ms messages)', async () => {
+  await withRelay({}, async ({ base, headers }) => {
+    for (const n of [1, 2, 3]) {
+      await fetch(`${base}/api/v1/messages`, {
+        method: 'POST', headers: j(headers), body: JSON.stringify({ room: 'c', body: `m${n}` }),
+      });
+    }
+    // A future `since` would exclude everything on its own; with `after=0`
+    // present it must be ignored, so all three still come back.
+    const future = new Date(Date.now() + 60_000).toISOString();
+    const res = await (await fetch(
+      `${base}/api/v1/rooms/c/messages?after=0&since=${encodeURIComponent(future)}`, { headers })).json();
+    assert.deepEqual(res.messages.map((m) => m.body), ['m1', 'm2', 'm3'], 'after wins; since ignored');
+  });
+});
+
+test('rows written before seq existed still get a positional seq (upgrade compat)', async () => {
+  const storePath = join(tmpdir(), `iak-relay-seed-${randomUUID()}.jsonl`);
+  // Simulate a #37-era store: rows with NO seq field.
+  const row = (id, body) => JSON.stringify({
+    id, room: 'r', from: '@x', body, created_at: '2026-01-01T00:00:00.000Z', origin: 'local-relay',
+  });
+  writeFileSync(storePath, `${row('a', 'old1')}\n${row('b', 'old2')}\n`);
+  const server = startRelay({ port: 0, storePath, logger: () => {} });
+  await once(server, 'listening');
+  const base = `http://127.0.0.1:${server.address().port}`;
+  try {
+    // after=0 must include BOTH pre-seq rows (not silently skip them)
+    const feed = await (await fetch(`${base}/api/v1/rooms/r/messages?after=0`)).json();
+    assert.deepEqual(feed.messages.map((m) => m.body), ['old1', 'old2']);
+    assert.deepEqual(feed.messages.map((m) => m.seq), [1, 2], 'positional seq assigned to legacy rows');
+    // a new POST continues the sequence past the seeded rows
+    const posted = await (await fetch(`${base}/api/v1/messages`, {
+      method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ room: 'r', body: 'new' }),
+    })).json();
+    assert.equal(posted.seq, 3, 'new message continues after the seeded rows');
+  } finally {
+    server.close();
+    await once(server, 'close');
+    rmSync(storePath, { force: true });
+  }
+});