feat(comms): P1 presence — minimal Synapse + fleet presence room + mosaic.presence heartbeat + liveness
All checks were successful
ci/woodpecker/pr/ci Pipeline was successful

Implements RFC-001 P1 (the first shippable slice): deterministic Matrix
presence/liveness on a single-instance dev Synapse.

- infra/matrix/: DEV Synapse (RFC-002 Mode B, federation OFF, self-signed TLS,
  enable_registration:false, appservice registration wired). Rendered from
  parameterized templates — zero hardcoded topology. .data is gitignored.
- packages/comms/: minimal MACP presence SDK — set presence, run the
  mosaic.presence heartbeat (seq + interval per RFC-001 §4.5), and a
  deterministic liveness reader (online/away/offline from heartbeat age, NOT
  native-presence-timeout). Liveness core written RED-FIRST. 19 vitest tests.
- tools/matrix-presence-harness/: minimal provisioner (registers >=3 agent
  MXIDs, creates the fleet presence room, joins them — reuses the existing
  @mosaicstack/appservice intent lib) + an E2E validation harness proving
  A2/A3/A4 against a real Synapse.
- eslint.config.mjs: register packages/comms/vitest.config.ts with the
  type-aware project service (same as other packages' vitest configs).

DEV-compose validated only; production deploy is a separate coordinated step
(deploy-holds respected).

Part of the comms-evolution program (RFC-001 P1)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0158NZqN2n2ymKFeJAZ4GUCb
This commit is contained in:
mosaic-coder
2026-07-24 20:18:18 -05:00
parent 529c177830
commit 2e7c124e4d
33 changed files with 2140 additions and 0 deletions

41
packages/comms/README.md Normal file
View File

@@ -0,0 +1,41 @@
# @mosaicstack/comms
MACP presence SDK — the **P1 (presence)** slice of RFC-001 (§4.5 liveness,
§4.2 event envelope). Minimal by design: set Matrix presence, run the
`mosaic.presence` heartbeat, and compute **deterministic** fleet liveness.
Out of P1 scope (later phases): enrollment/auto-detect, room taxonomy,
per-agent token minting, signed-authorship, federation.
## API
- `classifyLiveness(ageMs, policy)` / `computeFleetLiveness(observations, now, policy)`
— pure, deterministic online/away/offline from heartbeat age. The
authoritative liveness source (RFC-001 §4.5): native Matrix presence is _not_
relied upon.
- `HeartbeatEmitter` / `startHeartbeatLoop(...)` — build and drive the
`mosaic.presence` heartbeat (monotonic `seq`, `interval_ms`).
- `MinimalMatrixClient` — tiny C-S client: `setPresence`, `sendHeartbeat`,
`readHeartbeats`, `joinRoom`. Supports Application-Service masquerade
(`actAsUserId`) for the P1 provisioner, or a per-agent `accessToken`.
- `PresenceAgent` — high-level: join the fleet room, go present, heartbeat.
`pauseHeartbeat()` models a crash (no graceful signal).
- `FleetLivenessReader` — reads the fleet room and computes the liveness board
(`read()` / `formatBoard()`), the surface a human or watchdog reads.
## Liveness policy (RFC-001 §4.5)
```
online : age <= heartbeatIntervalMs * missTolerance
away : age < darkThresholdMs
offline: otherwise (or never-seen / non-finite age -> fail safe to offline)
```
Defaults: interval 30s, miss-tolerance 2, dark-threshold 10min
(`DEFAULT_LIVENESS_POLICY`). All runtime-tunable per RFC-002 §5.3.
## Tests
`pnpm --filter @mosaicstack/comms test` — the liveness core is written
RED-FIRST; an end-to-end proof against a real Synapse lives in
`tools/matrix-presence-harness`.

View File

@@ -0,0 +1,37 @@
{
"name": "@mosaicstack/comms",
"version": "0.0.1",
"type": "module",
"repository": {
"type": "git",
"url": "https://git.mosaicstack.dev/mosaicstack/stack.git",
"directory": "packages/comms"
},
"main": "dist/index.js",
"types": "dist/index.d.ts",
"exports": {
".": {
"types": "./dist/index.d.ts",
"default": "./dist/index.js"
}
},
"scripts": {
"build": "tsc",
"lint": "eslint src",
"typecheck": "tsc --noEmit",
"test": "vitest run --passWithNoTests"
},
"devDependencies": {
"@types/node": "^22.0.0",
"@vitest/coverage-v8": "^2.0.0",
"typescript": "^5.8.0",
"vitest": "^2.0.0"
},
"publishConfig": {
"registry": "https://git.mosaicstack.dev/api/packages/mosaicstack/npm/",
"access": "public"
},
"files": [
"dist"
]
}

View File

@@ -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();
});
});

