From 2e7c124e4dfa7314f6c06e751a655b4070279c5e Mon Sep 17 00:00:00 2001 From: mosaic-coder Date: Fri, 24 Jul 2026 20:18:18 -0500 Subject: [PATCH] =?UTF-8?q?feat(comms):=20P1=20presence=20=E2=80=94=20mini?= =?UTF-8?q?mal=20Synapse=20+=20fleet=20presence=20room=20+=20mosaic.presen?= =?UTF-8?q?ce=20heartbeat=20+=20liveness?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_0158NZqN2n2ymKFeJAZ4GUCb --- eslint.config.mjs | 1 + infra/matrix/.gitignore | 3 + infra/matrix/README.md | 53 +++++ .../matrix/appservice/mosaic-as.dev.yaml.tpl | 30 +++ infra/matrix/dev-down.sh | 12 + infra/matrix/dev-up.sh | 108 +++++++++ infra/matrix/docker-compose.dev.yml | 60 +++++ infra/matrix/synapse/homeserver.dev.yaml.tpl | 84 +++++++ infra/matrix/synapse/log.config | 16 ++ packages/comms/README.md | 41 ++++ packages/comms/package.json | 37 +++ .../comms/src/__tests__/heartbeat.test.ts | 90 ++++++++ packages/comms/src/__tests__/liveness.test.ts | 89 ++++++++ .../comms/src/__tests__/matrix-client.test.ts | 131 +++++++++++ .../comms/src/__tests__/presence-flow.test.ts | 151 +++++++++++++ packages/comms/src/heartbeat.ts | 124 +++++++++++ packages/comms/src/index.ts | 42 ++++ packages/comms/src/liveness-reader.ts | 58 +++++ packages/comms/src/liveness.ts | 63 ++++++ packages/comms/src/matrix-client.ts | 204 +++++++++++++++++ packages/comms/src/presence-agent.ts | 92 ++++++++ packages/comms/src/types.ts | 99 +++++++++ packages/comms/tsconfig.json | 9 + packages/comms/vitest.config.ts | 13 ++ pnpm-lock.yaml | 37 +++ pnpm-workspace.yaml | 1 + tools/matrix-presence-harness/README.md | 43 ++++ tools/matrix-presence-harness/agent-proc.ts | 70 ++++++ tools/matrix-presence-harness/package.json | 24 ++ tools/matrix-presence-harness/provision.ts | 89 ++++++++ tools/matrix-presence-harness/run.sh | 47 ++++ tools/matrix-presence-harness/tsconfig.json | 9 + tools/matrix-presence-harness/validate.ts | 210 ++++++++++++++++++ 33 files changed, 2140 insertions(+) create mode 100644 infra/matrix/.gitignore create mode 100644 infra/matrix/README.md create mode 100644 infra/matrix/appservice/mosaic-as.dev.yaml.tpl create mode 100755 infra/matrix/dev-down.sh create mode 100755 infra/matrix/dev-up.sh create mode 100644 infra/matrix/docker-compose.dev.yml create mode 100644 infra/matrix/synapse/homeserver.dev.yaml.tpl create mode 100644 infra/matrix/synapse/log.config create mode 100644 packages/comms/README.md create mode 100644 packages/comms/package.json create mode 100644 packages/comms/src/__tests__/heartbeat.test.ts create mode 100644 packages/comms/src/__tests__/liveness.test.ts create mode 100644 packages/comms/src/__tests__/matrix-client.test.ts create mode 100644 packages/comms/src/__tests__/presence-flow.test.ts create mode 100644 packages/comms/src/heartbeat.ts create mode 100644 packages/comms/src/index.ts create mode 100644 packages/comms/src/liveness-reader.ts create mode 100644 packages/comms/src/liveness.ts create mode 100644 packages/comms/src/matrix-client.ts create mode 100644 packages/comms/src/presence-agent.ts create mode 100644 packages/comms/src/types.ts create mode 100644 packages/comms/tsconfig.json create mode 100644 packages/comms/vitest.config.ts create mode 100644 tools/matrix-presence-harness/README.md create mode 100644 tools/matrix-presence-harness/agent-proc.ts create mode 100644 tools/matrix-presence-harness/package.json create mode 100644 tools/matrix-presence-harness/provision.ts create mode 100755 tools/matrix-presence-harness/run.sh create mode 100644 tools/matrix-presence-harness/tsconfig.json create mode 100644 tools/matrix-presence-harness/validate.ts diff --git a/eslint.config.mjs b/eslint.config.mjs index 0433f3a0..bcfe1995 100644 --- a/eslint.config.mjs +++ b/eslint.config.mjs @@ -29,6 +29,7 @@ export default tseslint.config( 'apps/web/playwright.config.ts', 'apps/gateway/vitest.config.ts', 'plugins/discord/vitest.config.ts', + 'packages/comms/vitest.config.ts', 'packages/db/vitest.config.ts', 'packages/storage/vitest.config.ts', 'packages/mosaic/vitest.config.ts', diff --git a/infra/matrix/.gitignore b/infra/matrix/.gitignore new file mode 100644 index 00000000..fc4a97aa --- /dev/null +++ b/infra/matrix/.gitignore @@ -0,0 +1,3 @@ +# DEV runtime state: rendered config, sqlite db, signing key, self-signed +# certs, throwaway secrets. Never committed. +.data/ diff --git a/infra/matrix/README.md b/infra/matrix/README.md new file mode 100644 index 00000000..51b950e6 --- /dev/null +++ b/infra/matrix/README.md @@ -0,0 +1,53 @@ +# infra/matrix — DEV Synapse for RFC-001 P1 (presence) + +> **DEV-ONLY. LOCAL SANDBOX.** This stack is for local presence validation. It +> must **never** be pointed at, or run alongside, a live/production homeserver. +> It is single-instance, federation-OFF, self-signed TLS — the smallest slice +> RFC-002 §9 says P1 needs (Mode B single-domain, no federation, no IP-only, no +> secret-rotation story). + +## What this is + +A single Synapse homeserver rendered from committed **templates** (no hardcoded +topology — RFC-002 G2). Every topology fact is an environment variable with a +dev default: + +| Var | Default | Meaning | +| -------------------- | ------------------ | --------------------------------------------------- | +| `MATRIX_SERVER_NAME` | `matrix.localhost` | Synapse `server_name` (the `:suffix` of every MXID) | +| `MOSAIC_AS_ID` | `mosaic-as` | appservice id / registration filename | +| `MATRIX_HTTP_PORT` | `18008` | host port → Synapse 8008 (plain HTTP) | +| `MATRIX_TLS_PORT` | `18448` | host port → Synapse 8448 (self-signed TLS) | + +Secrets (`as_token`, `hs_token`, Synapse macaroon/form/registration secrets) +are **throwaway values generated at bring-up** into `.data/dev-secrets.env` +(gitignored). In production these are crown-jewel secrets held by the +SecretBackend (RFC-001 §8 / RFC-002 §4) — never committed. + +## Files + +- `docker-compose.dev.yml` — Synapse (+ optional Element under `--profile element`). +- `synapse/homeserver.dev.yaml.tpl` — rendered Synapse config (Mode B, `enable_registration: false`, appservice wired, native TLS). +- `synapse/log.config` — Synapse logging. +- `appservice/mosaic-as.dev.yaml.tpl` — minimal AS registration (declares the `@agent-*` user namespace). +- `dev-up.sh` / `dev-down.sh` — bring up / tear down (`--purge` wipes `.data`). +- `.data/` — **gitignored** runtime state (rendered config, sqlite db, signing key, self-signed certs, dev secrets). + +## Usage + +```bash +./dev-up.sh # render config, gen signing key + TLS cert, boot Synapse +# ... run the validation harness (tools/matrix-presence-harness/run.sh) ... +./dev-down.sh # stop, keep .data +./dev-down.sh --purge # stop and wipe .data for a pristine next boot + +# optional human view (A4) — Element pointed at the dev server: +docker compose -f docker-compose.dev.yml --profile element up -d element +# -> http://127.0.0.1:18080 +``` + +## Acceptance evidence (A1) + +- TLS (self-signed) reachable: `curl -sk https://127.0.0.1:18448/_matrix/client/versions` → `200`. +- Open registration OFF: `POST /_matrix/client/v3/register` → `M_FORBIDDEN "Registration has been disabled"`. +- AS-token registration bypasses the flag by design (that is how agents are provisioned). diff --git a/infra/matrix/appservice/mosaic-as.dev.yaml.tpl b/infra/matrix/appservice/mosaic-as.dev.yaml.tpl new file mode 100644 index 00000000..4fa0014c --- /dev/null +++ b/infra/matrix/appservice/mosaic-as.dev.yaml.tpl @@ -0,0 +1,30 @@ +# ============================================================================ +# mosaic-as.dev.yaml.tpl — Mosaic Appservice registration (DEV) +# ============================================================================ +# +# *** DEV-ONLY. The tokens below are throwaway placeholders rendered at +# bring-up by dev-up.sh. NEVER commit real hs_token/as_token — in +# production they are crown-jewel secrets held by the SecretBackend +# (RFC-001 §8, RFC-002 §4). *** +# +# This is the MINIMAL P1 registration: it declares the @agent-* user +# namespace so the P1 provisioner can register a few virtual agent MXIDs and +# carry heartbeats. It intentionally does NOT model the full P2 taxonomy / +# token-minting appservice. +# +# Rendered by infra/matrix/dev-up.sh (envsubst -> .data/${MOSAIC_AS_ID}.yaml). +# ---------------------------------------------------------------------------- +id: "${MOSAIC_AS_ID}" +url: null # P1: provisioner drives the AS API directly; Synapse pushes no txns. +as_token: "${MOSAIC_AS_TOKEN}" +hs_token: "${MOSAIC_HS_TOKEN}" +sender_localpart: "mosaic-as" +rate_limited: false +namespaces: + users: + - exclusive: true + regex: "@agent-.*:${MATRIX_SERVER_NAME}" + aliases: + - exclusive: false + regex: "#mosaic-.*:${MATRIX_SERVER_NAME}" + rooms: [] diff --git a/infra/matrix/dev-down.sh b/infra/matrix/dev-down.sh new file mode 100755 index 00000000..01eca367 --- /dev/null +++ b/infra/matrix/dev-down.sh @@ -0,0 +1,12 @@ +#!/usr/bin/env bash +# dev-down.sh — tear down the DEV Synapse (matrix-p1-dev). DEV-ONLY. +# ./dev-down.sh # stop + remove containers, keep ./.data +# ./dev-down.sh --purge # also delete ./.data (fresh next bring-up) +set -euo pipefail +HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +docker compose -f "${HERE}/docker-compose.dev.yml" --profile element down -v --remove-orphans || true +if [[ "${1:-}" == "--purge" ]]; then + rm -rf "${HERE}/.data" + echo "[dev-down] purged ${HERE}/.data" +fi +echo "[dev-down] matrix-p1-dev stopped" diff --git a/infra/matrix/dev-up.sh b/infra/matrix/dev-up.sh new file mode 100755 index 00000000..a3e5316c --- /dev/null +++ b/infra/matrix/dev-up.sh @@ -0,0 +1,108 @@ +#!/usr/bin/env bash +# ============================================================================ +# dev-up.sh — bring up the DEV Synapse for RFC-001 P1 (presence) +# ============================================================================ +# +# *** DEV-ONLY. LOCAL SANDBOX. This script must never be pointed at live +# infra. It writes only into infra/matrix/.data (gitignored) and drives +# a dedicated compose project (matrix-p1-dev). *** +# +# It renders the Synapse config + appservice registration from the committed +# templates (envsubst), generates a DEV signing key + self-signed TLS cert, +# and starts Synapse. All topology facts come from the environment with dev +# defaults (RFC-002 G2: nothing hardcoded). +# ---------------------------------------------------------------------------- +set -euo pipefail + +HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +DATA="${HERE}/.data" +COMPOSE=(docker compose -f "${HERE}/docker-compose.dev.yml") + +# ---- Topology inputs (dev defaults; override via env) ---------------------- +export MATRIX_SERVER_NAME="${MATRIX_SERVER_NAME:-matrix.localhost}" +export MOSAIC_AS_ID="${MOSAIC_AS_ID:-mosaic-as}" +export MATRIX_HTTP_PORT="${MATRIX_HTTP_PORT:-18008}" +export MATRIX_TLS_PORT="${MATRIX_TLS_PORT:-18448}" + +# ---- DEV secrets (throwaway; regenerated if absent) ------------------------ +SECRETS_ENV="${DATA}/dev-secrets.env" +mkdir -p "${DATA}" +if [[ ! -f "${SECRETS_ENV}" ]]; then + { + echo "MOSAIC_AS_TOKEN=devas_$(openssl rand -hex 16)" + echo "MOSAIC_HS_TOKEN=devhs_$(openssl rand -hex 16)" + echo "SYNAPSE_REG_SHARED_SECRET=devreg_$(openssl rand -hex 16)" + echo "SYNAPSE_MACAROON_SECRET=devmac_$(openssl rand -hex 16)" + echo "SYNAPSE_FORM_SECRET=devform_$(openssl rand -hex 16)" + } > "${SECRETS_ENV}" + echo "[dev-up] generated throwaway DEV secrets -> ${SECRETS_ENV}" +fi +# shellcheck disable=SC1090 +set -a; source "${SECRETS_ENV}"; set +a + +# DEV: let the synapse container user (uid 991) write the sqlite db / media. +chmod 0777 "${DATA}" || true + +# ---- Render config from templates ------------------------------------------ +envsubst < "${HERE}/synapse/homeserver.dev.yaml.tpl" > "${DATA}/homeserver.yaml" +envsubst < "${HERE}/appservice/mosaic-as.dev.yaml.tpl" > "${DATA}/${MOSAIC_AS_ID}.yaml" +cp "${HERE}/synapse/log.config" "${DATA}/log.config" +echo "[dev-up] rendered homeserver.yaml + ${MOSAIC_AS_ID}.yaml (server_name=${MATRIX_SERVER_NAME})" + +# ---- DEV signing key ------------------------------------------------------- +# Generate as the synapse container user (uid 991) so the running container +# can read it; owned-by-991 mode-600 is fine (the container IS 991). +SIGNING_KEY="${DATA}/${MATRIX_SERVER_NAME}.signing.key" +if [[ ! -f "${SIGNING_KEY}" ]]; then + docker run --rm --user 991:991 -v "${DATA}:/data" --entrypoint generate_signing_key \ + matrixdotorg/synapse:latest -o "/data/${MATRIX_SERVER_NAME}.signing.key" + echo "[dev-up] generated DEV signing key" +fi + +# ---- DEV self-signed TLS cert (A1) ----------------------------------------- +if [[ ! -f "${DATA}/dev-cert.pem" ]]; then + openssl req -x509 -newkey rsa:2048 -nodes \ + -keyout "${DATA}/dev-key.pem" -out "${DATA}/dev-cert.pem" \ + -days 90 -subj "/CN=${MATRIX_SERVER_NAME}" \ + -addext "subjectAltName=DNS:${MATRIX_SERVER_NAME},DNS:localhost,IP:127.0.0.1" >/dev/null 2>&1 + echo "[dev-up] generated DEV self-signed TLS cert for ${MATRIX_SERVER_NAME}" +fi +# Files hermes owns must be world-readable so the synapse (uid 991) container +# can read them (DEV-only; the signing key stays owned by 991 from above). +chmod 0644 "${DATA}/dev-cert.pem" "${DATA}/dev-key.pem" \ + "${DATA}/homeserver.yaml" "${DATA}/${MOSAIC_AS_ID}.yaml" "${DATA}/log.config" || true + +# ---- Element config (optional human view, A4) ------------------------------ +cat > "${DATA}/element-config.json" </dev/null || true)" + if [[ "${status}" == "healthy" ]]; then echo " ... healthy"; break; fi + echo -n "."; sleep 2 +done + +if ! curl -fsS "http://127.0.0.1:${MATRIX_HTTP_PORT}/health" >/dev/null 2>&1; then + echo "[dev-up] ERROR: Synapse did not become healthy" >&2 + "${COMPOSE[@]}" logs --tail 60 synapse >&2 || true + exit 1 +fi + +echo "[dev-up] Synapse UP:" +echo " plain HTTP : http://127.0.0.1:${MATRIX_HTTP_PORT}" +echo " TLS : https://127.0.0.1:${MATRIX_TLS_PORT} (self-signed, DEV)" +echo " server_name: ${MATRIX_SERVER_NAME} registration: OFF" diff --git a/infra/matrix/docker-compose.dev.yml b/infra/matrix/docker-compose.dev.yml new file mode 100644 index 00000000..197270b3 --- /dev/null +++ b/infra/matrix/docker-compose.dev.yml @@ -0,0 +1,60 @@ +# ============================================================================ +# docker-compose.dev.yml — Synapse DEV homeserver for RFC-001 P1 (presence) +# ============================================================================ +# +# *** DEV-ONLY. LOCAL SANDBOX. Do NOT point this at, or run alongside, any +# live/production homeserver. *** +# +# Single-instance Synapse (RFC-002 Mode B, federation OFF) + an optional +# Element web client for the human view (A4). Brought up by ./dev-up.sh, which +# renders config into ./.data (gitignored) first. +# +# Isolated by design: dedicated project name + network + high host ports so it +# never collides with other stacks on the box (A5). +# +# Host ports: ${MATRIX_HTTP_PORT:-18008} -> Synapse 8008 (plain HTTP) +# ${MATRIX_TLS_PORT:-18448} -> Synapse 8448 (self-signed TLS) +# ${ELEMENT_PORT:-18080} -> Element web (profile: element) +# ---------------------------------------------------------------------------- +name: matrix-p1-dev + +services: + synapse: + image: matrixdotorg/synapse:latest + container_name: matrix-p1-synapse + restart: 'no' + # Run against the rendered config in /data (dev-up.sh puts it there). + environment: + SYNAPSE_CONFIG_DIR: /data + SYNAPSE_CONFIG_PATH: /data/homeserver.yaml + volumes: + - ./.data:/data + ports: + - '127.0.0.1:${MATRIX_HTTP_PORT:-18008}:8008' + - '127.0.0.1:${MATRIX_TLS_PORT:-18448}:8448' + networks: + - matrix-p1-net + healthcheck: + test: ['CMD-SHELL', 'curl -fsS http://localhost:8008/health || exit 1'] + interval: 3s + timeout: 5s + retries: 40 + start_period: 5s + + # Optional human view (A4). `docker compose --profile element up -d element`. + element: + image: vectorim/element-web:latest + container_name: matrix-p1-element + restart: 'no' + profiles: [element] + volumes: + - ./.data/element-config.json:/app/config.json:ro + ports: + - '127.0.0.1:${ELEMENT_PORT:-18080}:80' + networks: + - matrix-p1-net + +networks: + matrix-p1-net: + name: matrix-p1-net + driver: bridge diff --git a/infra/matrix/synapse/homeserver.dev.yaml.tpl b/infra/matrix/synapse/homeserver.dev.yaml.tpl new file mode 100644 index 00000000..28f88f1e --- /dev/null +++ b/infra/matrix/synapse/homeserver.dev.yaml.tpl @@ -0,0 +1,84 @@ +# ============================================================================ +# homeserver.dev.yaml.tpl — Synapse DEV homeserver config (RFC-002 Mode B) +# ============================================================================ +# +# *** DEV-ONLY. NOT FOR PRODUCTION. *** +# +# This is the RFC-001 P1 (presence) single-instance homeserver. It is +# RFC-002 "Mode B — single-domain" (server_name == homeserver host), with +# federation OFF (standalone), which is all P1 needs (RFC-002 §9 P1 row). +# +# NOTHING here is a production value. Every topology fact is a variable +# rendered by dev-up.sh from the environment; there is no baked-in fleet +# domain (RFC-002 G2 "zero hardcoded topology"). The secrets below are +# throwaway DEV placeholders rendered at bring-up — never reuse them. +# +# Rendered by: infra/matrix/dev-up.sh (envsubst -> .data/homeserver.yaml) +# Variables: MATRIX_SERVER_NAME, MOSAIC_AS_ID (+ dev secrets) +# ---------------------------------------------------------------------------- + +server_name: "${MATRIX_SERVER_NAME}" +pid_file: /data/homeserver.pid +report_stats: false +suppress_key_server_warning: true + +# --- Listeners ------------------------------------------------------------- +# 8008: plain HTTP (client + federation) for in-container/local tooling. +# 8448: TLS (self-signed in DEV) — satisfies A1 "reachable over TLS". +listeners: + - port: 8008 + type: http + tls: false + bind_addresses: ['0.0.0.0'] + x_forwarded: true + resources: + - names: [client, federation] + compress: false + + - port: 8448 + type: http + tls: true + bind_addresses: ['0.0.0.0'] + resources: + - names: [client, federation] + compress: false + +# --- DEV TLS (self-signed; generated by dev-up.sh) ------------------------- +tls_certificate_path: "/data/dev-cert.pem" +tls_private_key_path: "/data/dev-key.pem" + +# --- Store: sqlite is the simplest dev store (RFC-002 §1 allows it) -------- +database: + name: sqlite3 + args: + database: /data/homeserver.db + +log_config: "/data/log.config" +media_store_path: /data/media_store +signing_key_path: "/data/${MATRIX_SERVER_NAME}.signing.key" + +# --- Hardening (RFC-001 §8 / RFC-002 §8.5), rendered so a dev gets it ------ +# Registration is OFF: agents come ONLY via the appservice (A1). AS user +# registration bypasses this flag, which is exactly the design. +enable_registration: false +enable_registration_without_verification: false +registration_shared_secret: "${SYNAPSE_REG_SHARED_SECRET}" +macaroon_secret_key: "${SYNAPSE_MACAROON_SECRET}" +form_secret: "${SYNAPSE_FORM_SECRET}" + +# Presence EDUs ON so Element shows the native dot for humans (RFC-001 §4.5); +# the AUTHORITATIVE liveness is still the mosaic.presence heartbeat. +presence: + enabled: true + +# --- Federation: OFF for P1 (standalone). Empty whitelist = federate with +# nobody (RFC-002 §2.4 / NG5). No public-network federation. ---------------- +federation_domain_whitelist: [] +trusted_key_servers: [] + +# --- Appservice registration wired in (A1). The file is rendered next to +# this one by dev-up.sh. ----------------------------------------------------- +app_service_config_files: + - "/data/${MOSAIC_AS_ID}.yaml" + +# Keep default rate-limiting ON (RFC-001 §8). No overrides here. diff --git a/infra/matrix/synapse/log.config b/infra/matrix/synapse/log.config new file mode 100644 index 00000000..24d12232 --- /dev/null +++ b/infra/matrix/synapse/log.config @@ -0,0 +1,16 @@ +# Synapse DEV log config. DEV-ONLY. +version: 1 +formatters: + precise: + format: '%(asctime)s - %(name)s - %(lineno)d - %(levelname)s - %(request)s - %(message)s' +handlers: + console: + class: logging.StreamHandler + formatter: precise +loggers: + synapse.storage.SQL: + level: WARNING +root: + level: INFO + handlers: [console] +disable_existing_loggers: false diff --git a/packages/comms/README.md b/packages/comms/README.md new file mode 100644 index 00000000..ef7bd5c6 --- /dev/null +++ b/packages/comms/README.md @@ -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`. diff --git a/packages/comms/package.json b/packages/comms/package.json new file mode 100644 index 00000000..3ca2f244 --- /dev/null +++ b/packages/comms/package.json @@ -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" + ] +} diff --git a/packages/comms/src/__tests__/heartbeat.test.ts b/packages/comms/src/__tests__/heartbeat.test.ts new file mode 100644 index 00000000..bef29f84 --- /dev/null +++ b/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(); + }); +}); diff --git a/packages/comms/src/__tests__/liveness.test.ts b/packages/comms/src/__tests__/liveness.test.ts new file mode 100644 index 00000000..58d8b5e2 --- /dev/null +++ b/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', + }); + }); +}); diff --git a/packages/comms/src/__tests__/matrix-client.test.ts b/packages/comms/src/__tests__/matrix-client.test.ts new file mode 100644 index 00000000..45f8c967 --- /dev/null +++ b/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) => 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'); + }); +}); diff --git a/packages/comms/src/__tests__/presence-flow.test.ts b/packages/comms/src/__tests__/presence-flow.test.ts new file mode 100644 index 00000000..bf226e52 --- /dev/null +++ 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 = {}; + + constructor(private readonly clock: () => number) {} + + async joinRoom(roomId: string): Promise { + return roomId; + } + async setPresence(userId: string, status: PresenceStatus): Promise { + this.presence[userId] = status; + } + async sendHeartbeat( + _roomId: string, + content: { agent: { slug: string; mxid: string }; seq: number; status: PresenceStatus }, + ): Promise { + 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 { + const bySlug = new Map(); + 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'); + }); +}); diff --git a/packages/comms/src/heartbeat.ts b/packages/comms/src/heartbeat.ts new file mode 100644 index 00000000..dd798dca --- /dev/null +++ b/packages/comms/src/heartbeat.ts @@ -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; + +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)); + + 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); + }, + }; +} diff --git a/packages/comms/src/index.ts b/packages/comms/src/index.ts new file mode 100644 index 00000000..ba4c2cc0 --- /dev/null +++ b/packages/comms/src/index.ts @@ -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'; diff --git a/packages/comms/src/liveness-reader.ts b/packages/comms/src/liveness-reader.ts new file mode 100644 index 00000000..58211152 --- /dev/null +++ b/packages/comms/src/liveness-reader.ts @@ -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 { + 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 { + const rows = await this.read(); + rows.sort((a, b) => a.slug.localeCompare(b.slug)); + const dot: Record = { 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'); + } +} diff --git a/packages/comms/src/liveness.ts b/packages/comms/src/liveness.ts new file mode 100644 index 00000000..18494c58 --- /dev/null +++ b/packages/comms/src/liveness.ts @@ -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, + }; + }); +} diff --git a/packages/comms/src/matrix-client.ts b/packages/comms/src/matrix-client.ts new file mode 100644 index 00000000..67f7bdd3 --- /dev/null +++ b/packages/comms/src/matrix-client.ts @@ -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; body?: unknown } = {}, + ): Promise> { + 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; + 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 { + 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 { + 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, + ): Promise { + 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 { + return this.sendEvent(roomId, 'm.room.message', content as unknown as Record); + } + + /** Join a room (by id or alias). Idempotent on the server. */ + async joinRoom(roomIdOrAlias: string): Promise { + 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 { + 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>) : []; + const bySlug = new Map(); + + for (const ev of chunk) { + const content = ev.content as Record | undefined; + if (!content || content.msgtype !== 'mosaic.presence') continue; + const agent = content.agent as Record | 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()]; + } +} diff --git a/packages/comms/src/presence-agent.ts b/packages/comms/src/presence-agent.ts new file mode 100644 index 00000000..fbd7e20a --- /dev/null +++ b/packages/comms/src/presence-agent.ts @@ -0,0 +1,92 @@ +/** + * High-level presence agent (RFC-001 §4.1 steps 10–11, §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 { + 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 { + this.pauseHeartbeat(); + try { + await this.opts.client.setPresence(this.opts.agent.mxid, 'offline'); + } catch (err) { + this.opts.onError?.(err); + } + } +} diff --git a/packages/comms/src/types.ts b/packages/comms/src/types.ts new file mode 100644 index 00000000..49605258 --- /dev/null +++ b/packages/comms/src/types.ts @@ -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'; diff --git a/packages/comms/tsconfig.json b/packages/comms/tsconfig.json new file mode 100644 index 00000000..c9733863 --- /dev/null +++ b/packages/comms/tsconfig.json @@ -0,0 +1,9 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "outDir": "dist", + "rootDir": "src" + }, + "include": ["src/**/*"], + "exclude": ["node_modules", "dist"] +} diff --git a/packages/comms/vitest.config.ts b/packages/comms/vitest.config.ts new file mode 100644 index 00000000..b27ea14b --- /dev/null +++ b/packages/comms/vitest.config.ts @@ -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'], + }, + }, +}); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 33300ce9..6862d878 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -372,6 +372,21 @@ importers: specifier: ^2.0.0 version: 2.1.9(@types/node@24.12.0)(jsdom@29.0.0(@noble/hashes@2.0.1))(lightningcss@1.31.1) + packages/comms: + devDependencies: + '@types/node': + specifier: ^22.0.0 + version: 22.19.15 + '@vitest/coverage-v8': + specifier: ^2.0.0 + version: 2.1.9(vitest@2.1.9(@types/node@22.19.15)(jsdom@29.0.0(@noble/hashes@2.0.1))(lightningcss@1.31.1)) + typescript: + specifier: ^5.8.0 + version: 5.9.3 + vitest: + specifier: ^2.0.0 + version: 2.1.9(@types/node@22.19.15)(jsdom@29.0.0(@noble/hashes@2.0.1))(lightningcss@1.31.1) + packages/config: dependencies: '@mosaicstack/memory': @@ -796,6 +811,28 @@ importers: specifier: ^2.0.0 version: 2.1.9(@types/node@24.12.0)(jsdom@29.0.0(@noble/hashes@2.0.1))(lightningcss@1.31.1) + tools/matrix-presence-harness: + dependencies: + '@mosaicstack/appservice': + specifier: workspace:* + version: link:../../packages/appservice + '@mosaicstack/comms': + specifier: workspace:* + version: link:../../packages/comms + devDependencies: + '@types/node': + specifier: ^22.0.0 + version: 22.19.15 + tsx: + specifier: ^4.19.0 + version: 4.21.0 + typescript: + specifier: ^5.8.0 + version: 5.9.3 + vitest: + specifier: ^2.0.0 + version: 2.1.9(@types/node@22.19.15)(jsdom@29.0.0(@noble/hashes@2.0.1))(lightningcss@1.31.1) + packages: '@agentclientprotocol/sdk@0.17.0': diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index 1e3145cf..abb7a60e 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -2,6 +2,7 @@ packages: - 'apps/*' - 'packages/*' - 'plugins/*' + - 'tools/matrix-presence-harness' ignoredBuiltDependencies: - '@nestjs/core' diff --git a/tools/matrix-presence-harness/README.md b/tools/matrix-presence-harness/README.md new file mode 100644 index 00000000..62210831 --- /dev/null +++ b/tools/matrix-presence-harness/README.md @@ -0,0 +1,43 @@ +# tools/matrix-presence-harness — RFC-001 P1 validation (A1–A5) + +> **DEV-ONLY. LOCAL SANDBOX.** Drives the local `infra/matrix` Synapse only. +> Never point it at live infra. + +The dev validation harness for the presence slice. It is the **minimal P1 +provisioner** + an end-to-end demo that proves the acceptance criteria against +a real Synapse. + +## Pieces + +- `provision.ts` — the **minimal provisioner**: registers ≥3 agent MXIDs, creates + the fleet presence room, joins the agents. Reuses the existing tested + `@mosaicstack/appservice` intent library (register / createRoom / join). It is + deliberately **not** the P2 appservice (no auto-enroll / taxonomy / token minting). +- `agent-proc.ts` — one presence agent as its **own OS process** (via + `@mosaicstack/comms`): joins the fleet room and heartbeats `mosaic.presence`. + Standalone so the harness can `SIGKILL` it for a true hard-kill (A3). +- `validate.ts` — provisions, spawns the agents, then asserts: + - **A2** ≥3 agents show **online** in the fleet room. + - **A3** a `SIGKILL`'d agent flips to **offline within `dark_threshold`**, + deterministically (heartbeat-age based, not native-presence timeout), while + survivors stay online. + - **A4** prints the human-readable fleet liveness board. +- `run.sh` — wires dev secrets + env and runs `validate.ts` over the self-signed + TLS endpoint (A1). Exits non-zero on any failed assertion. + +## Usage + +```bash +../../infra/matrix/dev-up.sh # bring Synapse up first +./run.sh # provision + validate (A2/A3/A4) +``` + +Tunables (env): `HEARTBEAT_INTERVAL_MS` (1000), `MISS_TOLERANCE` (2), +`DARK_THRESHOLD_MS` (6000), `AGENT_SLUGS` (`alpha,bravo,charlie`), +`VICTIM_SLUG` (`charlie`), `MATRIX_CS_URL` (defaults to the TLS endpoint). + +## Note on `NODE_TLS_REJECT_UNAUTHORIZED=0` + +`run.sh` sets this **only** because the dev homeserver uses a self-signed cert. +It is a dev convenience for exercising the SDK over TLS and must never be used +against a real CA-issued endpoint. diff --git a/tools/matrix-presence-harness/agent-proc.ts b/tools/matrix-presence-harness/agent-proc.ts new file mode 100644 index 00000000..41a3134c --- /dev/null +++ b/tools/matrix-presence-harness/agent-proc.ts @@ -0,0 +1,70 @@ +/** + * tools/matrix-presence-harness/agent-proc.ts + * + * *** DEV harness — LOCAL SANDBOX ONLY. *** + * + * Runs ONE presence agent as its own OS process: joins the fleet presence + * room and heartbeats `mosaic.presence` via @mosaicstack/comms (RFC-001 P1). + * Kept as a standalone process so the validation harness can `kill -9` it to + * prove A3 (hard-kill -> offline within dark_threshold) against a real crash, + * not a graceful shutdown. + * + * Auth in DEV is Application-Service masquerade: the process presents the + * as_token and acts as its own @agent- MXID. (Per-agent minted tokens + * are P2, RFC-001 §2.2/§8.) + * + * ESM / NodeNext: .js import extensions. + */ +import { MinimalMatrixClient, PresenceAgent, type LivenessPolicy } from '@mosaicstack/comms'; + +const env = (k: string, fallback?: string): string => { + const v = process.env[k] ?? fallback; + if (v === undefined) throw new Error(`missing env ${k}`); + return v; +}; + +async function main(): Promise { + const homeserverUrl = env('MATRIX_CS_URL'); + const asToken = env('MOSAIC_AS_TOKEN'); + const serverName = env('MATRIX_SERVER_NAME'); + const slug = env('AGENT_SLUG'); + const roomId = env('FLEET_ROOM_ID'); + const intervalMs = Number(env('HEARTBEAT_INTERVAL_MS', '1000')); + + const mxid = `@agent-${slug}:${serverName}`; + const policy: LivenessPolicy = { + heartbeatIntervalMs: intervalMs, + missTolerance: Number(env('MISS_TOLERANCE', '2')), + darkThresholdMs: Number(env('DARK_THRESHOLD_MS', '6000')), + }; + + const client = new MinimalMatrixClient({ + homeserverUrl, + accessToken: asToken, + actAsUserId: mxid, + }); + const agent = new PresenceAgent({ + client, + agent: { mxid, slug, harness: 'claude-code' }, + roomId, + intervalMs, + policy, + onError: (err) => console.error(`[${slug}] heartbeat error:`, (err as Error).message), + }); + + await agent.connect(); + agent.start(); + console.log(`[${slug}] pid=${process.pid} mxid=${mxid} heartbeating every ${intervalMs}ms`); + + // Graceful stop only on SIGTERM; the A3 test uses SIGKILL (no cleanup runs). + process.on('SIGTERM', () => { + void agent.stop().finally(() => process.exit(0)); + }); + // Keep the event loop alive indefinitely. + setInterval(() => {}, 1 << 30); +} + +main().catch((err) => { + console.error('agent-proc fatal:', err); + process.exit(1); +}); diff --git a/tools/matrix-presence-harness/package.json b/tools/matrix-presence-harness/package.json new file mode 100644 index 00000000..848e1679 --- /dev/null +++ b/tools/matrix-presence-harness/package.json @@ -0,0 +1,24 @@ +{ + "name": "@mosaicstack/matrix-presence-harness", + "version": "0.0.1", + "type": "module", + "private": true, + "description": "DEV-ONLY validation harness for RFC-001 P1 (Matrix presence). Not shipped.", + "scripts": { + "build": "tsc --noEmit", + "lint": "eslint .", + "typecheck": "tsc --noEmit", + "test": "vitest run --passWithNoTests", + "validate": "./run.sh" + }, + "dependencies": { + "@mosaicstack/appservice": "workspace:*", + "@mosaicstack/comms": "workspace:*" + }, + "devDependencies": { + "@types/node": "^22.0.0", + "tsx": "^4.19.0", + "typescript": "^5.8.0", + "vitest": "^2.0.0" + } +} diff --git a/tools/matrix-presence-harness/provision.ts b/tools/matrix-presence-harness/provision.ts new file mode 100644 index 00000000..6f49c59f --- /dev/null +++ b/tools/matrix-presence-harness/provision.ts @@ -0,0 +1,89 @@ +/** + * tools/matrix-presence-harness/provision.ts + * + * *** DEV harness — LOCAL SANDBOX ONLY. *** + * + * The MINIMAL P1 provisioner (RFC-001 §10 P1): its ONLY job is to register a + * few agent MXIDs, create the fleet presence room, and join the agents. It is + * deliberately NOT the P2 appservice (no auto-enroll, no room taxonomy, no + * token minting, no introductions). + * + * It reuses the existing, tested `@mosaicstack/appservice` core library + * (AppserviceIntent: register / createRoom / join over the AS API) rather than + * reinventing Matrix plumbing. + * + * ESM / NodeNext: .js import extensions. + */ +import { AppserviceIntent, type AppserviceConfig } from '@mosaicstack/appservice'; + +export interface ProvisionOptions { + homeserverUrl: string; + serverName: string; + asToken: string; + hsToken: string; + agentSlugs: string[]; + fleetAlias?: string; // localpart, default "mosaic-fleet" +} + +export interface ProvisionResult { + roomId: string; + senderUserId: string; + agents: { slug: string; mxid: string }[]; +} + +/** Register agents, create the fleet presence room, join everyone. Idempotent. */ +export async function provision(opts: ProvisionOptions): Promise { + const cfg: AppserviceConfig = { + homeserverUrl: opts.homeserverUrl, + domain: opts.serverName, + asToken: opts.asToken, + hsToken: opts.hsToken, + }; + const intent = new AppserviceIntent(cfg); + const fleetAlias = opts.fleetAlias ?? 'mosaic-fleet'; + + // 1. Register the virtual agent MXIDs (bypasses enable_registration:false). + const agents = []; + for (const slug of opts.agentSlugs) { + const mxid = await intent.ensureRegistered(slug); + await intent.setDisplayName(slug, `agent ${slug} (DEV)`); + agents.push({ slug, mxid }); + } + + // 2. Create the single fleet presence room, inviting all agents (RFC-001 §4.6). + // Idempotent: if the alias already exists, reuse that room instead. + let roomId: string; + try { + const created = await intent.createRoom({ + name: 'Fleet Presence (DEV)', + alias: fleetAlias, + topic: 'RFC-001 P1 — mosaic.presence heartbeats. Who is alive?', + invite: agents.map((a) => a.mxid), + }); + roomId = created.roomId; + } catch (err) { + // Alias already taken (re-run): resolve the existing room id. + const aliasFq = `#${fleetAlias}:${opts.serverName}`; + roomId = await resolveAlias(opts, aliasFq); + void err; + } + + // 3. Join every agent into the fleet room (invite + join; idempotent). + for (const { slug } of agents) { + await intent.ensureJoined(roomId, slug); + } + + return { roomId, senderUserId: intent.senderUserId, agents }; +} + +async function resolveAlias( + opts: Pick, + aliasFq: string, +): Promise { + const url = `${opts.homeserverUrl.replace(/\/$/, '')}/_matrix/client/v3/directory/room/${encodeURIComponent(aliasFq)}`; + const res = await fetch(url, { headers: { Authorization: `Bearer ${opts.asToken}` } }); + if (!res.ok) throw new Error(`resolveAlias ${aliasFq} -> ${res.status}`); + const data = (await res.json()) as { room_id?: string }; + if (!data.room_id) throw new Error(`resolveAlias ${aliasFq} returned no room_id`); + return data.room_id; +} diff --git a/tools/matrix-presence-harness/run.sh b/tools/matrix-presence-harness/run.sh new file mode 100755 index 00000000..1e2c33f3 --- /dev/null +++ b/tools/matrix-presence-harness/run.sh @@ -0,0 +1,47 @@ +#!/usr/bin/env bash +# ============================================================================ +# run.sh — RFC-001 P1 presence validation harness (A1–A5) *** DEV-ONLY *** +# ============================================================================ +# +# Assumes the dev Synapse is already up (infra/matrix/dev-up.sh). Provisions 3 +# agents, heartbeats them, proves A2/A3/A4 against the LOCAL dev homeserver, +# exercising @mosaicstack/comms over the self-signed TLS endpoint (A1). +# +# NODE_TLS_REJECT_UNAUTHORIZED=0 is set ONLY because the dev cert is +# self-signed. This is a DEV convenience and must never be used in prod. +# ---------------------------------------------------------------------------- +set -euo pipefail +HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +REPO="$(cd "${HERE}/../.." && pwd)" +DATA="${REPO}/infra/matrix/.data" + +if [[ ! -f "${DATA}/dev-secrets.env" ]]; then + echo "run.sh: ${DATA}/dev-secrets.env not found — run infra/matrix/dev-up.sh first" >&2 + exit 1 +fi +# shellcheck disable=SC1091 +set -a; source "${DATA}/dev-secrets.env"; set +a + +export MATRIX_SERVER_NAME="${MATRIX_SERVER_NAME:-matrix.localhost}" +export MATRIX_TLS_PORT="${MATRIX_TLS_PORT:-18448}" +# Exercise the SDK over the self-signed TLS listener (A1). Use the plain HTTP +# port instead by exporting MATRIX_CS_URL=http://127.0.0.1:18008 before running. +export MATRIX_CS_URL="${MATRIX_CS_URL:-https://127.0.0.1:${MATRIX_TLS_PORT}}" +export NODE_TLS_REJECT_UNAUTHORIZED=0 + +export HEARTBEAT_INTERVAL_MS="${HEARTBEAT_INTERVAL_MS:-1000}" +export MISS_TOLERANCE="${MISS_TOLERANCE:-2}" +export DARK_THRESHOLD_MS="${DARK_THRESHOLD_MS:-6000}" +export AGENT_SLUGS="${AGENT_SLUGS:-alpha,bravo,charlie}" +export VICTIM_SLUG="${VICTIM_SLUG:-charlie}" + +TSX_CLI="$(ls -d "${REPO}"/node_modules/.pnpm/tsx@*/node_modules/tsx/dist/cli.mjs 2>/dev/null | head -1)" +if [[ -z "${TSX_CLI}" ]]; then + echo "run.sh: tsx not found under node_modules — run pnpm install first" >&2 + exit 1 +fi +export TSX_CLI + +echo "[run] CS=${MATRIX_CS_URL} server_name=${MATRIX_SERVER_NAME} dark_threshold=${DARK_THRESHOLD_MS}ms" +cd "${REPO}" +exec node "${TSX_CLI}" "${HERE}/validate.ts" diff --git a/tools/matrix-presence-harness/tsconfig.json b/tools/matrix-presence-harness/tsconfig.json new file mode 100644 index 00000000..02cd267d --- /dev/null +++ b/tools/matrix-presence-harness/tsconfig.json @@ -0,0 +1,9 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "noEmit": true, + "rootDir": "." + }, + "include": ["*.ts"], + "exclude": ["node_modules"] +} diff --git a/tools/matrix-presence-harness/validate.ts b/tools/matrix-presence-harness/validate.ts new file mode 100644 index 00000000..54d6604e --- /dev/null +++ b/tools/matrix-presence-harness/validate.ts @@ -0,0 +1,210 @@ +/** + * tools/matrix-presence-harness/validate.ts + * + * *** DEV harness — LOCAL SANDBOX ONLY. Never point at live infra. *** + * + * End-to-end proof of RFC-001 P1 acceptance criteria against the local dev + * Synapse (infra/matrix): + * + * A2 provision >=3 agents, they heartbeat, all show ONLINE in the fleet + * presence room. + * A3 hard-kill (SIGKILL) one agent's process; assert it flips to OFFLINE + * within dark_threshold, DETERMINISTICALLY (heartbeat-age based, not + * native-presence-timeout), while the survivors stay ONLINE. + * A4 dump the human-readable fleet liveness board. + * + * Exits 0 on success, 1 on any failed assertion. + * ESM / NodeNext: .js import extensions. + */ +import { spawn, type ChildProcess } from 'node:child_process'; +import { fileURLToPath } from 'node:url'; +import { dirname, resolve } from 'node:path'; + +import { + FleetLivenessReader, + MinimalMatrixClient, + type AgentLiveness, + type LivenessPolicy, +} from '@mosaicstack/comms'; + +import { provision } from './provision.js'; + +const HERE = dirname(fileURLToPath(import.meta.url)); + +const env = (k: string, fallback?: string): string => { + const v = process.env[k] ?? fallback; + if (v === undefined) throw new Error(`missing env ${k}`); + return v; +}; + +const sleep = (ms: number): Promise => new Promise((r) => setTimeout(r, ms)); +const board = (rows: AgentLiveness[]): Record => + Object.fromEntries(rows.map((r) => [r.slug, r.status])); + +async function main(): Promise { + const homeserverUrl = env('MATRIX_CS_URL'); + const asToken = env('MOSAIC_AS_TOKEN'); + const hsToken = env('MOSAIC_HS_TOKEN', 'unused-in-p1'); + const serverName = env('MATRIX_SERVER_NAME'); + const intervalMs = Number(env('HEARTBEAT_INTERVAL_MS', '1000')); + const missTolerance = Number(env('MISS_TOLERANCE', '2')); + const darkThresholdMs = Number(env('DARK_THRESHOLD_MS', '6000')); + const slugs = env('AGENT_SLUGS', 'alpha,bravo,charlie').split(','); + const victim = env('VICTIM_SLUG', 'charlie'); + + const policy: LivenessPolicy = { + heartbeatIntervalMs: intervalMs, + missTolerance, + darkThresholdMs, + }; + const failures: string[] = []; + const assert = (ok: boolean, msg: string): void => { + console.log(` ${ok ? 'PASS' : 'FAIL'} ${msg}`); + if (!ok) failures.push(msg); + }; + + // ── Provision (register agents, create fleet room, join) ────────────────── + console.log(`\n[validate] provisioning ${slugs.length} agents + fleet presence room...`); + const { roomId, senderUserId } = await provision({ + homeserverUrl, + serverName, + asToken, + hsToken, + agentSlugs: slugs, + }); + console.log(`[validate] fleet room ${roomId} (sender ${senderUserId})`); + + // Reader masquerades as the AS sender (a room member) to read heartbeats. + const reader = new FleetLivenessReader({ + client: new MinimalMatrixClient({ + homeserverUrl, + accessToken: asToken, + actAsUserId: senderUserId, + }), + roomId, + policy, + }); + + // ── Spawn each agent as its own OS process ──────────────────────────────── + const procs = new Map(); + for (const slug of slugs) { + // `--import tsx` runs agent-proc.ts IN-PROCESS (no tsx grandchild), so the + // spawned pid IS the agent — a SIGKILL to it is a true hard-kill (A3). + // cwd=HERE so the `tsx` specifier resolves against the harness node_modules. + const child = spawn('node', ['--import', 'tsx', resolve(HERE, 'agent-proc.ts')], { + cwd: HERE, + env: { + ...process.env, + AGENT_SLUG: slug, + FLEET_ROOM_ID: roomId, + MATRIX_CS_URL: homeserverUrl, + MOSAIC_AS_TOKEN: asToken, + MATRIX_SERVER_NAME: serverName, + HEARTBEAT_INTERVAL_MS: String(intervalMs), + MISS_TOLERANCE: String(missTolerance), + DARK_THRESHOLD_MS: String(darkThresholdMs), + }, + stdio: ['ignore', 'inherit', 'inherit'], + }); + procs.set(slug, child); + } + + const hardKill = (pid: number): void => { + try { + process.kill(pid, 'SIGKILL'); // pid is the in-process agent — a true kill + } catch { + /* already gone */ + } + }; + const cleanup = (): void => { + for (const [, c] of procs) { + if (c.pid) hardKill(c.pid); + } + }; + process.on('exit', cleanup); + + try { + // ── A2: let a few beats flow, assert all online ───────────────────────── + console.log(`\n[validate] A2 — waiting for heartbeats (${intervalMs}ms interval)...`); + await sleep(intervalMs * 4 + 1000); + const a2 = await reader.read(); + console.log('[validate] board:', board(a2)); + for (const slug of slugs) { + assert(a2.find((r) => r.slug === slug)?.status === 'online', `A2 ${slug} is online`); + } + assert(a2.length >= 3, 'A2 >=3 agents present in fleet room'); + + // ── A4: human-readable board ──────────────────────────────────────────── + console.log('\n[validate] A4 — fleet liveness board (human view):'); + console.log(await reader.formatBoard()); + + // ── A3: hard-kill the victim, measure flip-to-offline latency ─────────── + const victimProc = procs.get(victim); + if (!victimProc?.pid) throw new Error(`no pid for victim ${victim}`); + console.log( + `\n[validate] A3 — SIGKILL ${victim} (pid ${victimProc.pid}); dark_threshold=${darkThresholdMs}ms`, + ); + const killTs = Date.now(); + hardKill(victimProc.pid); + + let victimOffline: AgentLiveness | undefined; + let detectMs = 0; + const deadline = killTs + darkThresholdMs + 6000; + while (Date.now() < deadline) { + await sleep(400); + const rows = await reader.read(); + const v = rows.find((r) => r.slug === victim); + const elapsed = Date.now() - killTs; + if (v) { + console.log( + ` t+${String(elapsed).padStart(5)}ms ${victim}=${v.status} (age=${v.ageMs}ms) ` + + `survivors=${slugs + .filter((s) => s !== victim) + .map((s) => `${s}:${rows.find((r) => r.slug === s)?.status}`) + .join(',')}`, + ); + if (v.status === 'offline') { + victimOffline = v; + detectMs = elapsed; + break; + } + } + } + + assert(!!victimOffline, `A3 ${victim} flipped to offline after kill`); + if (victimOffline) { + // Deterministic: the flip is driven by heartbeat age reaching dark_threshold. + assert( + victimOffline.ageMs >= darkThresholdMs, + `A3 flip is heartbeat-deterministic (age ${victimOffline.ageMs}ms >= dark_threshold ${darkThresholdMs}ms)`, + ); + assert( + detectMs <= darkThresholdMs + 2000, + `A3 detected within dark_threshold + poll slack (detected at t+${detectMs}ms)`, + ); + } + // Survivors unaffected. + const finalRows = await reader.read(); + for (const slug of slugs.filter((s) => s !== victim)) { + assert( + finalRows.find((r) => r.slug === slug)?.status === 'online', + `A3 survivor ${slug} still online`, + ); + } + + console.log('\n[validate] final board after kill:'); + console.log(await reader.formatBoard()); + } finally { + cleanup(); + } + + console.log( + `\n[validate] ${failures.length === 0 ? 'ALL PASSED ✅' : `FAILED ❌ (${failures.length})`}`, + ); + process.exit(failures.length === 0 ? 0 : 1); +} + +main().catch((err) => { + console.error('[validate] fatal:', err); + process.exit(1); +});