stack/packages/comms/src/heartbeat.ts

/**
 * `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);
    },
  };
}