View File

@@ -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',
});
});
});

View File

@@ -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');
});
});

View File

@@ -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');
});
});

View File

@@ -0,0 +1,124 @@
/**
* `mosaic.presence` heartbeat construction and loop (RFC-001 §4.2/§4.5).
*
* The emitter is deterministic and side-effect free (easy to unit test): it
* owns the monotonic `seq` and stamps each beat. The loop wires the emitter to
* a sender on an interval; timers are injectable so the loop is testable with
* fake clocks.
*/
import { MACP_VERSION, type PresenceHeartbeatContent, type PresenceStatus } from './types.js';
export interface HeartbeatAgentIdentity {
mxid: string;
slug: string;
harness: string;
}
export interface HeartbeatEmitterOptions {
agent: HeartbeatAgentIdentity;
/** Nominal interval advertised in each beat (interval_ms). */
intervalMs: number;
/** Optional mission correlation (RFC-001 §4.2 envelope). */
missionId?: string;
/** Injectable clock for deterministic tests. Default Date.now. */
now?: () => number;
}
/**
* Produces successive heartbeat contents with a monotonically increasing seq.
* The first `next()` returns seq=1.
*/
export class HeartbeatEmitter {
private seq = 0;
private readonly now: () => number;
constructor(private readonly opts: HeartbeatEmitterOptions) {
this.now = opts.now ?? Date.now;
}
/** Current sequence number (0 before the first beat). */
get currentSeq(): number {
return this.seq;
}
/** Build the next heartbeat content, advancing the sequence. */
next(status: PresenceStatus = 'online'): PresenceHeartbeatContent {
this.seq += 1;
const ts = this.now();
const content: PresenceHeartbeatContent = {
macp_version: MACP_VERSION,
macp_type: 'presence',
msgtype: 'mosaic.presence',
agent: {
mxid: this.opts.agent.mxid,
slug: this.opts.agent.slug,
harness: this.opts.agent.harness,
},
ts,
body: `${this.opts.agent.slug} ${status} (seq ${this.seq})`,
status,
seq: this.seq,
interval_ms: this.opts.intervalMs,
};
if (this.opts.missionId !== undefined) {
content.mission_id = this.opts.missionId;
}
return content;
}
}
export type HeartbeatSender = (content: PresenceHeartbeatContent) => void | Promise<void>;
export interface HeartbeatLoopOptions {
emitter: HeartbeatEmitter;
send: HeartbeatSender;
intervalMs: number;
/** Status supplier evaluated each beat. Default: always 'online'. */
status?: () => PresenceStatus;
/** Called if a beat's send rejects (so a transient failure doesn't kill the loop). */
onError?: (err: unknown) => void;
/** Injectable timer (tests). Defaults to global setInterval/clearInterval. */
setIntervalFn?: (cb: () => void, ms: number) => unknown;
clearIntervalFn?: (handle: unknown) => void;
}
/** A running heartbeat loop; call stop() to end it. */
export interface HeartbeatLoopHandle {
stop: () => void;
}
/**
* Start a heartbeat loop: emits one beat immediately, then every intervalMs.
* Returns a handle whose `stop()` is idempotent.
*/
export function startHeartbeatLoop(opts: HeartbeatLoopOptions): HeartbeatLoopHandle {
const status = opts.status ?? (() => 'online' as PresenceStatus);
const onError = opts.onError ?? (() => {});
const setIntervalFn = opts.setIntervalFn ?? ((cb, ms) => setInterval(cb, ms));
const clearIntervalFn =
opts.clearIntervalFn ?? ((h) => clearInterval(h as ReturnType<typeof setInterval>));
const beat = (): void => {
try {
const result = opts.send(opts.emitter.next(status()));
if (result instanceof Promise) {
result.catch(onError);
}
} catch (err) {
onError(err);
}
};
beat(); // immediate first beat so liveness is fresh at once
const handle = setIntervalFn(beat, opts.intervalMs);
let stopped = false;
return {
stop: () => {
if (stopped) return;
stopped = true;
clearIntervalFn(handle);
},
};
}

