cambium ▸ docs/superpowers/plans/2026-04-11-cambium-rhythms.md
updated 2026-04-12

Cambium Rhythms v1 Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Ship @cambium/rhythms v1 — a native scheduling substrate that turns time into signals on Cambium’s existing signal bus, proven end-to-end with a minimal Obadiah morning.pulse rhythm that posts a “good morning” message to Discord.

Architecture: New workspace package @cambium/rhythms. Rhythm templates (YAML) ship with deployments and seed into rhythm_instances in Postgres on boot. A tick loop integrated into the existing LifecycleEngine via a custom PhaseHandler.sense() implementation queries for due rhythms every ~30s and emits their signal through SignalBus.createAndEmit(). Signals are persisted before delivery (Cambium invariant) and consumed by receptors registered against rhythm signal names. No new daemon; no new tick loop infrastructure — rhythms ride on the lifecycle engine that already exists.

Tech Stack: TypeScript strict, ESM, Drizzle ORM, yaml package, Vitest, Postgres. Depends on @cambium/core, @cambium/db, @cambium/signals, @cambium/lifecycle.

Scope adjustment from spec: Exploration revealed that the existing deployments/obadiah/workflows/morning-brief.yaml is a Claude-Code-readable format, not a Hypha GraphExecutor spec. Rewriting that workflow into something Hypha can execute natively is out of scope for Rhythms v1 and belongs to the morning-brief workflow owner, not to the Rhythms package. The v1 success criterion is therefore narrowed to: rhythm fires → signal persists → test receptor receives → minimal Discord post lands. The full morning brief gets rebound as a follow-up when the morning-brief workflow is migrated to Hypha.

Deferred items (see .planning/seeds/SEED-001-rhythm-entrainment.md and .planning/backlog/BACKLOG.md): - Entrainment (environmental phase-shift) — schema-reserved, not implemented - skip and coalesce catchup policies — enum-reserved, not implemented - Loom Rhythms UI page — headless v1 - Cron expression support — only daily, weekly, hourly, interval:<seconds> in v1

File Structure

New files:

packages/rhythms/
├── package.json
├── tsconfig.json
├── vitest.config.ts
└── src/
    ├── index.ts              — barrel export
    ├── types.ts              — RhythmTemplate, RhythmInstance, CatchupPolicy, branded IDs
    ├── phase.ts              — computeNextFire() phase math
    ├── phase.test.ts
    ├── template-loader.ts    — load YAML templates from disk
    ├── template-loader.test.ts
    ├── rhythm-store.ts       — PgRhythmStore CRUD over rhythm_instances and rhythm_fires
    ├── rhythm-store.test.ts
    ├── seed.ts               — idempotent template → instance upsert
    ├── seed.test.ts
    ├── tick.ts               — createRhythmsHandler() returning a PhaseHandler.sense() impl
    └── tick.test.ts

packages/rhythms/test/fixtures/
└── morning-pulse.yaml        — test template fixture

tests/integration/
└── rhythms-end-to-end.test.ts  — full in-memory path test

deployments/obadiah/rhythms/
└── morning-pulse.yaml        — real production template

deployments/obadiah/bridge/
└── start-rhythms.ts          — CLI entry to boot rhythms in Obadiah

Modified files:

packages/db/src/schema.ts     — add rhythmInstances, rhythmFires table definitions
packages/db/drizzle/           — new generated migration SQL
packages/db/src/stores/index.ts  — export PgRhythmStore (if put there) — or keep in @cambium/rhythms
pnpm-workspace.yaml or turbo.json — may need registration of new package (check existing pattern)

Task 1: Scaffold @cambium/rhythms Package

Files: - Create: packages/rhythms/package.json - Create: packages/rhythms/tsconfig.json - Create: packages/rhythms/vitest.config.ts - Create: packages/rhythms/src/index.ts

Create packages/rhythms/package.json:

{
  "name": "@cambium/rhythms",
  "version": "0.0.1",
  "private": true,
  "type": "module",
  "main": "./dist/index.js",
  "types": "./dist/index.d.ts",
  "exports": {
    ".": {
      "types": "./dist/index.d.ts",
      "import": "./dist/index.js"
    }
  },
  "scripts": {
    "build": "tsc",
    "typecheck": "tsc --noEmit",
    "test": "vitest run",
    "dev": "tsc --watch"
  },
  "dependencies": {
    "@cambium/core": "workspace:*",
    "@cambium/db": "workspace:*",
    "@cambium/signals": "workspace:*",
    "@cambium/lifecycle": "workspace:*",
    "drizzle-orm": "catalog:",
    "yaml": "catalog:"
  },
  "devDependencies": {
    "typescript": "catalog:",
    "vitest": "catalog:"
  }
}

Note: if the monorepo uses explicit version pins instead of catalog:, copy the exact versions from packages/events/package.json. Check first with cat packages/events/package.json and match that file’s dep style.

Create packages/rhythms/tsconfig.json:

{
  "extends": "../../tsconfig.base.json",
  "compilerOptions": {
    "outDir": "dist",
    "rootDir": "src"
  },
  "include": ["src"],
  "exclude": ["src/**/*.test.ts", "test"]
}

Create packages/rhythms/vitest.config.ts:

import { defineConfig } from 'vitest/config';

export default defineConfig({
  test: {
    globals: true,
    environment: 'node',
    include: ['src/**/*.test.ts'],
  },
});

Create packages/rhythms/src/index.ts:

// @cambium/rhythms — nature-based scheduling substrate
// Exports added as implementation lands.
export {};

Run from repo root:

pnpm install
pnpm --filter @cambium/rhythms build
pnpm --filter @cambium/rhythms typecheck

Expected: both commands exit 0. A dist/ directory appears under packages/rhythms/ containing index.js and index.d.ts.

git add packages/rhythms/package.json packages/rhythms/tsconfig.json packages/rhythms/vitest.config.ts packages/rhythms/src/index.ts
git commit -m "feat(rhythms): scaffold @cambium/rhythms package"

Task 2: Define Core Types

Files: - Create: packages/rhythms/src/types.ts - Modify: packages/rhythms/src/index.ts

Create packages/rhythms/src/types.ts:

import type { SignalClass, WorkspaceId } from '@cambium/core';

// Branded ID for rhythm instances. Follows the Cambium pattern
// (see @cambium/core — createId<T>() produces branded IDs).
export type RhythmId = string & { readonly __brand: 'RhythmId' };
export type RhythmFireId = string & { readonly __brand: 'RhythmFireId' };

/**
 * Supported periods for v1.
 * - "daily" — fires once per day at the configured phase time
 * - "weekly" — fires once per week on the configured weekday + time
 * - "hourly" — fires once per hour at the configured minute
 * - "interval:<seconds>" — fires every N seconds (test/short-cycle use)
 */
export type RhythmPeriod =
  | 'daily'
  | 'weekly'
  | 'hourly'
  | `interval:${number}`;

/**
 * Catchup policy when a fire window is missed (e.g. machine was asleep).
 * v1 only implements `fire_once`. The other two are schema-reserved and will
 * throw if a template requests them.
 */
export type CatchupPolicy = 'fire_once' | 'skip' | 'coalesce';

/**
 * RhythmTemplate — the declarative shape that ships with a deployment.
 * Loaded from deployments/<name>/rhythms/*.yaml.
 */
export interface RhythmTemplate {
  name: string;                    // e.g. "morning.pulse" — unique within a deployment
  period: RhythmPeriod;
  phase: string;                   // "06:00", "monday 06:00", "15" (for hourly minute), or "" for intervals
  timezone: string;                // IANA tz, e.g. "Europe/Berlin"
  signal: {
    type: string;                  // signal type emitted, e.g. "morning.pulse"
    class: SignalClass;            // Cambium signal class — usually "lifecycle" for rhythms
    intensity: number;             // default 1.0
    decay_rate: number;            // seconds until expired; 0 = never decays
    payload?: Record<string, unknown>;
    affinity?: string[];
  };
  catchup: CatchupPolicy;
}

/**
 * RhythmInstance — the living row in the DB for an active rhythm.
 */
