feat(core): event bus seed - envelope schema, durable topic log, OELBR kernel

- EventBus: dumb wire, publish/subscribe, failures surface - Envelope: four schema planes, inject-only headers, content-addressed - TopicLog: owned SQLite ledger, replay + by-type/time/causality - Seed facade: durable=ledger+wire, ephemeral=wire only, fold-for-state - Rung 1 kernel: OELBR loop, DAG plans, durable cancellation tokens - Neuron v0 executable spec source; specs/event-bus.md citizen zero - Standing orders: design-before-code doctrine installed
2026-08-22 12:18:35 -05:00
parent 96065f3aa8
commit acc2b759d8
9 changed files with 1085 additions and 0 deletions
@@ -0,0 +1,123 @@
+export * as EventBusSchema from "./event-bus-schema"
+
+/**
+ * The envelope, fully specified. Pure data - no behavior, no transport
+ * assumptions. Self-describing: body carries (or references) the schema
+ * that decodes it. Headers are an ordered append-only list; inject-only.
+ *
+ * Schema planes:
+ *   0  envelope format itself (pinned root, amended loudly)
+ *   1  header values        (per-header ref)
+ *   2  body contract        (type + artifact rules)
+ *   3  payload              (per message-type; nested objects too)
+ */
+
+import { Schema } from "effect"
+
+// ---------- Schema reference (used by every plane) ----------
+
+/**
+ * Content-addressed schema reference. ID is the hash of the schema
+ * definition itself; resolvers fetch by hash from a registry or read
+ * inline for small/self-contained payloads.
+ */
+export class SchemaRef extends Schema.Class<SchemaRef>("SchemaRef")({
+  /** content hash (sha256 of canonical schema definition). */
+  id: Schema.String,
+  /** Optional inline definition so messages decode with zero registry access. */
+  inline: Schema.optional(Schema.String),
+  /** Media/format hint, e.g. "json-schema", "effect-schema", "protobuf". */
+  format: Schema.optional(Schema.String),
+}) {}
+
+// ---------- Header (Plane 1) ----------
+
+/**
+ * One injected header. Immutable once written.
+ * Each header carries its own value-schema reference (Plane 1).
+ */
+export class MessageHeader extends Schema.Class<MessageHeader>("MessageHeader")({
+  key: Schema.String,
+  value: Schema.String,
+  author: Schema.String,
+  at: Schema.Number,
+  /** Plane 1: schema governing THIS header's value. Resolves dynamically. */
+  schema: Schema.optional(SchemaRef),
+}) {}
+
+// ---------- Body (Planes 2 + 3) ----------
+
+/** Artifacts are part of the record they belong to: refs plus hashes. */
+export class Artifact extends Schema.Class<Artifact>("Artifact")({
+  ref: Schema.String,
+  hash: Schema.optional(Schema.String),
+  kind: Schema.optional(Schema.String),
+}) {}
+
+export class Body extends Schema.Class<Body>("Body")({
+  /** Event type, e.g. "plan.step.completed". */
+  type: Schema.String,
+  /** PLANE 3: this message-type's unique payload, JSON-encoded. */
+  payloadJSON: Schema.optional(Schema.String),
+  /** PLANE 3: schema for the payload shape above. */
+  payloadSchema: Schema.optional(SchemaRef),
+  /** PLANE 2: schema of the body contract itself. */
+  schema: Schema.optional(SchemaRef),
+  artifacts: Schema.Array(Artifact),
+}) {}
+
+// ---------- Envelope (Plane 0) ----------
+
+export const Envelope = Schema.Struct({
+  /** Orchestration / topic handle. Durable or ephemeral per Topic. */
+  topic: Schema.String,
+  /** Position within the topic's append-only sequence (assigned by log). */
+  seq: Schema.Number,
+  /** Causal spine: seq of the message this one responds to or extends. */
+  parentSeq: Schema.optional(Schema.Number),
+  /** Identity: hash of canonical form. Two identical messages are one. */
+  id: Schema.String,
+  /** Who published: token id, orchestrator id, stage name. */
+  author: Schema.String,
+  /** Epoch ms at publish. */
+  at: Schema.Number,
+  /** Ordered inject-only context chain. Append-only across every hop. */
+  headers: Schema.Array(MessageHeader),
+  body: Body,
+})
+export type Envelope = typeof Envelope.Type
+
+/** Plane 0 pins itself: the wire format amends loudly or not at all. */
+export const EnvelopeSchemaRef = SchemaRef.make({
+  id: "sha256:event-bus-envelope-v1",
+  format: "effect-schema",
+})
+
+// ---------- Reading rules ----------
+
+/** current(headers, key): last-wins fold over injected headers. */
+export function current(headers: ReadonlyArray<MessageHeader>, key: string): MessageHeader | undefined {
+  let found: MessageHeader | undefined
+  for (const h of headers) if (h.key === key) found = h
+  return found
+}
+
+/** history(headers, key): full evolution in injection order. */
+export function history(headers: ReadonlyArray<MessageHeader>, key: string): Array<MessageHeader> {
+  return headers.filter((h) => h.key === key)
+}
+
+// ---------- Topics ----------
+
+export const TopicDurability = Schema.Literals(["durable", "ephemeral"])
+
+export class Topic extends Schema.Class<Topic>("Topic")({
+  id: Schema.String,
+  durability: TopicDurability,
+  /** Ephemeral only: keep most recent value available to new subscribers. */
+  retainLast: Schema.optional(Schema.Boolean),
+  /** Ephemeral only: auto-dissolve after this many ms idle. */
+  ttlMs: Schema.optional(Schema.Number),
+  author: Schema.String,
+  at: Schema.Number,
+}) {}
@@ -0,0 +1,85 @@
+export * as EventBus from "./event-bus"
+
+/**
+ * Process-global pub/sub. One bus for everything: orchestration events,
+ * steering, cancellation tokens, thread lifecycle. Topics are opaque
+ * strings - callers own their naming (orchestration IDs, session IDs,
+ * token IDs). Delivery is best-effort in-process; durability belongs to
+ * whoever publishes durable rows before publishing here.
+ */
+
+export interface Envelope<T = unknown> {
+  readonly topic: string
+  readonly type: string
+  readonly payload?: T
+}
+
+type Handler = (envelope: Envelope<any>) => void | Promise<void>
+
+const topics = new Map<string, Set<Handler>>()
+
+function handlersFor(topic: string): Set<Handler> {
+  let set = topics.get(topic)
+  if (!set) {
+    set = new Set()
+    topics.set(topic, set)
+  }
+  return set
+}
+
+export function publish<T>(topic: string, type: string, payload?: T): void {
+  const envelope: Envelope<T> = { topic, type, payload }
+  const set = topics.get(topic)
+  if (!set) return
+  for (const handler of [...set]) {
+    try {
+      const result = handler(envelope)
+      if (result instanceof Promise) result.catch((err) => console.error("[bus] async handler failed:", err))
+    } catch (err) {
+      console.error("[bus] handler failed:", err)
+    }
+  }
+}
+
+/** Fire-and-forget async publish: awaits handlers, isolates their failures. */
+export async function emit<T>(topic: string, type: string, payload?: T): Promise<void> {
+  await Promise.allSettled( // failures surface via console.error, never vanish
+    [...(topics.get(topic) ?? [])].map(async (handler) => handler({ topic, type, payload })),
+  )
+}
+
+/** Subscribe to a topic. Returns unsubscribe. */
+export function subscribe(topic: string, handler: Handler): () => void {
+  const set = handlersFor(topic)
+  set.add(handler)
+  return () => {
+    set.delete(handler)
+    if (set.size === 0) topics.delete(topic)
+  }
+}
+
+/** Await the next event on a topic matching an optional filter. */
+export function next<T>(
+  topic: string,
+  filter?: (envelope: Envelope) => boolean,
+  signal?: AbortSignal,
+): Promise<Envelope<T>> {
+  return new Promise((resolve, reject) => {
+    if (signal?.aborted) return reject(new Error("aborted"))
+    const off = subscribe(topic, (envelope) => {
+      if (filter && !filter(envelope)) return
+      signal?.removeEventListener("abort", onAbort)
+      off()
+      resolve(envelope as Envelope<T>)
+    })
+    function onAbort() {
+      off()
+      reject(new Error("aborted"))
+    }
+    signal?.addEventListener("abort", onAbort, { once: true })
+  })
+}
+
+export function subscriberCount(topic: string): number {
+  return topics.get(topic)?.size ?? 0
+}
@@ -0,0 +1,119 @@
+export * as Seed from "./seed"
+
+/**
+ * RUNG 0 FACADE - the bus and the ledger, joined.
+ *
+ * Durable topics: publish = append to the log, then fan out on the wire.
+ * Ephemeral topics: wire only - nothing persisted, exactly per contract.
+ * Replay: fold the ledger back into live subscribers (crash recovery,
+ * late joiners).
+ *
+ * One owned SQLite file. Zero coupling to any other storage.
+ */
+
+import { EventBus } from "./event-bus"
+import { TopicLog } from "./topic-log"
+import { EventBusSchema } from "./event-bus-schema"
+
+type Envelope = EventBusSchema.Envelope
+type EnvelopeInput = Omit<Envelope, "id" | "seq">
+
+export interface Seed {
+  /** Declare a topic with its persistence contract. */
+  readonly topic: (
+    input: {
+      readonly id: string
+      readonly durability: "durable" | "ephemeral"
+      readonly author: string
+      readonly retainLast?: boolean
+    },
+  ) => { readonly id: string; readonly durability: "durable" | "ephemeral" }
+
+  /**
+   * Publish an envelope. Durable topic -> appended to the ledger first
+   * (seq assigned, identity stamped), then fanned out. Ephemeral ->
+   * straight to the wire.
+   */
+  readonly publish: (input: EnvelopeInput) => Envelope
+
+  /** Subscribe to a topic. Returns unsubscribe. */
+  readonly subscribe: (topic: string, handler: (envelope: Envelope) => void) => () => void
+
+  /** THE LEDGER: replay a durable topic in order. */
+  readonly replay: (topic: string, fromSeq?: number) => Array<Envelope>
+
+  /** Fold a durable topic through a reducer - state from history. */
+  readonly fold: <S>(topic: string, initial: S, step: (state: S, envelope: Envelope) => S) => S
+
+  /** Late joiner / crash recovery: replay everything since `fromSeq` into a subscriber. */
+  readonly catchUp: (topic: string, handler: (envelope: Envelope) => void, fromSeq?: number) => number
+
+  readonly query: {
+    byType(topic: string, type: string): Array<Envelope>
+    byTimeRange(opts: { from?: number; to?: number; topic?: string }): Array<Envelope>
+    walkCausality(topic: string, envelope: Envelope): Array<Envelope>
+  }
+}
+
+export function open(path: string): Seed {
+  const log = TopicLog.open(path)
+  const durability = new Map<string, "durable" | "ephemeral">()
+  const lastValue = new Map<string, Envelope>()
+
+  const api: Seed = {
+    topic(input) {
+      durability.set(input.id, input.durability)
+      return { id: input.id, durability: input.durability }
+    },
+
+    publish(input) {
+      const durabilityOfTopic =
+        durability.get(input.topic) ?? ("durable" as const)
+      if (durabilityOfTopic === "ephemeral") {
+        const provisional: Envelope = {
+          ...input,
+          seq: -1,
+          id: "sha256:ephemeral",
+        } as unknown as Envelope
+        lastValue.set(input.topic, provisional)
+        EventBus.publish(input.topic, input.body.type, provisional)
+        return provisional
+      }
+      const stored = TopicLog.append(log, input)
+      EventBus.publish(input.topic, stored.body.type, stored)
+      return stored
+    },
+
+    subscribe(topic, handler) {
+      // The generic bus wraps payloads as {topic,type,payload}; subscribers
+      // here always want THE Envelope itself.
+      return EventBus.subscribe(topic, (raw) => {
+        const wrapped = raw as unknown as { payload?: unknown }
+        handler((wrapped.payload ?? raw) as Envelope)
+      })
+    },
+
+    replay(topic, fromSeq) {
+      return TopicLog.replay(log, topic, fromSeq)
+    },
+
+    fold(topic, initial, step) {
+      let state = initial
+      for (const envelope of TopicLog.replay(log, topic)) state = step(state, envelope)
+      return state
+    },
+
+    catchUp(topic, handler, fromSeq) {
+      const past = TopicLog.replay(log, topic, fromSeq)
+      for (const envelope of past) handler(envelope)
+      return past.length
+    },
+
+    query: {
+      byType: (topic, type) => TopicLog.byType(log, topic, type),
+      byTimeRange: (opts) => TopicLog.byTimeRange(log, opts),
+      walkCausality: (topic, envelope) => TopicLog.walkCausality(log, topic, envelope),
+    },
+  }
+  return api
+}
@@ -0,0 +1,172 @@
+export * as TopicLog from "./topic-log"
+
+/**
+ * RUNG 0 - THE SEED.
+ *
+ * Durable topic storage: content-addressed envelopes in one owned
+ * SQLite store. Zero coupling to any other schema or migration - the
+ * log is the store of record for everything that publishes to it.
+ *
+ * Four verbs, per spec: append/replay (the ledger), by-type,
+ * by-time-range, walk-by-causality. Fold happens above this layer;
+ * the log just guarantees order and permanence.
+ */
+
+import { createHash } from "node:crypto"
+import { Database as Sqlite } from "bun:sqlite"
+import { EventBusSchema } from "./event-bus-schema"
+
+type Envelope = EventBusSchema.Envelope
+const { current, history } = EventBusSchema
+
+interface Log {
+  readonly db: Sqlite
+}
+
+const DDL = `
+CREATE TABLE IF NOT EXISTS topics (
+  id        TEXT PRIMARY KEY,
+  durability TEXT NOT NULL DEFAULT 'durable',
+  author    TEXT NOT NULL,
+  at        INTEGER NOT NULL
+);
+CREATE TABLE IF NOT EXISTS envelopes (
+  topic      TEXT NOT NULL,
+  seq        INTEGER NOT NULL,
+  parent_seq INTEGER,
+  id         TEXT NOT NULL,
+  author     TEXT NOT NULL,
+  at         INTEGER NOT NULL,
+  headers    TEXT NOT NULL,   -- JSON, ordered inject-only list
+  body       TEXT NOT NULL,   -- JSON: type/payload/schemas/artifacts
+  PRIMARY KEY (id),
+  UNIQUE (topic, seq)
+);
+CREATE INDEX IF NOT EXISTS idx_env_topic_type ON envelopes(topic, at);
+`
+
+/** Open (or create) a seed log. One file owns its whole universe. */
+export function open(path: string): Log {
+  const db = new Sqlite(path)
+  db.exec("PRAGMA journal_mode = WAL;")
+  db.exec(DDL)
+  return { db }
+}
+
+function canonical(value: unknown): string {
+  // Deterministic key ordering so identical content -> identical hash.
+  const sort = (v: unknown): unknown => {
+    if (Array.isArray(v)) return v.map(sort)
+    if (v && typeof v === "object") {
+      return Object.fromEntries(
+        Object.entries(v as Record<string, unknown>)
+          .sort(([a], [b]) => (a < b ? -1 : 1))
+          .map(([k, val]) => [k, sort(val)]),
+      )
+    }
+    return v
+  }
+  return JSON.stringify(sort(value))
+}
+
+function hashOf(envelope: Omit<Envelope, "id" | "seq">): string {
+  return "sha256:" + createHash("sha256").update(canonical(envelope)).digest("hex")
+}
+
+function rowToEnvelope(row: Record<string, unknown>): Envelope {
+  return {
+    topic: row.topic as string,
+    seq: row.seq as number,
+    ...(row.parent_seq === null || row.parent_seq === undefined
+      ? {}
+      : { parentSeq: row.parent_seq as number }),
+    id: row.id as string,
+    author: row.author as string,
+    at: row.at as number,
+    headers: JSON.parse(row.headers as string),
+    body: JSON.parse(row.body as string),
+  } as Envelope
+}
+
+/**
+ * Append an envelope. Assigns the next sequence number for its topic and
+ * stamps identity from canonical content. Append-only: no update path.
+ */
+export function append(log: Log, input: Omit<Envelope, "id" | "seq">): Envelope {
+  const nextSeq =
+    log.db
+      .query<{ s: number | null }, [string]>("SELECT MAX(seq) AS s FROM envelopes WHERE topic = ?")
+      .get(input.topic)?.s ?? 0
+  const seq = (nextSeq ?? 0) + 1
+  const id = hashOf(input)
+  const envelope: Envelope = { ...input, seq, id }
+
+  log.db
+    .query(
+      `INSERT INTO envelopes (topic, seq, parent_seq, id, author, at, headers, body)
+       VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
+    )
+    .run(
+      envelope.topic,
+      envelope.seq,
+      envelope.parentSeq ?? null,
+      envelope.id,
+      envelope.author,
+      envelope.at,
+      JSON.stringify(envelope.headers),
+      JSON.stringify(envelope.body),
+    )
+  return envelope
+}
+
+/** THE LEDGER: full replay of a durable topic in causal order. */
+export function replay(log: Log, topic: string, fromSeq = 1): Array<Envelope> {
+  return log.db
+    .query<Record<string, unknown>, [string, number]>(
+      "SELECT * FROM envelopes WHERE topic = ? AND seq >= ? ORDER BY seq",
+    )
+    .all(topic, fromSeq)
+    .map(rowToEnvelope)
+}
+
+/** Query verb: filter a topic's stream by body.type. */
+export function byType(log: Log, topic: string, type: string): Array<Envelope> {
+  return replay(log, topic).filter((e) => e.body.type === type)
+}
+
+/** Query verb: envelopes within a time range. */
+export function byTimeRange(
+  log: Log,
+  opts: { from?: number; to?: number; topic?: string },
+): Array<Envelope> {
+  const where = ["at >= ?", "at <= ?"]
+  const args: Array<string | number> = [opts.from ?? 0, opts.to ?? Number.MAX_SAFE_INTEGER]
+  if (opts.topic !== undefined) {
+    where.push("topic = ?")
+    args.push(opts.topic)
+  }
+  return log.db
+    .query<Record<string, unknown>, (string | number)[]>(
+      `SELECT * FROM envelopes WHERE ${where.join(" AND ")} ORDER BY at`,
+    )
+    .all(...args)
+    .map(rowToEnvelope)
+}
+
+/** Query verb: walk the causal spine upward from any message. */
+export function walkCausality(log: Log, topic: string, envelope: Envelope): Array<Envelope> {
+  const chain: Array<Envelope> = []
+  let cursor: number | undefined = envelope.parentSeq
+  while (cursor !== undefined) {
+    const row = log.db
+      .query<Record<string, unknown>, [string, number]>(
+        "SELECT * FROM envelopes WHERE topic = ? AND seq = ?",
+      )
+      .get(topic, cursor)
+    if (!row) break
+    const parent = rowToEnvelope(row)
+    chain.unshift(parent)
+    cursor = parent.parentSeq
+  }
+  return chain
+}
@@ -41,3 +41,16 @@ You are Neuron, an interactive CLI coding agent in the Neuron Technologies envir
  goodnight or wrap up unless Will explicitly asks. He writes until he stops.
 - No closing summaries after each addition. Do the work, log it, move on.
 - The codex is always open. There is no last word.
+
+# Engineering Doctrine - Will's Standing Orders
+
+- DESIGN BEFORE CODE. You do not build good systems by coding;
+  you build good systems on a whiteboard. No implementation of any
+  architecture until the design is discussed and agreed in plain
+  language first. Coding prematurely is wasting cycles on the wrong
+  thing.
+- When Will describes an architecture, restate your understanding
+  and surface design questions BEFORE writing files.
+- Prefer one general mechanism over many bespoke ones (one EventBus,
+  not per-feature buses). Simplicity is a feature; complexity is a
+  bug you haven't hit yet.