View File

@@ -0,0 +1,42 @@
/**
* @mosaicstack/comms — MACP presence SDK (RFC-001 P1).
*
* Minimal, dev-validated slice: set Matrix presence, run the `mosaic.presence`
* heartbeat, and compute deterministic fleet liveness. Enrollment, room
* taxonomy, token minting and signed-authorship are explicitly out of P1.
*/
export { classifyLiveness, computeFleetLiveness } from './liveness.js';
export {
HeartbeatEmitter,
startHeartbeatLoop,
type HeartbeatAgentIdentity,
type HeartbeatEmitterOptions,
type HeartbeatSender,
type HeartbeatLoopOptions,
type HeartbeatLoopHandle,
} from './heartbeat.js';
export {
MinimalMatrixClient,
MatrixError,
toMatrixPresence,
type MatrixClientConfig,
} from './matrix-client.js';
export { FleetLivenessReader, type FleetLivenessReaderOptions } from './liveness-reader.js';
export { PresenceAgent, type PresenceAgentOptions } from './presence-agent.js';
export {
DEFAULT_LIVENESS_POLICY,
MACP_VERSION,
type AgentLiveness,
type HeartbeatObservation,
type LivenessPolicy,
type MacpEnvelope,
type MatrixPresence,
type PresenceHeartbeatContent,
type PresenceStatus,
} from './types.js';

View File

@@ -0,0 +1,58 @@
/**
* Fleet liveness reader (RFC-001 §4.5, A2/A4).
*
* Reads `mosaic.presence` heartbeats from the fleet presence room and computes
* deterministic online/away/offline for every agent. This is the surface a
* human (or the escalation watchdog, P2+) reads to answer "who's alive?".
*/
import { computeFleetLiveness } from './liveness.js';
import type { MinimalMatrixClient } from './matrix-client.js';
import { DEFAULT_LIVENESS_POLICY, type AgentLiveness, type LivenessPolicy } from './types.js';
export interface FleetLivenessReaderOptions {
client: MinimalMatrixClient;
/** The fleet presence room (id or resolved id). */
roomId: string;
policy?: LivenessPolicy;
/** Injectable clock for tests. Default Date.now. */
now?: () => number;
/** How many timeline events to scan back. Default 200. */
scanLimit?: number;
}
export class FleetLivenessReader {
private readonly policy: LivenessPolicy;
private readonly now: () => number;
constructor(private readonly opts: FleetLivenessReaderOptions) {
this.policy = opts.policy ?? DEFAULT_LIVENESS_POLICY;
this.now = opts.now ?? Date.now;
}
/** Read the room and compute current liveness for every seen agent. */
async read(): Promise<AgentLiveness[]> {
const observations = await this.opts.client.readHeartbeats(
this.opts.roomId,
this.opts.scanLimit ?? 200,
);
return computeFleetLiveness(observations, this.now(), this.policy);
}
/** A compact human-readable liveness board (A4 CLI view). */
async formatBoard(): Promise<string> {
const rows = await this.read();
rows.sort((a, b) => a.slug.localeCompare(b.slug));
const dot: Record<string, string> = { online: '🟢', away: '🟡', offline: '🔴' };
const lines = rows.map(
(r) =>
`${dot[r.status] ?? '⚪'} ${r.slug.padEnd(16)} ${r.status.padEnd(8)} ` +
`age=${(r.ageMs / 1000).toFixed(1)}s seq=${r.lastSeq} ${r.mxid}`,
);
const summary =
`online=${rows.filter((r) => r.status === 'online').length} ` +
`away=${rows.filter((r) => r.status === 'away').length} ` +
`offline=${rows.filter((r) => r.status === 'offline').length}`;
return [`Fleet presence — ${summary}`, ...lines].join('\n');
}
}