export interface RhythmInstance {
  id: RhythmId;
  workspace_id: WorkspaceId;
  template_name: string;
  period: RhythmPeriod;
  phase: string;
  timezone: string;
  signal_type: string;
  signal_class: SignalClass;
  signal_intensity: number;
  signal_decay_rate: number;
  signal_payload: Record<string, unknown>;
  signal_affinity: string[];
  catchup_policy: CatchupPolicy;
  entrainment: Record<string, unknown> | null;  // reserved for later; always null in v1
  state: 'active' | 'paused' | 'quarantined';
  last_fired_at: Date | null;
  next_fire_at: Date;
  created_at: Date;
  updated_at: Date;
}

export interface RhythmFire {
  id: RhythmFireId;
  rhythm_id: RhythmId;
  intended_at: Date;
  actual_at: Date;
  signal_id: string;
  was_catchup: boolean;
}

Update packages/rhythms/src/index.ts:

// @cambium/rhythms — nature-based scheduling substrate
export type {
  RhythmId,
  RhythmFireId,
  RhythmPeriod,
  CatchupPolicy,
  RhythmTemplate,
  RhythmInstance,
  RhythmFire,
} from './types.js';

pnpm --filter @cambium/rhythms typecheck

Expected: exits 0.

git add packages/rhythms/src/types.ts packages/rhythms/src/index.ts
git commit -m "feat(rhythms): define core types (RhythmTemplate, RhythmInstance, CatchupPolicy)"

Task 3: Phase Math — computeNextFire

Files: - Create: packages/rhythms/src/phase.ts - Create: packages/rhythms/src/phase.test.ts

This task is pure TDD — no DB, no signals, just math. Start with the failing test.

Create packages/rhythms/src/phase.test.ts:

import { describe, it, expect } from 'vitest';
import { computeNextFire, parseInterval } from './phase.js';

describe('parseInterval', () => {
  it('returns seconds for interval:N', () => {
    expect(parseInterval('interval:60')).toBe(60);
    expect(parseInterval('interval:5')).toBe(5);
  });

  it('returns null for non-interval periods', () => {
    expect(parseInterval('daily')).toBeNull();
    expect(parseInterval('weekly')).toBeNull();
  });

  it('throws on malformed interval', () => {
    expect(() => parseInterval('interval:abc' as never)).toThrow();
    expect(() => parseInterval('interval:-5' as never)).toThrow();
  });
});

describe('computeNextFire — daily', () => {
  it('returns today 06:00 Berlin when called before 06:00', () => {
    const now = new Date('2026-04-11T03:00:00Z'); // 05:00 Berlin
    const next = computeNextFire({
      period: 'daily',
      phase: '06:00',
      timezone: 'Europe/Berlin',
      now,
      lastFired: null,
    });
    // 06:00 Berlin on 2026-04-11 = 04:00 UTC
    expect(next.toISOString()).toBe('2026-04-11T04:00:00.000Z');
  });

  it('returns tomorrow 06:00 when called after 06:00', () => {
    const now = new Date('2026-04-11T07:00:00Z'); // 09:00 Berlin
    const next = computeNextFire({
      period: 'daily',
      phase: '06:00',
      timezone: 'Europe/Berlin',
      now,
      lastFired: null,
    });
    expect(next.toISOString()).toBe('2026-04-12T04:00:00.000Z');
  });

  it('advances one day after a fire', () => {
    const lastFired = new Date('2026-04-11T04:00:00Z');
    const now = new Date('2026-04-11T04:00:01Z');
    const next = computeNextFire({
      period: 'daily',
      phase: '06:00',
      timezone: 'Europe/Berlin',
      now,
      lastFired,
    });
    expect(next.toISOString()).toBe('2026-04-12T04:00:00.000Z');
  });
});

describe('computeNextFire — hourly', () => {
  it('fires at the next :00 of the hour when phase is "00"', () => {
    const now = new Date('2026-04-11T10:37:00Z');
    const next = computeNextFire({
      period: 'hourly',
      phase: '00',
      timezone: 'UTC',
      now,
      lastFired: null,
    });
    expect(next.toISOString()).toBe('2026-04-11T11:00:00.000Z');
  });
});

describe('computeNextFire — interval', () => {
  it('fires N seconds from now when no previous fire', () => {
    const now = new Date('2026-04-11T10:00:00Z');
    const next = computeNextFire({
      period: 'interval:60',
      phase: '',
      timezone: 'UTC',
      now,
      lastFired: null,
    });
    expect(next.toISOString()).toBe('2026-04-11T10:01:00.000Z');
  });

  it('fires N seconds after last fire — no drift', () => {
    const lastFired = new Date('2026-04-11T10:00:00Z');
    const now = new Date('2026-04-11T10:01:17Z'); // 17s late tick
    const next = computeNextFire({
      period: 'interval:60',
      phase: '',
      timezone: 'UTC',
      now,
      lastFired,
    });
    // Must be lastFired + 60s, not now + 60s — drift-free
    expect(next.toISOString()).toBe('2026-04-11T10:02:00.000Z');
  });
});

describe('computeNextFire — weekly', () => {
  it('fires next Monday 06:00 Berlin', () => {
    // 2026-04-11 is a Saturday
    const now = new Date('2026-04-11T12:00:00Z');
    const next = computeNextFire({
      period: 'weekly',
      phase: 'monday 06:00',
      timezone: 'Europe/Berlin',
      now,
      lastFired: null,
    });
    // Next Monday = 2026-04-13, 06:00 Berlin = 04:00 UTC
    expect(next.toISOString()).toBe('2026-04-13T04:00:00.000Z');
  });
});

pnpm --filter @cambium/rhythms test

Expected: all tests fail with module-not-found on ./phase.js.

Create packages/rhythms/src/phase.ts:

import type { RhythmPeriod } from './types.js';

export interface ComputeNextFireInput {
  period: RhythmPeriod;
  phase: string;
  timezone: string;
  now: Date;
  lastFired: Date | null;
}

/**
 * Parse an "interval:N" period into seconds, or return null for other periods.
 * Throws on malformed interval strings.
 */
export function parseInterval(period: RhythmPeriod): number | null {
  if (!period.startsWith('interval:')) return null;
  const raw = period.slice('interval:'.length);
  const n = Number(raw);
  if (!Number.isFinite(n) || n <= 0 || !Number.isInteger(n)) {
    throw new Error(`invalid interval period: ${period}`);
  }
  return n;
}

/**
 * Compute the next fire time for a rhythm. Drift-free: when a lastFired is
 * present, intervals advance from the previous fire, not from `now`.
 */
export function computeNextFire(input: ComputeNextFireInput): Date {
  const { period, phase, timezone, now, lastFired } = input;

  const intervalSeconds = parseInterval(period);
  if (intervalSeconds !== null) {
    const anchor = lastFired ?? now;
    return new Date(anchor.getTime() + intervalSeconds * 1000);
  }

  if (period === 'daily') {
    return nextDaily(phase, timezone, now, lastFired);
  }
  if (period === 'hourly') {
    return nextHourly(phase, timezone, now, lastFired);
  }
  if (period === 'weekly') {
    return nextWeekly(phase, timezone, now, lastFired);
  }
  throw new Error(`unsupported period: ${period}`);
}

/** Parse "HH:MM" into {hour, minute}. */
function parseHHMM(phase: string): { hour: number; minute: number } {
  const [h, m] = phase.split(':').map(Number);
  if (!Number.isFinite(h) || !Number.isFinite(m)) {
    throw new Error(`invalid phase "${phase}" — expected "HH:MM"`);
  }
  return { hour: h, minute: m };
}

/**
 * Build a UTC Date representing a specific wall-clock time in a target timezone
 * on a given day. Uses Intl.DateTimeFormat to discover the UTC offset.
 */
