feat(comms): P1 presence — minimal Synapse + fleet presence room + mosaic.presence heartbeat + liveness (#888)

Co-authored-by: jason.woltje <jason@diversecanvas.com>
Co-committed-by: jason.woltje <jason@diversecanvas.com>

packages/comms/src/__tests__/heartbeat.test.ts

@@ -0,0 +1,90 @@
+import { afterEach, describe, expect, it, vi } from 'vitest';
+
+import { HeartbeatEmitter, startHeartbeatLoop } from '../heartbeat.js';
+import type { PresenceHeartbeatContent } from '../types.js';
+
+const agent = { mxid: '@agent-alpha:matrix.localhost', slug: 'alpha', harness: 'claude-code' };
+
+describe('HeartbeatEmitter', () => {
+  it('increments seq starting at 1 and stamps the envelope', () => {
+    let t = 1000;
+    const em = new HeartbeatEmitter({ agent, intervalMs: 5000, now: () => t });
+    const a = em.next();
+    t = 6000;
+    const b = em.next('away');
+
+    expect(a.seq).toBe(1);
+    expect(a.ts).toBe(1000);
+    expect(a.status).toBe('online');
+    expect(a.macp_type).toBe('presence');
+    expect(a.msgtype).toBe('mosaic.presence');
+    expect(a.macp_version).toBe('1.0');
+    expect(a.interval_ms).toBe(5000);
+    expect(a.agent).toEqual(agent);
+    expect(a.body).toContain('alpha');
+
+    expect(b.seq).toBe(2);
+    expect(b.ts).toBe(6000);
+    expect(b.status).toBe('away');
+    expect(em.currentSeq).toBe(2);
+  });
+
+  it('includes mission_id only when provided', () => {
+    const withMission = new HeartbeatEmitter({
+      agent,
+      intervalMs: 1000,
+      missionId: 'KBN-101',
+    }).next();
+    const without = new HeartbeatEmitter({ agent, intervalMs: 1000 }).next();
+    expect(withMission.mission_id).toBe('KBN-101');
+    expect(without.mission_id).toBeUndefined();
+  });
+});
+
+describe('startHeartbeatLoop', () => {
+  afterEach(() => vi.useRealTimers());
+
+  it('emits immediately, then once per interval, until stopped', () => {
+    vi.useFakeTimers();
+    const sent: PresenceHeartbeatContent[] = [];
+    const em = new HeartbeatEmitter({ agent, intervalMs: 1000, now: () => Date.now() });
+    const loop = startHeartbeatLoop({
+      emitter: em,
+      intervalMs: 1000,
+      send: (c) => {
+        sent.push(c);
+      },
+    });
+
+    expect(sent).toHaveLength(1); // immediate beat
+    vi.advanceTimersByTime(3000);
+    expect(sent).toHaveLength(4); // +3 beats
+    expect(sent.map((s) => s.seq)).toEqual([1, 2, 3, 4]);
+
+    loop.stop();
+    vi.advanceTimersByTime(5000);
+    expect(sent).toHaveLength(4); // no more after stop
+    loop.stop(); // idempotent
+  });
+
+  it('routes a rejected async send to onError without killing the loop', async () => {
+    vi.useFakeTimers();
+    const onError = vi.fn();
+    let n = 0;
+    const em = new HeartbeatEmitter({ agent, intervalMs: 1000 });
+    const loop = startHeartbeatLoop({
+      emitter: em,
+      intervalMs: 1000,
+      onError,
+      send: () => {
+        n += 1;
+        return Promise.reject(new Error(`boom ${n}`));
+      },
+    });
+
+    await vi.advanceTimersByTimeAsync(2000); // immediate + 2
+    expect(n).toBe(3);
+    expect(onError).toHaveBeenCalledTimes(3);
+    loop.stop();
+  });
+});

packages/comms/src/__tests__/liveness.test.ts

@@ -0,0 +1,89 @@
+import { describe, expect, it } from 'vitest';
+
+import { classifyLiveness, computeFleetLiveness } from '../liveness.js';
+import type { HeartbeatObservation, LivenessPolicy } from '../types.js';
+
+// Small, dev-scale policy so the arithmetic is obvious:
+//   online window = interval * missTolerance = 1000 * 2 = 2000ms
+//   dark threshold = 5000ms
+const policy: LivenessPolicy = {
+  heartbeatIntervalMs: 1000,
+  missTolerance: 2,
+  darkThresholdMs: 5000,
+};
+
+describe('classifyLiveness (deterministic, heartbeat-age based — RFC-001 §4.5)', () => {
+  it('is online when age is within interval * missTolerance', () => {
+    expect(classifyLiveness(0, policy)).toBe('online');
+    expect(classifyLiveness(1999, policy)).toBe('online');
+    expect(classifyLiveness(2000, policy)).toBe('online'); // inclusive boundary
+  });
+
+  it('is away when past the online window but before dark threshold', () => {
+    expect(classifyLiveness(2001, policy)).toBe('away');
+    expect(classifyLiveness(4999, policy)).toBe('away');
+  });
+
+  it('is offline/dark at or past the dark threshold', () => {
+    expect(classifyLiveness(5000, policy)).toBe('offline');
+    expect(classifyLiveness(50_000, policy)).toBe('offline');
+  });
+
+  it('treats a never-seen agent (Infinity age) as offline', () => {
+    expect(classifyLiveness(Number.POSITIVE_INFINITY, policy)).toBe('offline');
+  });
+
+  it('never returns online for a negative-but-huge misconfig (guards NaN)', () => {
+    // A NaN age must fail safe to offline, not silently report online.
+    expect(classifyLiveness(Number.NaN, policy)).toBe('offline');
+  });
+});
+
+describe('computeFleetLiveness (A2/A3 core)', () => {
+  const now = 100_000;
+  const obs = (slug: string, lastSeenTs: number, lastSeq = 1): HeartbeatObservation => ({
+    slug,
+    mxid: `@agent-${slug}:matrix.localhost`,
+    lastSeenTs,
+    lastSeq,
+    assertedStatus: 'online',
+  });
+
+  it('classifies a live fleet: fresh=online, stale=away, dark=offline', () => {
+    const result = computeFleetLiveness(
+      [
+        obs('alpha', now - 500), // 500ms old  -> online
+        obs('bravo', now - 3000), // 3000ms old -> away
+        obs('charlie', now - 8000), // 8000ms old -> offline
+      ],
+      now,
+      policy,
+    );
+    const byslug = Object.fromEntries(result.map((r) => [r.slug, r.status]));
+    expect(byslug).toEqual({ alpha: 'online', bravo: 'away', charlie: 'offline' });
+  });
+
+  it('A3: a previously-online agent flips to offline once age crosses dark threshold', () => {
+    const lastBeat = 100_000; // agent was hard-killed right after this beat
+    // Just before the threshold it is still merely "away"...
+    const justBefore = computeFleetLiveness([obs('victim', lastBeat, 7)], lastBeat + 4999, policy);
+    expect(justBefore[0]?.status).toBe('away');
+    // ...and the instant age reaches darkThresholdMs it is deterministically offline,
+    // with no dependence on native Matrix presence timeouts.
+    const atThreshold = computeFleetLiveness([obs('victim', lastBeat, 7)], lastBeat + 5000, policy);
+    expect(atThreshold[0]?.status).toBe('offline');
+    expect(atThreshold[0]?.ageMs).toBe(5000);
+    expect(atThreshold[0]?.lastSeq).toBe(7);
+  });
+
+  it('reports ageMs and preserves mxid/slug/seq for the human view', () => {
+    const [row] = computeFleetLiveness([obs('alpha', now - 1200, 42)], now, policy);
+    expect(row).toMatchObject({
+      slug: 'alpha',
+      mxid: '@agent-alpha:matrix.localhost',
+      ageMs: 1200,
+      lastSeq: 42,
+      status: 'online',
+    });
+  });
+});

packages/comms/src/__tests__/matrix-client.test.ts

@@ -0,0 +1,131 @@
+import { describe, expect, it, vi } from 'vitest';
+
+import { MatrixError, MinimalMatrixClient, toMatrixPresence } from '../matrix-client.js';
+
+const jsonResponse = (status: number, body: unknown): Response =>
+  new Response(JSON.stringify(body), { status, headers: { 'Content-Type': 'application/json' } });
+
+// A fetch mock typed with the (URL, RequestInit?) shape the client actually
+// calls, so mock.calls has a proper tuple type under noUncheckedIndexedAccess.
+const mkFetch = (impl: (url: URL, init?: RequestInit) => Promise<Response>) => vi.fn(impl);
+
+const cfg = {
+  homeserverUrl: 'https://matrix.localhost:8448',
+  accessToken: 'as-secret',
+  actAsUserId: '@agent-alpha:matrix.localhost',
+};
+
+describe('toMatrixPresence', () => {
+  it('maps liveness states to native presence EDU values', () => {
+    expect(toMatrixPresence('online')).toBe('online');
+    expect(toMatrixPresence('away')).toBe('unavailable');
+    expect(toMatrixPresence('offline')).toBe('offline');
+  });
+});
+
+describe('MinimalMatrixClient', () => {
+  it('setPresence PUTs native presence and masquerades via user_id', async () => {
+    const fetchMock = mkFetch(async () => jsonResponse(200, {}));
+    const client = new MinimalMatrixClient(cfg, fetchMock as unknown as typeof fetch);
+    await client.setPresence('@agent-alpha:matrix.localhost', 'away', 'hb');
+
+    const [url, init] = fetchMock.mock.calls[0]!;
+    const u = new URL((url as URL).toString());
+    expect(u.pathname).toBe('/_matrix/client/v3/presence/%40agent-alpha%3Amatrix.localhost/status');
+    expect(u.searchParams.get('user_id')).toBe('@agent-alpha:matrix.localhost');
+    expect(JSON.parse((init as RequestInit).body as string)).toEqual({
+      presence: 'unavailable',
+      status_msg: 'hb',
+    });
+    expect((init as RequestInit).method).toBe('PUT');
+  });
+
+  it('sendHeartbeat posts an m.room.message and returns the event_id', async () => {
+    const fetchMock = mkFetch(async () => jsonResponse(200, { event_id: '$evt1' }));
+    const client = new MinimalMatrixClient(cfg, fetchMock as unknown as typeof fetch);
+    const id = await client.sendHeartbeat('!room:matrix.localhost', {
+      macp_version: '1.0',
+      macp_type: 'presence',
+      msgtype: 'mosaic.presence',
+      agent: { mxid: cfg.actAsUserId, slug: 'alpha', harness: 'claude-code' },
+      ts: 1,
+      body: 'alpha online (seq 1)',
+      status: 'online',
+      seq: 1,
+      interval_ms: 1000,
+    });
+    expect(id).toBe('$evt1');
+    const [url] = fetchMock.mock.calls[0]!;
+    expect((url as URL).pathname).toContain('/rooms/!room%3Amatrix.localhost/send/m.room.message/');
+  });
+
+  it('throws a MatrixError carrying errcode on a non-2xx', async () => {
+    const fetchMock = mkFetch(async () =>
+      jsonResponse(403, { errcode: 'M_FORBIDDEN', error: 'nope' }),
+    );
+    const client = new MinimalMatrixClient(cfg, fetchMock as unknown as typeof fetch);
+    await expect(client.whoami()).rejects.toMatchObject({
+      name: 'MatrixError',
+      status: 403,
+      errcode: 'M_FORBIDDEN',
+    });
+    await expect(client.whoami()).rejects.toBeInstanceOf(MatrixError);
+  });
+
+  it('readHeartbeats reduces the timeline to the latest beat per agent', async () => {
+    // Timeline (dir=b => most-recent first). alpha has two beats; keep highest seq.
+    const chunk = [
+      {
+        sender: '@agent-bravo:matrix.localhost',
+        origin_server_ts: 9000,
+        content: {
+          msgtype: 'mosaic.presence',
+          agent: { slug: 'bravo', mxid: '@agent-bravo:matrix.localhost' },
+          seq: 5,
+          status: 'online',
+          ts: 8999,
+        },
+      },
+      {
+        sender: '@agent-alpha:matrix.localhost',
+        origin_server_ts: 8000,
+        content: {
+          msgtype: 'mosaic.presence',
+          agent: { slug: 'alpha', mxid: '@agent-alpha:matrix.localhost' },
+          seq: 12,
+          status: 'online',
+          ts: 7999,
+        },
+      },
+      {
+        // an ordinary chat message must be ignored
+        sender: '@human:matrix.localhost',
+        origin_server_ts: 7000,
+        content: { msgtype: 'm.text', body: 'hi' },
+      },
+      {
+        sender: '@agent-alpha:matrix.localhost',
+        origin_server_ts: 6000,
+        content: {
+          msgtype: 'mosaic.presence',
+          agent: { slug: 'alpha', mxid: '@agent-alpha:matrix.localhost' },
+          seq: 11,
+          status: 'online',
+          ts: 5999,
+        },
+      },
+    ];
+    const fetchMock = mkFetch(async () => jsonResponse(200, { chunk }));
+    const client = new MinimalMatrixClient(cfg, fetchMock as unknown as typeof fetch);
+    const obs = await client.readHeartbeats('!room:matrix.localhost');
+
+    const bySlug = Object.fromEntries(obs.map((o) => [o.slug, o]));
+    expect(Object.keys(bySlug).sort()).toEqual(['alpha', 'bravo']);
+    expect(bySlug.alpha).toMatchObject({ lastSeq: 12, lastSeenTs: 8000 }); // highest seq wins, server ts
+    expect(bySlug.bravo).toMatchObject({ lastSeq: 5, lastSeenTs: 9000 });
+
+    const [url] = fetchMock.mock.calls[0]!;
+    const u = new URL((url as URL).toString());
+    expect(u.searchParams.get('dir')).toBe('b');
+  });
+});

packages/comms/src/__tests__/presence-flow.test.ts

@@ -0,0 +1,151 @@
+import { afterEach, describe, expect, it, vi } from 'vitest';
+
+import { FleetLivenessReader } from '../liveness-reader.js';
+import type { MinimalMatrixClient } from '../matrix-client.js';
+import { PresenceAgent } from '../presence-agent.js';
+import type { HeartbeatObservation, LivenessPolicy, PresenceStatus } from '../types.js';
+
+/**
+ * An in-memory fake homeserver room: records heartbeats with a controllable
+ * server clock and reduces them exactly like the real readHeartbeats. Lets us
+ * prove the PresenceAgent -> room -> FleetLivenessReader flow (including the A3
+ * hard-kill -> offline transition) deterministically, with no network.
+ */
+class FakeRoomClient {
+  readonly beats: Array<{
+    slug: string;
+    mxid: string;
+    seq: number;
+    ts: number;
+    status: PresenceStatus;
+  }> = [];
+  presence: Record<string, PresenceStatus> = {};
+
+  constructor(private readonly clock: () => number) {}
+
+  async joinRoom(roomId: string): Promise<string> {
+    return roomId;
+  }
+  async setPresence(userId: string, status: PresenceStatus): Promise<void> {
+    this.presence[userId] = status;
+  }
+  async sendHeartbeat(
+    _roomId: string,
+    content: { agent: { slug: string; mxid: string }; seq: number; status: PresenceStatus },
+  ): Promise<string> {
+    this.beats.push({
+      slug: content.agent.slug,
+      mxid: content.agent.mxid,
+      seq: content.seq,
+      ts: this.clock(), // server receive time
+      status: content.status,
+    });
+    return `$evt${this.beats.length}`;
+  }
+  async readHeartbeats(): Promise<HeartbeatObservation[]> {
+    const bySlug = new Map<string, HeartbeatObservation>();
+    for (const b of this.beats) {
+      const prev = bySlug.get(b.slug);
+      if (!prev || b.seq > prev.lastSeq) {
+        bySlug.set(b.slug, {
+          slug: b.slug,
+          mxid: b.mxid,
+          lastSeenTs: b.ts,
+          lastSeq: b.seq,
+          assertedStatus: b.status,
+        });
+      }
+    }
+    return [...bySlug.values()];
+  }
+}
+
+const policy: LivenessPolicy = {
+  heartbeatIntervalMs: 1000,
+  missTolerance: 2,
+  darkThresholdMs: 5000,
+};
+
+describe('presence flow (A2 + A3 at unit level)', () => {
+  afterEach(() => vi.useRealTimers());
+
+  it('shows agents online while beating, then A3: a hard-killed agent goes offline within dark_threshold', async () => {
+    vi.useFakeTimers();
+    vi.setSystemTime(0);
+
+    const fake = new FakeRoomClient(() => Date.now());
+    const client = fake as unknown as MinimalMatrixClient;
+    const reader = new FleetLivenessReader({
+      client,
+      roomId: '!fleet',
+      policy,
+      now: () => Date.now(),
+    });
+
+    const mk = (slug: string) =>
+      new PresenceAgent({
+        client,
+        agent: { mxid: `@agent-${slug}:matrix.localhost`, slug, harness: 'claude-code' },
+        roomId: '!fleet',
+        intervalMs: 1000,
+        policy,
+      });
+
+    const alpha = mk('alpha');
+    const bravo = mk('bravo');
+    const charlie = mk('charlie');
+
+    for (const a of [alpha, bravo, charlie]) {
+      await a.connect();
+      a.start();
+    }
+    // native presence set online for all three (Element dot)
+    expect(fake.presence['@agent-alpha:matrix.localhost']).toBe('online');
+
+    // let a couple of beats flow — all three fresh => online (A2)
+    await vi.advanceTimersByTimeAsync(1500);
+    const board1 = Object.fromEntries((await reader.read()).map((r) => [r.slug, r.status]));
+    expect(board1).toEqual({ alpha: 'online', bravo: 'online', charlie: 'online' });
+
+    // HARD-KILL charlie: stop its loop, no more beats. alpha/bravo keep beating.
+    charlie.pauseHeartbeat(); // hard-kill: no graceful presence signal
+
+    // advance to just before dark threshold from charlie's last beat...
+    await vi.advanceTimersByTimeAsync(3000);
+    const mid = Object.fromEntries((await reader.read()).map((r) => [r.slug, r.status]));
+    expect(mid.alpha).toBe('online');
+    expect(mid.charlie).not.toBe('online'); // already stale (away)
+
+    // ...advance past dark_threshold: charlie is deterministically offline.
+    await vi.advanceTimersByTimeAsync(4000);
+    const final = await reader.read();
+    const byslug = Object.fromEntries(final.map((r) => [r.slug, r]));
+    expect(byslug.charlie!.status).toBe('offline');
+    expect(byslug.alpha!.status).toBe('online');
+    expect(byslug.bravo!.status).toBe('online');
+
+    for (const a of [alpha, bravo]) await a.stop();
+  });
+
+  it('formatBoard renders a human-readable liveness board (A4)', async () => {
+    const fake = new FakeRoomClient(() => 10_000);
+    fake.beats.push({
+      slug: 'alpha',
+      mxid: '@agent-alpha:matrix.localhost',
+      seq: 3,
+      ts: 9_500,
+      status: 'online',
+    });
+    const reader = new FleetLivenessReader({
+      client: fake as unknown as MinimalMatrixClient,
+      roomId: '!fleet',
+      policy,
+      now: () => 10_000,
+    });
+    const board = await reader.formatBoard();
+    expect(board).toContain('Fleet presence');
+    expect(board).toContain('alpha');
+    expect(board).toContain('online');
+    expect(board).toContain('online=1');
+  });
+});