View File

@@ -0,0 +1,63 @@
/**
* Deterministic liveness computation (RFC-001 §4.5).
*
* The authoritative liveness signal is the `mosaic.presence` heartbeat, NOT
* native Matrix presence. Given the age of an agent's last heartbeat and a
* policy, these pure functions classify online/away/offline the same way every
* time — which is exactly what makes the A3 "hard-killed agent flips to
* offline within dark_threshold" guarantee deterministic and testable without
* standing up a homeserver.
*/
import type {
AgentLiveness,
HeartbeatObservation,
LivenessPolicy,
PresenceStatus,
} from './types.js';
/**
* Classify a single agent from the age (ms) of its last heartbeat.
*
* - `age <= heartbeatIntervalMs * missTolerance` → **online**
* - `age < darkThresholdMs` → **away**
* - otherwise (or non-finite age) → **offline / dark**
*
* A non-finite age (never seen / NaN) fails safe to `offline`: we never assert
* a liveness we cannot substantiate.
*/
export function classifyLiveness(ageMs: number, policy: LivenessPolicy): PresenceStatus {
if (!Number.isFinite(ageMs)) {
return 'offline';
}
const onlineWindowMs = policy.heartbeatIntervalMs * policy.missTolerance;
if (ageMs <= onlineWindowMs) {
return 'online';
}
if (ageMs < policy.darkThresholdMs) {
return 'away';
}
return 'offline';
}
/**
* Compute liveness for every observed agent at wall-clock `nowMs`.
* The result order mirrors the input order (stable for display).
*/
export function computeFleetLiveness(
observations: readonly HeartbeatObservation[],
nowMs: number,
policy: LivenessPolicy,
): AgentLiveness[] {
return observations.map((o) => {
const ageMs = nowMs - o.lastSeenTs;
return {
slug: o.slug,
mxid: o.mxid,
status: classifyLiveness(ageMs, policy),
lastSeenTs: o.lastSeenTs,
ageMs,
lastSeq: o.lastSeq,
};
});
}

View File