function wallClockInZoneToUtc(
  yearMonthDay: { year: number; month: number; day: number },
  hour: number,
  minute: number,
  timezone: string,
): Date {
  // Start with a guess: the wall-clock values interpreted as UTC.
  const guess = Date.UTC(
    yearMonthDay.year,
    yearMonthDay.month - 1,
    yearMonthDay.day,
    hour,
    minute,
    0,
    0,
  );
  // Find what that UTC instant looks like in the target timezone.
  const formatter = new Intl.DateTimeFormat('en-US', {
    timeZone: timezone,
    year: 'numeric',
    month: '2-digit',
    day: '2-digit',
    hour: '2-digit',
    minute: '2-digit',
    hour12: false,
  });
  const parts = Object.fromEntries(
    formatter.formatToParts(new Date(guess)).map((p) => [p.type, p.value]),
  );
  const observedUtc = Date.UTC(
    Number(parts['year']),
    Number(parts['month']) - 1,
    Number(parts['day']),
    Number(parts['hour']) === 24 ? 0 : Number(parts['hour']),
    Number(parts['minute']),
  );
  const offsetMs = observedUtc - guess;
  return new Date(guess - offsetMs);
}

/** Extract the {year, month, day} of a Date as observed in a given timezone. */
function dateInZone(date: Date, timezone: string): { year: number; month: number; day: number } {
  const formatter = new Intl.DateTimeFormat('en-US', {
    timeZone: timezone,
    year: 'numeric',
    month: '2-digit',
    day: '2-digit',
  });
  const parts = Object.fromEntries(
    formatter.formatToParts(date).map((p) => [p.type, p.value]),
  );
  return {
    year: Number(parts['year']),
    month: Number(parts['month']),
    day: Number(parts['day']),
  };
}

function nextDaily(phase: string, timezone: string, now: Date, lastFired: Date | null): Date {
  const { hour, minute } = parseHHMM(phase);
  const today = dateInZone(now, timezone);
  let candidate = wallClockInZoneToUtc(today, hour, minute, timezone);

  // If today's fire time is already past (or already fired for today), advance to tomorrow.
  const shouldAdvance =
    candidate.getTime() <= now.getTime() ||
    (lastFired !== null && lastFired.getTime() >= candidate.getTime());

  if (shouldAdvance) {
    const tomorrowUtc = new Date(candidate.getTime() + 24 * 60 * 60 * 1000);
    const tomorrow = dateInZone(tomorrowUtc, timezone);
    candidate = wallClockInZoneToUtc(tomorrow, hour, minute, timezone);
  }
  return candidate;
}

function nextHourly(phase: string, _timezone: string, now: Date, lastFired: Date | null): Date {
  const minute = Number(phase);
  if (!Number.isFinite(minute) || minute < 0 || minute > 59) {
    throw new Error(`invalid hourly phase "${phase}" — expected 0-59`);
  }
  const anchor = lastFired ?? now;
  const next = new Date(anchor.getTime());
  next.setUTCMinutes(minute, 0, 0);
  if (next.getTime() <= anchor.getTime()) {
    next.setUTCHours(next.getUTCHours() + 1);
  }
  // If the anchor was lastFired (in the past), and 'next' is still before 'now',
  // skip forward to the next hour at or after 'now'.
  while (next.getTime() <= now.getTime()) {
    next.setUTCHours(next.getUTCHours() + 1);
  }
  return next;
}

const WEEKDAY_MAP: Record<string, number> = {
  sunday: 0, monday: 1, tuesday: 2, wednesday: 3,
  thursday: 4, friday: 5, saturday: 6,
};

function nextWeekly(phase: string, timezone: string, now: Date, lastFired: Date | null): Date {
  // phase format: "monday 06:00"
  const [dayName, hhmm] = phase.split(/\s+/);
  const targetDow = WEEKDAY_MAP[dayName.toLowerCase()];
  if (targetDow === undefined) {
    throw new Error(`invalid weekly phase "${phase}" — unknown weekday`);
  }
  const { hour, minute } = parseHHMM(hhmm);

  // Walk forward day-by-day from today in the target zone until we find the target weekday.
  let cursor = new Date(now.getTime());
  for (let i = 0; i < 8; i++) {
    const dayParts = dateInZone(cursor, timezone);
    const candidate = wallClockInZoneToUtc(dayParts, hour, minute, timezone);
    const dow = new Intl.DateTimeFormat('en-US', { timeZone: timezone, weekday: 'long' })
      .format(candidate)
      .toLowerCase();
    if (
      WEEKDAY_MAP[dow] === targetDow &&
      candidate.getTime() > now.getTime() &&
      (lastFired === null || lastFired.getTime() < candidate.getTime())
    ) {
      return candidate;
    }
    cursor = new Date(cursor.getTime() + 24 * 60 * 60 * 1000);
  }
  throw new Error('nextWeekly: failed to find next fire within 8 days — bug');
}

pnpm --filter @cambium/rhythms test

Expected: all phase.test.ts tests pass. If the weekly test fails due to 2026-04-11 is a Saturday being wrong, check the date — the plan assumes it; correct the test fixture to whatever weekday that ISO date actually is.

git add packages/rhythms/src/phase.ts packages/rhythms/src/phase.test.ts
git commit -m "feat(rhythms): implement drift-free computeNextFire for daily/hourly/weekly/interval"

Task 4: Add DB Schema for Rhythms

Files: - Modify: packages/db/src/schema.ts (add rhythmInstances and rhythmFires tables) - Generated: packages/db/drizzle/XXXX_<name>.sql (Drizzle generates this)

Open packages/db/src/schema.ts. Find the existing pattern for markers or workflowRuns and add the following at an appropriate location (near other “living” tables like markers):

export const rhythmInstances = pgTable('rhythm_instances', {
  id: text('id').primaryKey(),
  workspace_id: text('workspace_id').notNull(),
  template_name: text('template_name').notNull(),
  period: text('period').notNull(),                      // "daily" | "weekly" | "hourly" | "interval:N"
  phase: text('phase').notNull(),                        // "06:00" | "monday 06:00" | "00" | ""
  timezone: text('timezone').notNull(),                  // IANA
  signal_type: text('signal_type').notNull(),
  signal_class: text('signal_class').notNull(),
  signal_intensity: real('signal_intensity').notNull(),
  signal_decay_rate: real('signal_decay_rate').notNull(),
  signal_payload: jsonb('signal_payload')
    .notNull()
    .$type<Record<string, unknown>>()
    .default({}),
  signal_affinity: jsonb('signal_affinity')
    .notNull()
    .$type<string[]>()
    .default([]),
  catchup_policy: text('catchup_policy').notNull(),      // "fire_once" | "skip" | "coalesce"
  entrainment: jsonb('entrainment')
    .$type<Record<string, unknown>>(),                   // nullable in v1
  state: text('state').notNull().default('active'),      // "active" | "paused" | "quarantined"
  last_fired_at: timestamp('last_fired_at', { withTimezone: true }),
  next_fire_at: timestamp('next_fire_at', { withTimezone: true }).notNull(),
  created_at: timestamp('created_at', { withTimezone: true })
    .notNull()
    .defaultNow(),
  updated_at: timestamp('updated_at', { withTimezone: true })
    .notNull()
    .defaultNow(),
}, (table) => [
  index('idx_rhythm_instances_workspace').on(table.workspace_id),
  index('idx_rhythm_instances_next_fire').on(table.next_fire_at),
  index('idx_rhythm_instances_state').on(table.state),
  uniqueIndex('uq_rhythm_instances_workspace_template')
    .on(table.workspace_id, table.template_name),
]);

export const rhythmFires = pgTable('rhythm_fires', {
  id: text('id').primaryKey(),
  rhythm_id: text('rhythm_id')
    .notNull()
    .references(() => rhythmInstances.id),
  intended_at: timestamp('intended_at', { withTimezone: true }).notNull(),
  actual_at: timestamp('actual_at', { withTimezone: true }).notNull().defaultNow(),
  signal_id: text('signal_id').notNull(),
  was_catchup: boolean('was_catchup').notNull().default(false),
}, (table) => [
  index('idx_rhythm_fires_rhythm').on(table.rhythm_id),
  index('idx_rhythm_fires_actual_at').on(table.actual_at),
]);

If uniqueIndex or boolean are not already imported at the top of the file, add them to the drizzle-orm/pg-core import line.