@@ -0,0 +1,204 @@
/**
* Minimal Matrix Client-Server API client for the P1 presence slice.
*
* Deliberately tiny: only the calls presence needs (whoami, set native
* presence, send a timeline event, read recent timeline). Auth is a single
* bearer token; an optional `actAsUserId` enables Application-Service
* masquerade (`?user_id=`) so the P1 provisioner can drive several virtual
* agents with one as_token in dev (RFC-001 §2.2 step 4/Appendix A). Agents
* holding their own access_token simply omit `actAsUserId`.
*
* `fetch` is injectable for unit tests.
*/
import crypto from 'node:crypto';
import type {
HeartbeatObservation,
MatrixPresence,
PresenceHeartbeatContent,
PresenceStatus,
} from './types.js';
export interface MatrixClientConfig {
/** Client-Server API base, e.g. https://matrix.localhost:8448 */
homeserverUrl: string;
/** Bearer token (a per-agent access_token, or an as_token for masquerade). */
accessToken: string;
/** If set, all calls masquerade as this MXID via ?user_id= (AS mode). */
actAsUserId?: string;
}
export class MatrixError extends Error {
constructor(
readonly status: number,
readonly errcode: string | undefined,
message: string,
) {
super(message);
this.name = 'MatrixError';
}
}
type FetchLike = typeof fetch;
/** Map our authoritative liveness state to the native Matrix presence EDU. */
export function toMatrixPresence(status: PresenceStatus): MatrixPresence {
switch (status) {
case 'online':
return 'online';
case 'away':
return 'unavailable';
case 'offline':
return 'offline';
}
}
export class MinimalMatrixClient {
private readonly fetchImpl: FetchLike;
constructor(
private readonly cfg: MatrixClientConfig,
fetchImpl?: FetchLike,
) {
this.fetchImpl = fetchImpl ?? fetch;
}
private async request(
method: string,
path: string,
options: { query?: Record<string, string>; body?: unknown } = {},
): Promise<Record<string, unknown>> {
const url = new URL(this.cfg.homeserverUrl.replace(/\/$/, '') + path);
if (this.cfg.actAsUserId) {
url.searchParams.set('user_id', this.cfg.actAsUserId);
}
for (const [k, v] of Object.entries(options.query ?? {})) {
url.searchParams.set(k, v);
}
const res = await this.fetchImpl(url, {
method,
headers: {
Authorization: `Bearer ${this.cfg.accessToken}`,
'Content-Type': 'application/json',
},
body: options.body === undefined ? undefined : JSON.stringify(options.body),
});
const text = await res.text();
const data = (text ? JSON.parse(text) : {}) as Record<string, unknown>;
if (!res.ok) {
throw new MatrixError(
res.status,
typeof data.errcode === 'string' ? data.errcode : undefined,
`${method} ${path} -> ${res.status}: ${text.slice(0, 300)}`,
);
}
return data;
}
/** GET /account/whoami — resolves the acting MXID. */
async whoami(): Promise<string> {
const data = await this.request('GET', '/_matrix/client/v3/account/whoami');
if (typeof data.user_id !== 'string') {
throw new MatrixError(500, undefined, 'whoami returned no user_id');
}
return data.user_id;
}
/**
* Set the native Matrix presence EDU (so Element shows the right dot for
* humans). NOT the authoritative liveness signal — the heartbeat is.
*/
async setPresence(userId: string, status: PresenceStatus, statusMsg?: string): Promise<void> {
const user = encodeURIComponent(userId);
await this.request('PUT', `/_matrix/client/v3/presence/${user}/status`, {
body: {
presence: toMatrixPresence(status),
...(statusMsg ? { status_msg: statusMsg } : {}),
},
});
}
/** Send an arbitrary timeline event; returns its event_id. */
async sendEvent(
roomId: string,
eventType: string,
content: Record<string, unknown>,
): Promise<string> {
const room = encodeURIComponent(roomId);
const txn = `mosaic-comms-${crypto.randomUUID()}`;
const data = await this.request(
'PUT',
`/_matrix/client/v3/rooms/${room}/send/${encodeURIComponent(eventType)}/${txn}`,
{ body: content },
);
if (typeof data.event_id !== 'string') {
throw new MatrixError(500, undefined, 'send returned no event_id');
}
return data.event_id;
}
/** Post a `mosaic.presence` heartbeat (m.room.message carrier) to the room. */
async sendHeartbeat(roomId: string, content: PresenceHeartbeatContent): Promise<string> {
return this.sendEvent(roomId, 'm.room.message', content as unknown as Record<string, unknown>);
}
/** Join a room (by id or alias). Idempotent on the server. */
async joinRoom(roomIdOrAlias: string): Promise<string> {
const data = await this.request(
'POST',
`/_matrix/client/v3/join/${encodeURIComponent(roomIdOrAlias)}`,
{ body: {} },
);
if (typeof data.room_id !== 'string') {
throw new MatrixError(500, undefined, 'join returned no room_id');
}
return data.room_id;
}
/**
* Read recent `mosaic.presence` heartbeats from a room and reduce them to the
* latest observation per agent. Walks the timeline backwards (most-recent
* first) and keeps, per slug, the beat with the highest seq.
*
* `lastSeenTs` uses the server's `origin_server_ts` (honest "when we last
* heard from it"), falling back to the agent-stamped envelope `ts`.
*/
async readHeartbeats(roomId: string, limit = 200): Promise<HeartbeatObservation[]> {
const room = encodeURIComponent(roomId);
const data = await this.request('GET', `/_matrix/client/v3/rooms/${room}/messages`, {
query: { dir: 'b', limit: String(limit) },
});
const chunk = Array.isArray(data.chunk) ? (data.chunk as Array<Record<string, unknown>>) : [];
const bySlug = new Map<string, HeartbeatObservation>();
for (const ev of chunk) {
const content = ev.content as Record<string, unknown> | undefined;
if (!content || content.msgtype !== 'mosaic.presence') continue;
const agent = content.agent as Record<string, unknown> | undefined;
const slug = agent && typeof agent.slug === 'string' ? agent.slug : undefined;
const mxid =
agent && typeof agent.mxid === 'string'
? agent.mxid
: typeof ev.sender === 'string'
? ev.sender
: undefined;
if (!slug || !mxid) continue;
const seq = typeof content.seq === 'number' ? content.seq : 0;
const serverTs = typeof ev.origin_server_ts === 'number' ? ev.origin_server_ts : undefined;
const envelopeTs = typeof content.ts === 'number' ? content.ts : undefined;
const lastSeenTs = serverTs ?? envelopeTs ?? 0;
const assertedStatus =
content.status === 'online' || content.status === 'away' || content.status === 'offline'
? (content.status as PresenceStatus)
: 'offline';
const prev = bySlug.get(slug);
if (!prev || seq > prev.lastSeq) {
bySlug.set(slug, { slug, mxid, lastSeenTs, lastSeq: seq, assertedStatus });
}
}
return [...bySlug.values()];
}
}

View File

@@ -0,0 +1,92 @@
/**
* High-level presence agent (RFC-001 §4.1 steps 1011, §4.5).
*
* Ties the pieces together for one agent: join the fleet presence room, set
* native Matrix presence online (for Element's dot), and run the authoritative
* `mosaic.presence` heartbeat loop. This is the P1 slice of what a harness does
* on spin — no enrollment/token-minting/introductions (those are P2).
*/
import {
HeartbeatEmitter,
startHeartbeatLoop,
type HeartbeatAgentIdentity,
type HeartbeatLoopHandle,
} from './heartbeat.js';
import type { MinimalMatrixClient } from './matrix-client.js';
import { DEFAULT_LIVENESS_POLICY, type LivenessPolicy, type PresenceStatus } from './types.js';
export interface PresenceAgentOptions {
client: MinimalMatrixClient;
agent: HeartbeatAgentIdentity;
/** Fleet presence room id (or alias) to heartbeat into. */
roomId: string;
/** Heartbeat cadence; defaults to the policy interval. */
intervalMs?: number;
policy?: LivenessPolicy;
missionId?: string;
onError?: (err: unknown) => void;
}
export class PresenceAgent {
private readonly intervalMs: number;
private readonly emitter: HeartbeatEmitter;
private loop: HeartbeatLoopHandle | undefined;
private resolvedRoomId: string | undefined;
constructor(private readonly opts: PresenceAgentOptions) {
const policy = opts.policy ?? DEFAULT_LIVENESS_POLICY;
this.intervalMs = opts.intervalMs ?? policy.heartbeatIntervalMs;
this.emitter = new HeartbeatEmitter({
agent: opts.agent,
intervalMs: this.intervalMs,
missionId: opts.missionId,
});
}
/** Join the fleet room and go present. Returns the resolved room id. */
async connect(): Promise<string> {
this.resolvedRoomId = await this.opts.client.joinRoom(this.opts.roomId);
await this.opts.client.setPresence(this.opts.agent.mxid, 'online', 'mosaic.presence heartbeat');
return this.resolvedRoomId;
}
/** Start the heartbeat loop (emits immediately, then every intervalMs). */
start(status: () => PresenceStatus = () => 'online'): void {
const roomId = this.resolvedRoomId ?? this.opts.roomId;
this.loop = startHeartbeatLoop({
emitter: this.emitter,
intervalMs: this.intervalMs,
status,
onError: this.opts.onError,
send: async (content) => {
await this.opts.client.sendHeartbeat(roomId, content);
},
});
}
get currentSeq(): number {
return this.emitter.currentSeq;
}
/**
* Stop only the heartbeat loop, sending NO graceful signal. This models a
* hard crash/kill: the authoritative liveness path must detect it purely from
* the absence of heartbeats (RFC-001 §4.5, A3), not from any native presence
* change. Idempotent.
*/
pauseHeartbeat(): void {
this.loop?.stop();
this.loop = undefined;
}
/** Graceful stop: stop heartbeating and drop native presence to offline. */
async stop(): Promise<void> {
this.pauseHeartbeat();
try {
await this.opts.client.setPresence(this.opts.agent.mxid, 'offline');
} catch (err) {
this.opts.onError?.(err);
}
}
}