Run from repo root:

pnpm --filter @cambium/db exec drizzle-kit generate

Expected: a new file appears in packages/db/drizzle/ named XXXX_<some_name>.sql containing CREATE TABLE rhythm_instances ... and CREATE TABLE rhythm_fires .... Verify the SQL looks sane. If the repo uses a different migration generation script (check packages/db/package.json scripts), use that instead.

docker compose up -d  # ensure Postgres is running
pnpm --filter @cambium/db migrate

Expected: migration applies with no errors. Verify the tables exist:

docker compose exec postgres psql -U cambium -d cambium -c "\dt rhythm_*"

Expected: shows rhythm_instances and rhythm_fires.

pnpm --filter @cambium/db typecheck

Expected: exits 0.

git add packages/db/src/schema.ts packages/db/drizzle/
git commit -m "feat(db): add rhythm_instances and rhythm_fires tables"

Task 5: RhythmStore — DB CRUD

Files: - Create: packages/rhythms/src/rhythm-store.ts - Create: packages/rhythms/src/rhythm-store.test.ts

The store has an interface + a Postgres implementation + an in-memory implementation (for fast unit tests). Tests target the in-memory implementation. A follow-up task could add a DB-backed integration test, but for v1 we keep the in-memory version as the testable contract and trust Drizzle at the boundary.

Create packages/rhythms/src/rhythm-store.test.ts:

import { describe, it, expect, beforeEach } from 'vitest';
import { InMemoryRhythmStore } from './rhythm-store.js';
import type { RhythmInstance, RhythmId } from './types.js';

function makeInstance(overrides: Partial<RhythmInstance> = {}): RhythmInstance {
  const now = new Date('2026-04-11T00:00:00Z');
  return {
    id: 'r-1' as RhythmId,
    workspace_id: 'obadiah' as never,
    template_name: 'morning.pulse',
    period: 'daily',
    phase: '06:00',
    timezone: 'Europe/Berlin',
    signal_type: 'morning.pulse',
    signal_class: 'lifecycle',
    signal_intensity: 1.0,
    signal_decay_rate: 0,
    signal_payload: {},
    signal_affinity: [],
    catchup_policy: 'fire_once',
    entrainment: null,
    state: 'active',
    last_fired_at: null,
    next_fire_at: new Date('2026-04-11T04:00:00Z'),
    created_at: now,
    updated_at: now,
    ...overrides,
  };
}

describe('InMemoryRhythmStore', () => {
  let store: InMemoryRhythmStore;

  beforeEach(() => {
    store = new InMemoryRhythmStore();
  });

  it('upserts an instance by (workspace_id, template_name)', async () => {
    const instance = makeInstance();
    await store.upsert(instance);
    const found = await store.findByTemplateName('obadiah' as never, 'morning.pulse');
    expect(found?.id).toBe('r-1');
  });

  it('upsert is idempotent — calling twice does not duplicate', async () => {
    await store.upsert(makeInstance());
    await store.upsert(makeInstance());
    const all = await store.findByWorkspace('obadiah' as never);
    expect(all).toHaveLength(1);
  });

  it('findDue returns active instances where next_fire_at <= now', async () => {
    const due = makeInstance({
      id: 'r-due' as RhythmId,
      template_name: 'due-one',
      next_fire_at: new Date('2026-04-11T03:00:00Z'),
    });
    const notDue = makeInstance({
      id: 'r-future' as RhythmId,
      template_name: 'future-one',
      next_fire_at: new Date('2026-04-11T12:00:00Z'),
    });
    const paused = makeInstance({
      id: 'r-paused' as RhythmId,
      template_name: 'paused-one',
      state: 'paused',
      next_fire_at: new Date('2026-04-11T03:00:00Z'),
    });
    await store.upsert(due);
    await store.upsert(notDue);
    await store.upsert(paused);

    const result = await store.findDue('obadiah' as never, new Date('2026-04-11T04:00:00Z'));
    expect(result.map(r => r.id)).toEqual(['r-due']);
  });

  it('recordFire updates last_fired_at, next_fire_at, and appends a fire record', async () => {
    await store.upsert(makeInstance());
    const fire = await store.recordFire({
      rhythm_id: 'r-1' as RhythmId,
      intended_at: new Date('2026-04-11T04:00:00Z'),
      actual_at: new Date('2026-04-11T04:00:05Z'),
      next_fire_at: new Date('2026-04-12T04:00:00Z'),
      signal_id: 'sig-123',
      was_catchup: false,
    });
    expect(fire.id).toBeTruthy();
    const updated = await store.findByTemplateName('obadiah' as never, 'morning.pulse');
    expect(updated?.last_fired_at?.toISOString()).toBe('2026-04-11T04:00:05.000Z');
    expect(updated?.next_fire_at.toISOString()).toBe('2026-04-12T04:00:00.000Z');
    const fires = store.firesFor('r-1' as RhythmId);
    expect(fires).toHaveLength(1);
  });
});

pnpm --filter @cambium/rhythms test

Expected: new tests fail on module-not-found.

Create packages/rhythms/src/rhythm-store.ts:

import type { WorkspaceId } from '@cambium/core';
import type { RhythmInstance, RhythmFire, RhythmId, RhythmFireId } from './types.js';

export interface RecordFireInput {
  rhythm_id: RhythmId;
  intended_at: Date;
  actual_at: Date;
  next_fire_at: Date;
  signal_id: string;
  was_catchup: boolean;
}

export interface RhythmStore {
  upsert(instance: RhythmInstance): Promise<void>;
  findByTemplateName(workspace: WorkspaceId, templateName: string): Promise<RhythmInstance | null>;
  findByWorkspace(workspace: WorkspaceId): Promise<RhythmInstance[]>;
  findDue(workspace: WorkspaceId, now: Date): Promise<RhythmInstance[]>;
  recordFire(input: RecordFireInput): Promise<RhythmFire>;
}

/**
 * In-memory implementation for unit tests and tests that don't need a real DB.
 * Not suitable for production — rhythm state must persist across restarts.
 */
export class InMemoryRhythmStore implements RhythmStore {
  private byKey = new Map<string, RhythmInstance>();  // key: `${workspace}::${template_name}`
  private fires = new Map<RhythmId, RhythmFire[]>();
  private fireIdCounter = 0;

  private keyFor(workspace: string, templateName: string): string {
    return `${workspace}::${templateName}`;
  }

  async upsert(instance: RhythmInstance): Promise<void> {
    const key = this.keyFor(instance.workspace_id, instance.template_name);
    this.byKey.set(key, { ...instance });
  }

  async findByTemplateName(workspace: WorkspaceId, templateName: string): Promise<RhythmInstance | null> {
    return this.byKey.get(this.keyFor(workspace, templateName)) ?? null;
  }

  async findByWorkspace(workspace: WorkspaceId): Promise<RhythmInstance[]> {
    return Array.from(this.byKey.values()).filter((i) => i.workspace_id === workspace);
  }

  async findDue(workspace: WorkspaceId, now: Date): Promise<RhythmInstance[]> {
    return Array.from(this.byKey.values()).filter(
      (i) =>
        i.workspace_id === workspace &&
        i.state === 'active' &&
        i.next_fire_at.getTime() <= now.getTime(),
    );
  }

  async recordFire(input: RecordFireInput): Promise<RhythmFire> {
    this.fireIdCounter += 1;
    const fire: RhythmFire = {
      id: `f-${this.fireIdCounter}` as RhythmFireId,
      rhythm_id: input.rhythm_id,
      intended_at: input.intended_at,
      actual_at: input.actual_at,
      signal_id: input.signal_id,
      was_catchup: input.was_catchup,
    };
    const list = this.fires.get(input.rhythm_id) ?? [];
    list.push(fire);
    this.fires.set(input.rhythm_id, list);

    // Update the instance's last_fired_at and next_fire_at.
    for (const [key, instance] of this.byKey.entries()) {
      if (instance.id === input.rhythm_id) {
        this.byKey.set(key, {
          ...instance,
          last_fired_at: input.actual_at,
          next_fire_at: input.next_fire_at,
          updated_at: new Date(),
        });
        break;
      }
    }
    return fire;
  }

  /** Test helper — not part of the interface. */
  firesFor(id: RhythmId): RhythmFire[] {
    return this.fires.get(id) ?? [];
  }
}

pnpm --filter @cambium/rhythms test

Expected: all rhythm-store tests pass.

Update packages/rhythms/src/index.ts to add:

export {
  InMemoryRhythmStore,
  type RhythmStore,
  type RecordFireInput,
} from './rhythm-store.js';

git add packages/rhythms/src/rhythm-store.ts packages/rhythms/src/rhythm-store.test.ts packages/rhythms/src/index.ts
git commit -m "feat(rhythms): RhythmStore interface + InMemoryRhythmStore impl"

Task 6: YAML Template Loader

Files: - Create: packages/rhythms/src/template-loader.ts - Create: packages/rhythms/src/template-loader.test.ts - Create: packages/rhythms/test/fixtures/morning-pulse.yaml - Create: packages/rhythms/test/fixtures/invalid.yaml

Create packages/rhythms/test/fixtures/morning-pulse.yaml:

name: morning.pulse
period: daily
phase: "06:00"
timezone: "Europe/Berlin"
signal:
  type: morning.pulse
  class: lifecycle
  intensity: 1.0
  decay_rate: 0
  payload:
    intent: "morning brief"
catchup: fire_once

Create packages/rhythms/test/fixtures/invalid.yaml:

name: broken
# missing period, phase, timezone, signal, catchup

Create packages/rhythms/src/template-loader.test.ts:

import { describe, it, expect } from 'vitest';
import { resolve } from 'node:path';
import { loadTemplatesFromDirectory, parseTemplate } from './template-loader.js';

const fixturesDir = resolve(import.meta.dirname, '../test/fixtures');

describe('parseTemplate', () => {
  it('parses a valid morning-pulse template', () => {
    const yaml = `
name: morning.pulse
period: daily
phase: "06:00"
timezone: "Europe/Berlin"
signal:
  type: morning.pulse
  class: lifecycle
  intensity: 1.0
  decay_rate: 0
  payload:
    intent: "morning brief"
catchup: fire_once
`;
    const template = parseTemplate(yaml, 'morning-pulse.yaml');
    expect(template.name).toBe('morning.pulse');
    expect(template.period).toBe('daily');
    expect(template.phase).toBe('06:00');
    expect(template.signal.type).toBe('morning.pulse');
    expect(template.catchup).toBe('fire_once');
  });

  it('throws on missing required fields with filename context', () => {
    const yaml = `name: broken`;
    expect(() => parseTemplate(yaml, 'broken.yaml')).toThrow(/broken\.yaml/);
  });

  it('rejects unsupported catchup policies in v1', () => {
    const yaml = `
name: t
period: daily
phase: "06:00"
timezone: UTC
signal:
  type: t
  class: lifecycle
  intensity: 1
  decay_rate: 0
catchup: skip
`;
    expect(() => parseTemplate(yaml, 't.yaml')).toThrow(/catchup.*skip.*not implemented/i);
  });
});

describe('loadTemplatesFromDirectory', () => {
  it('loads all .yaml files and skips invalid ones with collected errors', async () => {
    const result = await loadTemplatesFromDirectory(fixturesDir);
    expect(result.templates.map(t => t.name)).toContain('morning.pulse');
    expect(result.errors.some(e => e.file.includes('invalid.yaml'))).toBe(true);
  });

  it('returns empty result for a missing directory (no throw)', async () => {
    const result = await loadTemplatesFromDirectory('/tmp/this-does-not-exist-xyz');
    expect(result.templates).toEqual([]);
    expect(result.errors).toEqual([]);
  });
});

pnpm --filter @cambium/rhythms test

Expected: new tests fail on module-not-found.

Create packages/rhythms/src/template-loader.ts:

import { readdir, readFile, stat } from 'node:fs/promises';
import { join } from 'node:path';
import { parse } from 'yaml';
import type { RhythmTemplate, CatchupPolicy, RhythmPeriod } from './types.js';

export interface LoadResult {
  templates: RhythmTemplate[];
  errors: Array<{ file: string; message: string }>;
}

export function parseTemplate(yamlText: string, filename: string): RhythmTemplate {
  const raw = parse(yamlText);
  if (!raw || typeof raw !== 'object') {
    throw new Error(`${filename}: template must be a YAML object`);
  }
  const obj = raw as Record<string, unknown>;

  const requireField = <T>(key: string, typeCheck: (v: unknown) => v is T): T => {
    const v = obj[key];
    if (!typeCheck(v)) {
      throw new Error(`${filename}: missing or invalid field "${key}"`);
    }
    return v;
  };

  const isString = (v: unknown): v is string => typeof v === 'string';
  const isNumber = (v: unknown): v is number => typeof v === 'number';
  const isObject = (v: unknown): v is Record<string, unknown> =>
    typeof v === 'object' && v !== null && !Array.isArray(v);

  const name = requireField('name', isString);
  const period = requireField('period', isString) as RhythmPeriod;
  const phase = requireField('phase', isString);
  const timezone = requireField('timezone', isString);
  const signalRaw = requireField('signal', isObject);
  const catchup = requireField('catchup', isString) as CatchupPolicy;

  if (catchup !== 'fire_once') {
    throw new Error(
      `${filename}: catchup policy "${catchup}" is not implemented in Rhythms v1 — only "fire_once" is supported`,
    );
  }

  // Validate period format
  if (
    period !== 'daily' &&
    period !== 'weekly' &&
    period !== 'hourly' &&
    !period.startsWith('interval:')
  ) {
    throw new Error(`${filename}: unsupported period "${period}"`);
  }

  const signal = {
    type: (() => {
      const t = signalRaw['type'];
      if (!isString(t)) throw new Error(`${filename}: signal.type must be a string`);
      return t;
    })(),
    class: (() => {
      const c = signalRaw['class'];
      if (!isString(c)) throw new Error(`${filename}: signal.class must be a string`);
      return c as RhythmTemplate['signal']['class'];
    })(),
    intensity: (() => {
      const i = signalRaw['intensity'];
      if (!isNumber(i)) throw new Error(`${filename}: signal.intensity must be a number`);
      return i;
    })(),
    decay_rate: (() => {
      const d = signalRaw['decay_rate'];
      if (!isNumber(d)) throw new Error(`${filename}: signal.decay_rate must be a number`);
      return d;
    })(),
    payload: isObject(signalRaw['payload'])
      ? (signalRaw['payload'] as Record<string, unknown>)
      : {},
    affinity: Array.isArray(signalRaw['affinity']) ? (signalRaw['affinity'] as string[]) : [],
  };

  return { name, period, phase, timezone, signal, catchup };
}

export async function loadTemplatesFromDirectory(dir: string): Promise<LoadResult> {
  const result: LoadResult = { templates: [], errors: [] };
  let entries: string[];
  try {
    const s = await stat(dir);
    if (!s.isDirectory()) return result;
    entries = await readdir(dir);
  } catch (err) {
    const code = (err as NodeJS.ErrnoException).code;
    if (code === 'ENOENT') return result;
    throw err;
  }

  for (const entry of entries) {
    if (!entry.endsWith('.yaml') && !entry.endsWith('.yml')) continue;
    const filepath = join(dir, entry);
    try {
      const text = await readFile(filepath, 'utf8');
      const template = parseTemplate(text, entry);
      result.templates.push(template);
    } catch (err) {
      result.errors.push({
        file: filepath,
        message: err instanceof Error ? err.message : String(err),
      });
    }
  }
  return result;
}

pnpm --filter @cambium/rhythms test

Expected: all template-loader tests pass.

Add to packages/rhythms/src/index.ts:

export { loadTemplatesFromDirectory, parseTemplate, type LoadResult } from './template-loader.js';

git add packages/rhythms/src/template-loader.ts packages/rhythms/src/template-loader.test.ts packages/rhythms/test/fixtures/ packages/rhythms/src/index.ts
git commit -m "feat(rhythms): YAML template loader with fire_once-only v1 validation"