View File

@@ -0,0 +1,99 @@
/**
* @mosaicstack/comms — MACP P1 (presence) types.
*
* Implements the presence/liveness slice of RFC-001 §4.5 and the MACP event
* envelope of RFC-001 §4.2. P1 scope only: presence heartbeat + deterministic
* liveness. No enrollment, room-taxonomy, token-minting or signed-authorship
* (those are P2+).
*/
/** The three human-visible liveness states (RFC-001 §4.5). */
export type PresenceStatus = 'online' | 'away' | 'offline';
/**
* Native Matrix presence EDU states. We still emit these (so Element shows the
* right dot for humans, RFC-001 §4.5) but they are NOT the authoritative
* liveness source — the heartbeat is.
*/
export type MatrixPresence = 'online' | 'unavailable' | 'offline';
/**
* Common MACP event envelope carried in `content` on every custom event
* (RFC-001 §4.2). P1 uses only the fields the presence heartbeat needs; the
* `signature` field (gate actions, §4.4) is intentionally absent in P1.
*/
export interface MacpEnvelope {
macp_version: string;
macp_type: string;
agent: {
mxid: string;
slug: string;
harness: string;
};
ts: number;
mission_id?: string;
}
/**
* `mosaic.presence` heartbeat content (RFC-001 §4.2 "presence" row + §4.5).
* Carried as an `m.room.message` with `msgtype: "mosaic.presence"` and a
* human-visible `body` fallback, posted into the fleet presence room.
*/
export interface PresenceHeartbeatContent extends MacpEnvelope {
macp_type: 'presence';
msgtype: 'mosaic.presence';
/** Human-visible fallback so the event renders in a stock client. */
body: string;
/** Liveness state the agent asserts about itself. */
status: PresenceStatus;
/** Monotonic per-agent sequence number, increments once per beat. */
seq: number;
/** The agent's configured heartbeat interval, so readers can reason. */
interval_ms: number;
}
/**
* Deterministic liveness policy (RFC-001 §4.5). Defaults per §4.5/§5.3:
* interval 30s, miss-tolerance 2, dark threshold a policy value (10 min in
* prod §5; small in dev harness).
*/
export interface LivenessPolicy {
/** Nominal heartbeat interval in ms. Default 30_000. */
heartbeatIntervalMs: number;
/** How many intervals may be missed before "away". Default 2. */
missTolerance: number;
/** Age past which an agent is declared offline/dark. Default 600_000. */
darkThresholdMs: number;
}
/** A single agent's last observed heartbeat, as read from the fleet room. */
export interface HeartbeatObservation {
slug: string;
mxid: string;
/** Wall-clock ms of the last heartbeat seen for this agent. */
lastSeenTs: number;
/** Last seq observed (monotonic per agent). */
lastSeq: number;
/** The status the agent last asserted about itself. */
assertedStatus: PresenceStatus;
}
/** Computed liveness for one agent (what a human/watchdog reads). */
export interface AgentLiveness {
slug: string;
mxid: string;
/** Authoritative, heartbeat-derived status. */
status: PresenceStatus;
lastSeenTs: number;
/** now - lastSeenTs, in ms. */
ageMs: number;
lastSeq: number;
}
export const DEFAULT_LIVENESS_POLICY: LivenessPolicy = {
heartbeatIntervalMs: 30_000,
missTolerance: 2,
darkThresholdMs: 600_000,
};
export const MACP_VERSION = '1.0';

View File

@@ -0,0 +1,9 @@
{
"extends": "../../tsconfig.base.json",
"compilerOptions": {
"outDir": "dist",
"rootDir": "src"
},
"include": ["src/**/*"],
"exclude": ["node_modules", "dist"]
}

View File

@@ -0,0 +1,13 @@
import { defineConfig } from 'vitest/config';
export default defineConfig({
test: {
globals: true,
environment: 'node',
coverage: {
provider: 'v8',
include: ['src/**/*.ts'],
exclude: ['src/index.ts'],
},
},
});