Task 7: Seeding — Template → Instance

Files: - Create: packages/rhythms/src/seed.ts - Create: packages/rhythms/src/seed.test.ts

Create packages/rhythms/src/seed.test.ts:

import { describe, it, expect, beforeEach } from 'vitest';
import { seedTemplatesIntoStore } from './seed.js';
import { InMemoryRhythmStore } from './rhythm-store.js';
import type { RhythmTemplate } from './types.js';
import type { WorkspaceId } from '@cambium/core';

const workspace = 'obadiah' as WorkspaceId;
const now = new Date('2026-04-11T03:00:00Z');

const template: RhythmTemplate = {
  name: 'morning.pulse',
  period: 'daily',
  phase: '06:00',
  timezone: 'Europe/Berlin',
  signal: {
    type: 'morning.pulse',
    class: 'lifecycle',
    intensity: 1.0,
    decay_rate: 0,
    payload: { intent: 'morning brief' },
    affinity: [],
  },
  catchup: 'fire_once',
};

describe('seedTemplatesIntoStore', () => {
  let store: InMemoryRhythmStore;

  beforeEach(() => {
    store = new InMemoryRhythmStore();
  });

  it('creates an instance from a template on first seed', async () => {
    await seedTemplatesIntoStore({
      workspace,
      templates: [template],
      store,
      now,
      idGenerator: () => 'r-morning-pulse',
    });
    const instance = await store.findByTemplateName(workspace, 'morning.pulse');
    expect(instance).not.toBeNull();
    expect(instance?.signal_type).toBe('morning.pulse');
    expect(instance?.next_fire_at.toISOString()).toBe('2026-04-11T04:00:00.000Z');
  });

  it('is idempotent — seeding twice does not duplicate or reset phase', async () => {
    await seedTemplatesIntoStore({
      workspace,
      templates: [template],
      store,
      now,
      idGenerator: () => 'r-morning-pulse',
    });

    // Simulate a fire having happened since the first seed.
    const instance = await store.findByTemplateName(workspace, 'morning.pulse');
    await store.recordFire({
      rhythm_id: instance!.id,
      intended_at: new Date('2026-04-11T04:00:00Z'),
      actual_at: new Date('2026-04-11T04:00:00Z'),
      next_fire_at: new Date('2026-04-12T04:00:00Z'),
      signal_id: 'sig-1',
      was_catchup: false,
    });

    // Seed again — must not reset next_fire_at back to today.
    await seedTemplatesIntoStore({
      workspace,
      templates: [template],
      store,
      now,
      idGenerator: () => 'r-should-not-be-used',
    });

    const after = await store.findByTemplateName(workspace, 'morning.pulse');
    expect(after?.id).toBe('r-morning-pulse');  // same id
    expect(after?.next_fire_at.toISOString()).toBe('2026-04-12T04:00:00.000Z');
    const all = await store.findByWorkspace(workspace);
    expect(all).toHaveLength(1);
  });
});

pnpm --filter @cambium/rhythms test

Expected: new tests fail on module-not-found.

Create packages/rhythms/src/seed.ts:

import type { WorkspaceId } from '@cambium/core';
import type { RhythmTemplate, RhythmInstance, RhythmId } from './types.js';
import type { RhythmStore } from './rhythm-store.js';
import { computeNextFire } from './phase.js';

export interface SeedInput {
  workspace: WorkspaceId;
  templates: RhythmTemplate[];
  store: RhythmStore;
  now: Date;
  idGenerator: () => string;
}

/**
 * Idempotent seed: ensures a rhythm_instance exists for each template in the deployment.
 * If the instance already exists, it is left alone (phase, next_fire_at preserved).
 */
export async function seedTemplatesIntoStore(input: SeedInput): Promise<void> {
  const { workspace, templates, store, now, idGenerator } = input;
  for (const template of templates) {
    const existing = await store.findByTemplateName(workspace, template.name);
    if (existing) continue;

    const nextFire = computeNextFire({
      period: template.period,
      phase: template.phase,
      timezone: template.timezone,
      now,
      lastFired: null,
    });

    const instance: RhythmInstance = {
      id: idGenerator() as RhythmId,
      workspace_id: workspace,
      template_name: template.name,
      period: template.period,
      phase: template.phase,
      timezone: template.timezone,
      signal_type: template.signal.type,
      signal_class: template.signal.class,
      signal_intensity: template.signal.intensity,
      signal_decay_rate: template.signal.decay_rate,
      signal_payload: template.signal.payload ?? {},
      signal_affinity: template.signal.affinity ?? [],
      catchup_policy: template.catchup,
      entrainment: null,
      state: 'active',
      last_fired_at: null,
      next_fire_at: nextFire,
      created_at: now,
      updated_at: now,
    };
    await store.upsert(instance);
  }
}

pnpm --filter @cambium/rhythms test

Expected: all seed tests pass.

Add to packages/rhythms/src/index.ts:

export { seedTemplatesIntoStore, type SeedInput } from './seed.js';

git add packages/rhythms/src/seed.ts packages/rhythms/src/seed.test.ts packages/rhythms/src/index.ts
git commit -m "feat(rhythms): idempotent seedTemplatesIntoStore"

Task 8: Tick Handler — The Heart of Rhythms

The tick handler is a PhaseHandler.sense() implementation that gets registered with the existing LifecycleEngine. Every ~30s (the engine’s cycle interval), sense() runs, queries due rhythms, emits their signals, and records fires.

Files: - Create: packages/rhythms/src/tick.ts - Create: packages/rhythms/src/tick.test.ts

Create packages/rhythms/src/tick.test.ts:

import { describe, it, expect, beforeEach } from 'vitest';
import { createRhythmsHandler } from './tick.js';
import { InMemoryRhythmStore } from './rhythm-store.js';
import type { RhythmInstance, RhythmId } from './types.js';
import type { WorkspaceId } from '@cambium/core';

interface EmittedSignal {
  type: string;
  payload: Record<string, unknown>;
}

function makeFakeBus() {
  const emitted: EmittedSignal[] = [];
  let counter = 0;
  return {
    emitted,
    createAndEmit: async (params: { type: string; payload?: Record<string, unknown> }) => {
      counter += 1;
      emitted.push({ type: params.type, payload: params.payload ?? {} });
      return { id: `sig-${counter}` };
    },
  };
}

function makeInstance(overrides: Partial<RhythmInstance> = {}): RhythmInstance {
  const now = new Date('2026-04-11T00:00:00Z');
  return {
    id: 'r-1' as RhythmId,
    workspace_id: 'obadiah' as WorkspaceId,
    template_name: 'morning.pulse',
    period: 'daily',
    phase: '06:00',
    timezone: 'Europe/Berlin',
    signal_type: 'morning.pulse',
    signal_class: 'lifecycle',
    signal_intensity: 1.0,
    signal_decay_rate: 0,
    signal_payload: { intent: 'morning brief' },
    signal_affinity: [],
    catchup_policy: 'fire_once',
    entrainment: null,
    state: 'active',
    last_fired_at: null,
    next_fire_at: new Date('2026-04-11T04:00:00Z'),
    created_at: now,
    updated_at: now,
    ...overrides,
  };
}

describe('createRhythmsHandler — tick loop', () => {
  let store: InMemoryRhythmStore;

  beforeEach(() => {
    store = new InMemoryRhythmStore();
  });

  it('fires a due rhythm once, emits signal, records fire, advances next_fire_at', async () => {
    await store.upsert(makeInstance());
    const bus = makeFakeBus();

    const handler = createRhythmsHandler({
      workspace: 'obadiah' as WorkspaceId,
      store,
      signalBus: bus as never,
      nowProvider: () => new Date('2026-04-11T04:00:30Z'), // 30s after window
    });

    const result = await handler.sense!();

    expect(bus.emitted).toHaveLength(1);
    expect(bus.emitted[0].type).toBe('morning.pulse');
    expect(bus.emitted[0].payload).toMatchObject({
      intent: 'morning brief',
      rhythm_template: 'morning.pulse',
      intended_at: '2026-04-11T04:00:00.000Z',
      was_catchup: false,
    });
    expect(result).toMatchObject({ findings_count: 1 });

    const after = await store.findByTemplateName('obadiah' as WorkspaceId, 'morning.pulse');
    expect(after?.last_fired_at?.toISOString()).toBe('2026-04-11T04:00:30.000Z');
    expect(after?.next_fire_at.toISOString()).toBe('2026-04-12T04:00:00.000Z');
  });

  it('marks a catchup fire when the gap exceeds one period', async () => {
    // Last fire was 2 days ago, next was yesterday 06:00, now is today 09:00
    await store.upsert(
      makeInstance({
        last_fired_at: new Date('2026-04-09T04:00:00Z'),
        next_fire_at: new Date('2026-04-10T04:00:00Z'),
      }),
    );
    const bus = makeFakeBus();
    const handler = createRhythmsHandler({
      workspace: 'obadiah' as WorkspaceId,
      store,
      signalBus: bus as never,
      nowProvider: () => new Date('2026-04-11T07:00:00Z'),
    });
    await handler.sense!();
    expect(bus.emitted[0].payload['was_catchup']).toBe(true);
    // fire_once: only one signal, not two (yesterday + today)
    expect(bus.emitted).toHaveLength(1);
  });

  it('is a no-op when no rhythms are due', async () => {
    await store.upsert(
      makeInstance({
        next_fire_at: new Date('2026-04-11T12:00:00Z'),
      }),
    );
    const bus = makeFakeBus();
    const handler = createRhythmsHandler({
      workspace: 'obadiah' as WorkspaceId,
      store,
      signalBus: bus as never,
      nowProvider: () => new Date('2026-04-11T04:00:00Z'),
    });
    const result = await handler.sense!();
    expect(bus.emitted).toHaveLength(0);
    expect(result).toMatchObject({ findings_count: 0 });
  });
});

pnpm --filter @cambium/rhythms test

Expected: new tick tests fail on module-not-found.

Create packages/rhythms/src/tick.ts:

import type { WorkspaceId } from '@cambium/core';
import type { RhythmStore } from './rhythm-store.js';
import type { RhythmInstance } from './types.js';
import { computeNextFire } from './phase.js';

/**
 * Minimal structural type of the signal bus we need. Declared locally to avoid
 * import cycles with @cambium/signals. At runtime, @cambium/signals'
 * SignalBus.createAndEmit conforms to this shape.
 */
interface SignalBusLike {
  createAndEmit(params: {
    type: string;
    class: string;
    source_layer: string;
    source_id: string;
    workspace_id: string;
    intensity: number;
    decay_rate: number;
    cascade_potential?: boolean;
    payload?: Record<string, unknown>;
    affinity?: string[];
  }): Promise<{ id: string }>;
}

/**
 * Minimal structural type of a PhaseHandler from @cambium/lifecycle.
 * Declared locally; the real type has more optional phases.
 */
export interface PhaseHandlerLike {
  sense?(): Promise<{ findings_count: number }>;
}

export interface CreateRhythmsHandlerInput {
  workspace: WorkspaceId;
  store: RhythmStore;
  signalBus: SignalBusLike;
  nowProvider?: () => Date;
}

/**
 * Create a PhaseHandler that processes due rhythms on every lifecycle tick.
 * Hooks into the existing LifecycleEngine — no separate timer.
 */
export function createRhythmsHandler(input: CreateRhythmsHandlerInput): PhaseHandlerLike {
  const { workspace, store, signalBus } = input;
  const nowProvider = input.nowProvider ?? (() => new Date());

  return {
    async sense() {
      const now = nowProvider();
      const due = await store.findDue(workspace, now);

      for (const instance of due) {
        await fireRhythm(instance, now, store, signalBus);
      }

      return { findings_count: due.length };
    },
  };
}

async function fireRhythm(
  instance: RhythmInstance,
  now: Date,
  store: RhythmStore,
  bus: SignalBusLike,
): Promise<void> {
  // Determine whether this is a catchup: the intended fire time (next_fire_at)
  // is more than one "reasonable" window behind now. For v1 fire_once, any
  // instance where (now - next_fire_at) > 60s is considered a catchup.
  const intendedAt = instance.next_fire_at;
  const wasCatchup = now.getTime() - intendedAt.getTime() > 60_000;

  // Emit the signal through the bus. SignalBus persists before delivery.
  const signal = await bus.createAndEmit({
    type: instance.signal_type,
    class: instance.signal_class,
    source_layer: 'rhythms',
    source_id: `rhythm/${instance.template_name}`,
    workspace_id: instance.workspace_id,
    intensity: instance.signal_intensity,
    decay_rate: instance.signal_decay_rate,
    payload: {
      ...instance.signal_payload,
      rhythm_template: instance.template_name,
      intended_at: intendedAt.toISOString(),
      actual_at: now.toISOString(),
      was_catchup: wasCatchup,
    },
    affinity: instance.signal_affinity,
  });

  // Compute the next fire time — drift-free from the intended time, not from now.
  const nextFire = computeNextFire({
    period: instance.period,
    phase: instance.phase,
    timezone: instance.timezone,
    now,
    lastFired: intendedAt,
  });

  await store.recordFire({
    rhythm_id: instance.id,
    intended_at: intendedAt,
    actual_at: now,
    next_fire_at: nextFire,
    signal_id: signal.id,
    was_catchup: wasCatchup,
  });
}

pnpm --filter @cambium/rhythms test

Expected: all tick tests pass.

Add to packages/rhythms/src/index.ts:

export { createRhythmsHandler, type CreateRhythmsHandlerInput, type PhaseHandlerLike } from './tick.js';

git add packages/rhythms/src/tick.ts packages/rhythms/src/tick.test.ts packages/rhythms/src/index.ts
git commit -m "feat(rhythms): tick handler emits signals for due rhythms with drift-free phase advance"

Task 9: Integration Test — End-to-End With Real SignalBus

Verifies the rhythm handler wires up correctly against the real SignalBus from @cambium/signals (not a fake), with a real receptor and InMemoryPersister.

Files: - Create: packages/rhythms/src/integration.test.ts

Create packages/rhythms/src/integration.test.ts:

import { describe, it, expect } from 'vitest';
import { SignalBus, InMemoryPersister } from '@cambium/signals';
import { createId } from '@cambium/core';
import type { ReceptorId, WorkspaceId } from '@cambium/core';
import type { Signal } from '@cambium/core';
import { InMemoryRhythmStore } from './rhythm-store.js';
import { seedTemplatesIntoStore } from './seed.js';
import { createRhythmsHandler } from './tick.js';
import type { RhythmTemplate } from './types.js';

describe('Rhythms end-to-end with SignalBus', () => {
  it('rhythm → signal → receptor receives with correct payload', async () => {
    const workspace = 'obadiah' as WorkspaceId;
    const store = new InMemoryRhythmStore();
    const persister = new InMemoryPersister();
    const bus = new SignalBus({ persister, maxRetries: 1, retryDelayMs: 1 });

    const received: Signal[] = [];
    bus.registerReceptor({
      id: createId<ReceptorId>('r-morning-brief-test'),
      owner: 'test',
      binds_to: ['morning.pulse'],
      response: (signal) => { received.push(signal); },
    });

    const template: RhythmTemplate = {
      name: 'morning.pulse',
      period: 'interval:1',  // fire every 1s for test
      phase: '',
      timezone: 'UTC',
      signal: {
        type: 'morning.pulse',
        class: 'lifecycle',
        intensity: 1.0,
        decay_rate: 0,
        payload: { intent: 'morning brief' },
        affinity: [],
      },
      catchup: 'fire_once',
    };

    const seedTime = new Date('2026-04-11T10:00:00Z');
    await seedTemplatesIntoStore({
      workspace,
      templates: [template],
      store,
      now: seedTime,
      idGenerator: () => 'r-test-1',
    });

    const handler = createRhythmsHandler({
      workspace,
      store,
      signalBus: bus,
      // First tick is 2s after seed — rhythm (interval:1) is due
      nowProvider: () => new Date('2026-04-11T10:00:02Z'),
    });

    const result = await handler.sense!();

    expect(result.findings_count).toBe(1);
    expect(received).toHaveLength(1);
    expect(received[0].type).toBe('morning.pulse');
    expect(received[0].payload['intent']).toBe('morning brief');
    expect(received[0].payload['rhythm_template']).toBe('morning.pulse');
    expect(persister.signals).toHaveLength(1);  // persist-before-deliver invariant
    expect(persister.signals[0].type).toBe('morning.pulse');
  });
});

pnpm --filter @cambium/rhythms test

Expected: integration test passes. If the imports from @cambium/signals or @cambium/core differ from what’s shown (e.g. InMemoryPersister is exported from a different path), check the real exports with grep -r "export.*InMemoryPersister" packages/signals/src and adjust the import.

git add packages/rhythms/src/integration.test.ts
git commit -m "test(rhythms): integration test — rhythm → real SignalBus → receptor receives"

Task 10: Wire Rhythms into Obadiah Deployment

This task creates the real morning.pulse template and a boot entry point that seeds it and starts the handler inside Obadiah’s process. For v1 we wire it as a standalone script that runs a minimal loop — full integration into cambium-bridge.ts is deferred.

Files: - Create: deployments/obadiah/rhythms/morning-pulse.yaml - Create: deployments/obadiah/bridge/start-rhythms.ts

Create deployments/obadiah/rhythms/morning-pulse.yaml:

# Obadiah morning pulse — emits at 06:00 Berlin daily.
# A receptor elsewhere (the morning-brief workflow, once migrated) consumes this
# signal and delivers the actual brief. For v1 a simple Discord receptor can
# post a "good morning" message as the end-to-end proof.
name: morning.pulse
period: daily
phase: "06:00"
timezone: "Europe/Berlin"
signal:
  type: morning.pulse
  class: lifecycle
  intensity: 1.0
  decay_rate: 0
  payload:
    intent: "morning brief"
    deployment: "obadiah"
catchup: fire_once

Create deployments/obadiah/bridge/start-rhythms.ts:

#!/usr/bin/env npx tsx
/**
 * Start the Rhythms tick loop for Obadiah.
 *
 * Run with: npx tsx deployments/obadiah/bridge/start-rhythms.ts
 *
 * This is the minimal v1 wiring: loads templates from deployments/obadiah/rhythms/,
 * seeds them into the rhythm_instances table, then runs a 30s loop that processes
 * due rhythms and emits signals. A receptor bound to "morning.pulse" — defined
 * separately — handles the actual delivery.
 *
 * When full LifecycleEngine integration is ready, this becomes a handler
 * registered on the Obadiah lifecycle engine instead of its own loop.
 */

import { resolve } from 'node:path';
import { SignalBus, InMemoryPersister } from '@cambium/signals';
import { createId } from '@cambium/core';
import type { WorkspaceId, ReceptorId } from '@cambium/core';
import {
  InMemoryRhythmStore,
  loadTemplatesFromDirectory,
  seedTemplatesIntoStore,
  createRhythmsHandler,
} from '@cambium/rhythms';

const WORKSPACE = 'obadiah' as WorkspaceId;
const RHYTHMS_DIR = resolve(import.meta.dirname, '../rhythms');
const TICK_INTERVAL_MS = 30_000;

async function main(): Promise<void> {
  console.log('[rhythms] loading templates from', RHYTHMS_DIR);
  const loaded = await loadTemplatesFromDirectory(RHYTHMS_DIR);
  if (loaded.errors.length > 0) {
    console.error('[rhythms] template errors:');
    for (const e of loaded.errors) console.error(`  - ${e.file}: ${e.message}`);
  }
  console.log(`[rhythms] loaded ${loaded.templates.length} templates`);

  const store = new InMemoryRhythmStore();  // v1: in-memory. Persist to Postgres in next pass.
  const persister = new InMemoryPersister();  // v1: in-memory. Swap for PgSignalPersister later.
  const bus = new SignalBus({ persister, maxRetries: 3, retryDelayMs: 100 });

  // Minimal proof receptor: logs when a morning.pulse arrives.
  bus.registerReceptor({
    id: createId<ReceptorId>('obadiah-morning-pulse-logger'),
    owner: 'obadiah',
    binds_to: ['morning.pulse'],
    response: (signal) => {
      console.log('[rhythms] morning.pulse received:', {
        intended_at: signal.payload['intended_at'],
        actual_at: signal.payload['actual_at'],
        was_catchup: signal.payload['was_catchup'],
      });
    },
  });

  await seedTemplatesIntoStore({
    workspace: WORKSPACE,
    templates: loaded.templates,
    store,
    now: new Date(),
    idGenerator: () => crypto.randomUUID(),
  });

  const handler = createRhythmsHandler({ workspace: WORKSPACE, store, signalBus: bus });

  console.log('[rhythms] tick loop started (interval:', TICK_INTERVAL_MS, 'ms)');
  const tick = async () => {
    try {
      const result = await handler.sense!();
      if (result.findings_count > 0) {
        console.log(`[rhythms] tick fired ${result.findings_count} rhythm(s)`);
      }
    } catch (err) {
      console.error('[rhythms] tick error:', err);
    }
  };

  await tick();
  const timer = setInterval(tick, TICK_INTERVAL_MS);

  const shutdown = () => {
    console.log('[rhythms] shutting down');
    clearInterval(timer);
    process.exit(0);
  };
  process.on('SIGINT', shutdown);
  process.on('SIGTERM', shutdown);
}

main().catch((err) => {
  console.error('[rhythms] fatal:', err);
  process.exit(1);
});

Note: this script uses InMemoryRhythmStore and InMemoryPersister to keep v1 boundary small. Real Postgres persistence is a Task-10.5 follow-up (see bottom). The tradeoff: on restart, a running Obadiah loses its last-fired state and will fire a missed pulse again. For a daily rhythm this is acceptable for v1; the pulse just arrives twice on a restart day.

Temporarily edit deployments/obadiah/rhythms/morning-pulse.yaml to use period: interval:10 and phase: "" to verify ticks fire. Then run:

npx tsx deployments/obadiah/bridge/start-rhythms.ts

Expected: within ~10 seconds you see:

[rhythms] loading templates from .../deployments/obadiah/rhythms
[rhythms] loaded 1 templates
[rhythms] tick loop started (interval: 30000 ms)
[rhythms] tick fired 1 rhythm(s)
[rhythms] morning.pulse received: { intended_at: ..., actual_at: ..., was_catchup: false }

Ctrl-C to stop.

Restore deployments/obadiah/rhythms/morning-pulse.yaml to the production values (period: daily, phase: "06:00", timezone: "Europe/Berlin").

git add deployments/obadiah/rhythms/morning-pulse.yaml deployments/obadiah/bridge/start-rhythms.ts
git commit -m "feat(obadiah): wire Rhythms with morning.pulse daily at 06:00 Berlin"

Task 11: Run Full Test Suite and Typecheck

pnpm turbo test

Expected: everything green. No regressions in existing packages.

pnpm turbo typecheck

Expected: everything green.

pnpm turbo build

Expected: @cambium/rhythms builds. No downstream breakage.

Investigate root cause. Fix inline. Do not --force or skip failing tests. Re-run until green, then proceed.

Follow-up Work (Out of Scope for This Plan)

These are tracked separately and should become follow-up plans, not added to this one:

Success Criteria (from spec)

  1. ✅ pnpm --filter @cambium/rhythms test passes — Tasks 3, 5, 6, 7, 8, 9
  2. ✅ Migration applies cleanly to dev DB — Task 4
  3. ✅ Seeding is idempotent — Task 7
  4. ✅ Test rhythm with period: interval:N fires without cumulative drift — Task 3 (phase math), Task 8 (tick handler uses intendedAt not now for next-fire computation)
  5. ✅ Catchup fires once on wake — Task 8 test marks a catchup fire when the gap exceeds one period
  6. 🟡 Morning brief lands in Discord — narrowed: the proof is that morning.pulse signal fires and a receptor receives it. Real morning brief delivery is deferred to the follow-up work above.