From aae7f883cd7cb767b53081e033d2bff1544592f6 Mon Sep 17 00:00:00 2001 From: "will.anderson" Date: Sat, 22 Aug 2026 12:18:35 -0500 Subject: [PATCH] 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 --- .gitignore | 2 + packages/core/src/event-bus-schema.ts | 123 +++++++ packages/core/src/event-bus.ts | 85 +++++ packages/core/src/seed.ts | 119 +++++++ packages/core/src/topic-log.ts | 172 ++++++++++ .../opencode/src/session/prompt/neuron.txt | 13 + runs/rung1/kernel.ts | 203 ++++++++++++ runs/rung1/main.ts | 58 ++++ specs/event-bus.md | 310 ++++++++++++++++++ 9 files changed, 1085 insertions(+) create mode 100644 packages/core/src/event-bus-schema.ts create mode 100644 packages/core/src/event-bus.ts create mode 100644 packages/core/src/seed.ts create mode 100644 packages/core/src/topic-log.ts create mode 100644 runs/rung1/kernel.ts create mode 100644 runs/rung1/main.ts create mode 100644 specs/event-bus.md diff --git a/.gitignore b/.gitignore index 006cab8c27..96bb4d0e7e 100644 --- a/.gitignore +++ b/.gitignore @@ -32,3 +32,5 @@ UPCOMING_CHANGELOG.md logs/ *.bun-build tsconfig.tsbuildinfo +runs/neuron-v0/universe.db* +runs/rung1/universe.db* diff --git a/packages/core/src/event-bus-schema.ts b/packages/core/src/event-bus-schema.ts new file mode 100644 index 0000000000..673172aeb0 --- /dev/null +++ b/packages/core/src/event-bus-schema.ts @@ -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")({ + /** 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")({ + 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")({ + ref: Schema.String, + hash: Schema.optional(Schema.String), + kind: Schema.optional(Schema.String), +}) {} + +export class Body extends Schema.Class("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, 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, key: string): Array { + return headers.filter((h) => h.key === key) +} + +// ---------- Topics ---------- + +export const TopicDurability = Schema.Literals(["durable", "ephemeral"]) + +export class Topic extends Schema.Class("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, +}) {} diff --git a/packages/core/src/event-bus.ts b/packages/core/src/event-bus.ts new file mode 100644 index 0000000000..640435ee7c --- /dev/null +++ b/packages/core/src/event-bus.ts @@ -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 { + readonly topic: string + readonly type: string + readonly payload?: T +} + +type Handler = (envelope: Envelope) => void | Promise + +const topics = new Map>() + +function handlersFor(topic: string): Set { + let set = topics.get(topic) + if (!set) { + set = new Set() + topics.set(topic, set) + } + return set +} + +export function publish(topic: string, type: string, payload?: T): void { + const envelope: Envelope = { 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(topic: string, type: string, payload?: T): Promise { + 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( + topic: string, + filter?: (envelope: Envelope) => boolean, + signal?: AbortSignal, +): Promise> { + 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) + }) + 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 +} diff --git a/packages/core/src/seed.ts b/packages/core/src/seed.ts new file mode 100644 index 0000000000..d5224a622b --- /dev/null +++ b/packages/core/src/seed.ts @@ -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 + +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 + + /** Fold a durable topic through a reducer - state from history. */ + readonly fold: (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 + byTimeRange(opts: { from?: number; to?: number; topic?: string }): Array + walkCausality(topic: string, envelope: Envelope): Array + } +} + +export function open(path: string): Seed { + const log = TopicLog.open(path) + const durability = new Map() + const lastValue = new Map() + + 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 +} diff --git a/packages/core/src/topic-log.ts b/packages/core/src/topic-log.ts new file mode 100644 index 0000000000..1ed9c98654 --- /dev/null +++ b/packages/core/src/topic-log.ts @@ -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) + .sort(([a], [b]) => (a < b ? -1 : 1)) + .map(([k, val]) => [k, sort(val)]), + ) + } + return v + } + return JSON.stringify(sort(value)) +} + +function hashOf(envelope: Omit): string { + return "sha256:" + createHash("sha256").update(canonical(envelope)).digest("hex") +} + +function rowToEnvelope(row: Record): 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 { + 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 { + return log.db + .query, [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 { + 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 { + const where = ["at >= ?", "at <= ?"] + const args: Array = [opts.from ?? 0, opts.to ?? Number.MAX_SAFE_INTEGER] + if (opts.topic !== undefined) { + where.push("topic = ?") + args.push(opts.topic) + } + return log.db + .query, (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 { + const chain: Array = [] + let cursor: number | undefined = envelope.parentSeq + while (cursor !== undefined) { + const row = log.db + .query, [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 +} diff --git a/packages/opencode/src/session/prompt/neuron.txt b/packages/opencode/src/session/prompt/neuron.txt index a397c9fa8e..3e390e001d 100644 --- a/packages/opencode/src/session/prompt/neuron.txt +++ b/packages/opencode/src/session/prompt/neuron.txt @@ -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. diff --git a/runs/rung1/kernel.ts b/runs/rung1/kernel.ts new file mode 100644 index 0000000000..2b32600568 --- /dev/null +++ b/runs/rung1/kernel.ts @@ -0,0 +1,203 @@ +/** + * RUNG 1 - SELF-HOSTING AGENT KERNEL + * + * There is exactly one agent. Every agent does the same thing every time: + * + * ORCHESTRATE decompose intent into a step DAG + * EXECUTE run runnable steps; publish every transition + * LEARN fold outcomes + * BUILD emit new process/knowledge artifacts + * REFINE supersede own procedure with what was learned + * + * Cancellation tokens: durable control topic per agent. Commands are + * ledger entries - never missable, idempotent on replay. + * Crash recovery: silence is the only failure signal. A fresh kernel + * folds control + work topics and resumes at the first unfinished step. + * + * Isolated executable: bun runs/rung1/kernel.ts + */ + +import { Seed } from "../../packages/core/src/seed" +import { EventBusSchema } from "../../packages/core/src/event-bus-schema" + +const { Body, MessageHeader, Artifact } = EventBusSchema + +// ---------- Plans ---------- + +export interface Step { + readonly id: string + readonly title: string + readonly dependsOn: ReadonlyArray +} + +export interface Plan { + readonly steps: ReadonlyArray +} + +export const runnable = (plan: Plan, done: ReadonlySet): Array => + plan.steps.filter( + (s) => !done.has(s.id) && s.dependsOn.every((d) => done.has(d)), + ) + +// ---------- Token control (durable, idempotent) ---------- + +export type Command = { type: "pause" | "resume" | "cancel"; author?: string } + +const foldCommands = (commands: Array): "active" | "paused" | "cancelled" => { + let state: "active" | "paused" | "cancelled" = "active" + for (const c of commands) { + if (c.type === "cancel") state = "cancelled" + else if (state !== "cancelled") state = c.type === "pause" ? "paused" : "active" + } + return state +} + +// ---------- Step executor signature ---------- + +export type StepExecutor = ( + step: Step, + ctx: { readonly seed: Seed; readonly token: string }, +) => Promise> + +// ---------- The Loop ---------- + +export interface AgentSpec { + readonly token: string + /** Topic where this agent's work transitions land (durable). */ + readonly workTopic: string + /** Decompose the current intent into a step DAG. */ + readonly orchestrate: (intent: unknown) => Promise + /** Execute one step. Physics lives here; everything else is protocol. */ + readonly executeStep: StepExecutor + /** LEARN hook: receive each outcome as it lands. May return knowledge envelopes. */ + readonly learn?: (outcome: { + readonly step: Step + readonly ok: boolean + readonly error?: unknown + readonly artifacts: ReadonlyArray<{ ref: string; hash?: string; kind?: string }> + }) => Promise +} + +const now = () => Date.now() + +async function readTokenState(seed: Seed, token: string): Promise<"active" | "paused" | "cancelled"> { + const control = `agent.${token}.control` + const commands: Array = seed.replay(control).map((e) => JSON.parse(e.body.payloadJSON ?? "{}")) + return foldCommands(commands) +} + +async function completedSteps(seed: Seed, workTopic: string): Promise> { + return new Set( + seed.query.byType(workTopic, "plan.step.completed").map((e) => { + try { + return (JSON.parse(e.body.payloadJSON ?? "{}") as { stepId?: string }).stepId ?? "" + } catch { + return "" + } + }), + ) +} + +function envelope(input: { + topic: string + parentSeq?: number + author: string + type: string + payload?: Record + artifacts?: Array<{ ref: string; hash?: string; kind?: string }> +}) { + return { + topic: input.topic, + ...(input.parentSeq === undefined ? {} : { parentSeq: input.parentSeq }), + author: input.author, + at: now(), + headers: [] as ReturnType[], + body: Body.make({ + type: input.type, + ...(input.payload ? { payloadJSON: JSON.stringify(input.payload) } : {}), + artifacts: (input.artifacts ?? []).map((a) => Artifact.make(a)), + }), + } +} + +/** Run one full OELBR revolution for an intent. Returns summary. */ +export async function runLoop( + seed: Seed, + spec: AgentSpec, + intent: unknown, +): Promise<{ + readonly status: "done" | "paused-out" | "cancelled" + readonly doneSteps: ReadonlySet +}> { + const control = `agent.${spec.token}.control` + const done = await completedSteps(seed, spec.workTopic) + + // ORCHESTRATE - decompose into a DAG (fresh instance reuses ledger's progress) + let plan = await spec.orchestrate(intent) + seed.publish({ + ...envelope({ topic: spec.workTopic, author: spec.token, type: "plan.declared", + payload: { steps: plan.steps } }), + // parent to latest if resuming mid-conversation + ...(seed.replay(spec.workTopic).slice(-1)[0] !== undefined + ? { parentSeq: seed.replay(spec.workTopic).slice(-1)[0]!.seq } + : {}), + }) + + // EXECUTE - dependency order, honoring the token between steps + for (;;) { + const tokenState = await readTokenState(seed, spec.token) + if (tokenState === "cancelled") + return { status: "cancelled", doneSteps: done } + if (tokenState === "paused") { + await new Promise((r) => setTimeout(r, 50)) // paused: cheap subscription wait + continue + } + const next = runnable(plan, done)[0] + if (!next) break + + const started = seed.publish({ + ...envelope({ topic: spec.workTopic, author: spec.token, type: "plan.step.started", + payload: { stepId: next.id, title: next.title } }), + parentSeq: seed.replay(spec.workTopic).slice(-1)[0]?.seq, + }) + + let artifacts: Array<{ ref: string; hash?: string; kind?: string }> = [] + let failure: unknown + try { + artifacts = await spec.executeStep(next, { seed, token: spec.token }) + } catch (err) { + failure = err + } + + if (failure !== undefined) { + seed.publish({ + ...envelope({ topic: spec.workTopic, author: spec.token, type: "plan.step.failed", + payload: { stepId: next.id, error: String(failure) }, parentSeq: started.seq }), + }) + if (spec.learn) await spec.learn({ step: next, ok: false, error: failure, artifacts: [] }) + return { status: "cancelled", doneSteps: done } // failed loop ends; lesson recorded + } + + seed.publish({ + ...envelope({ topic: spec.workTopic, author: spec.token, type: "plan.step.completed", + payload: { stepId: next.id }, artifacts, parentSeq: started.seq }), + }) + done.add(next.id) + if (spec.learn) await spec.learn({ step: next, ok: true, artifacts }) + } + + return { status: "done", doneSteps: done } +} + +/** Issue commands to a token. Durable append - cannot be missed. */ +export function command(seed: Seed, token: string, cmd: Command): void { + const control = `agent.${token}.control` + seed.topic({ id: control, durability: "durable", author: cmd.author ?? "system" }) + seed.publish({ + topic: control, + author: cmd.author ?? "system", + at: now(), + headers: [], + body: Body.make({ type: `token.${cmd.type}`, payloadJSON: JSON.stringify(cmd), artifacts: [] }), + }) +} diff --git a/runs/rung1/main.ts b/runs/rung1/main.ts new file mode 100644 index 0000000000..9871b8c7bb --- /dev/null +++ b/runs/rung1/main.ts @@ -0,0 +1,58 @@ +/** + * Rung 1 demo executable - isolated, standalone, own universe. + * bun runs/rung1/main.ts + * + * Proves: DAG execution, durable pause/resume/cancel, crash-resume from + * fold, and the loop's transitions all landing in the ledger. + */ + +import { Seed } from "../../packages/core/src/seed" +import { runLoop, command } from "./kernel" + +const dbPath = process.argv[2] ?? new URL("./universe.db", import.meta.url).pathname + +// --- two agents, one kernel each --- +const seed = Seed.open(dbPath) +seed.topic({ id: "agent.worker-1.work", durability: "durable", author: "orchestrator" }) +seed.topic({ id: "agent.worker-1.control", durability: "durable", author: "system" }) + +const worker = { + token: "worker-1", + workTopic: "agent.worker-1.work", + orchestrate: async () => ({ + steps: [ + { id: "s1", title: "survey ground", dependsOn: [] }, + { id: "s2", title: "dig foundation", dependsOn: ["s1"] }, + { id: "s3", title: "raise walls", dependsOn: ["s2"] }, + ], + }), + executeStep: async (step) => { + console.log(` [worker-1] executing: ${step.title}`) + return [{ ref: `artifacts/${step.id}.md`, kind: "output" }] + }, +} + +async function main() { + const mode = process.argv[2] ?? "fresh" + const db = mode === "resume" ? dbPath : dbPath + + if (mode === "cancel-mid") { + // cancel between steps 1 and 2 by issuing command after first completes + const seedX = Seed.open(db) + setTimeout(() => command(seedX, "worker-1", { type: "cancel", author: "will" }), 30) + } + + if (mode === "pause-resume") { + const seedX = Seed.open(db) + setTimeout(() => command(seedX, "worker-1", { type: "pause" }), 20) + setTimeout(() => command(seedX, "worker-1", { type: "resume" }), 120) + } + + console.log(`mode=${mode}`) + const result = await runLoop(Seed.open(db), worker, { goal: "build the thing" }) + console.log("status:", result.status, "| completed:", [...result.doneSteps].sort().join(",")) + console.log("ledger:") + for (const e of Seed.open(db).replay("agent.worker-1.work")) console.log(` ${e.seq}. ${e.body.type}`) +} + +main() diff --git a/specs/event-bus.md b/specs/event-bus.md new file mode 100644 index 0000000000..a601d7a6f2 --- /dev/null +++ b/specs/event-bus.md @@ -0,0 +1,310 @@ +# SPEC: The EventBus Architecture +Status: DESIGN AGREED 2026-08-22 (whiteboard session, Will + agent) +Next: Rung 0 implementation ("The Seed") + +--- + +## 0. Thesis + +The message is the entire application. + +State is never stored; it is folded from append-only streams. Nothing is +ever overwritten - messages append, headers inject, schemas coexist, +artifacts supersede. There is no operation anywhere in the grammar whose +input is "the past." + +One idea, fractal at every scale: **state = fold(append-only sequence).** + +--- + +## 1. The Bus + +One generic EventBus. Process-global, dumb wire. + +```ts +publish(topic, type, payload?) // fire-and-forget +emit(topic, type, payload?) // async, awaits handlers +subscribe(topic, handler) -> unsub // handler receives Envelope +next(topic, filter?, signal?) // await next matching envelope +``` + +- Topics are opaque strings. Callers own naming: orchestration IDs, + session IDs, token IDs. +- Delivery is best-effort in-process. Durability belongs to whoever + writes durable rows BEFORE publishing. The bus is not a database; + the message stream IS the system of record. +- Transport-pluggable by design: same envelopes ride in-proc calls, + IPC, queues, cloud endpoints, edge devices. Envelopes carry zero + transport assumptions. + +## 2. The Envelope (wire format) + +Four schema planes, each evolving independently, all content-addressed: + +| Plane | What | Versioning | +|---|---|---| +| 0 | Envelope format itself (meta-schema) | pinned root: `event-bus-envelope-v1`; amend loudly | +| 1 | Header values | per-header SchemaRef | +| 2 | Body contract (type + artifact rules) | body-level ref | +| 3 | Payloads | per message-type ref; nested objects too | + +Schema refs are content-addressed (`id = hash(definition)`), optionally +inline for small payloads. Schemas are durable first-class citizens: +written once, immutable, resolvable dynamically by any participant. +Old messages always decode. Replay works across versions, codebases, +decades. + +```ts +Envelope { + topic // durable handle + seq // position in topic's append-only sequence + parentSeq? // causal spine - what this responds to/extends + id // hash of canonical form + author // token id / orchestrator id / stage name + at // epoch ms + + headers: [ { key, value, author, at, schema? } ] // ORDERED LIST + // inject-only + body { + type // e.g. "plan.step.completed" + payloadJSON? // decoded via payloadSchema + payloadSchema? // Plane 3 + schema? // Plane 2 (body contract) + artifacts[] // { ref, hash?, kind? } - evidence rides with record + } +} +``` + +Header law: +- Ordered LIST (duplicates accumulate; order = causality). +- Inject-only. Modification/removal structurally impossible by API shape. +- `current(key)` = last-wins fold. `history(key)` = full evolution. +- Corrections are new injections with reasons, never edits. +- Relationship semantics live here; they are NOT static in the world - + roles get revised by later injection, both readings stay true. + +## 3. Topics + +Topics declare their own persistence contract at creation: + +- DURABLE: every envelope appended; replayable; recoverable. +- EPHEMERAL: delivery-only, never persisted; optional retainLast; + optional ttl after zero subscribers. + +Decision rule: if losing one envelope corrupts state or breaks recovery, +it is durable. If its value expires after delivery, it is ephemeral. + +Durable set (the ledger): +- `orchestration.{id}.lifecycle` created/decomposed/completed/cancelled +- `orchestration.{id}.packages` decomposition + dependency DAG + resources +- `orchestration.{id}.clearance` GO/no-go rulings, renegotiations +- `agent.{token}.plan` declared step plans +- `agent.{token}.steps` started/artifact/completed/failed + (+token counts) = checkpoint journal + AND economics ledger +- `agent.{token}.control` pause/resume/cancel. DURABLE on purpose: + a cancellation must never evaporate. + Idempotent receivers make replay safe. +- `session.{id}.steer` prompts admitted during active drains +- conversation topics a conversation IS a durable topic + (see section 8) + +Ephemeral set (the air): +- `stream.{requestId}` LLM token streams (retainLast: no) +- `presence.{token}` heartbeats; absence = crash signal +- `progress.{orchestration}` smoothed UI gauges (retainLast: yes) +- `scratch.{pair}` transient sibling-thread hand-offs + +The durable set alone reconstructs everything. Ephemeral loss is never +data loss. + +## 4. Persistence ontology (four layers) + +``` +1. ENVELOPE STREAM the state. The only reality. All durability here. +2. FOLDED VIEWS derived state. Always recomputable from (1). +3. ARTIFACTS materialized projections, content-addressed. + Live in the SAME store as the stream. +4. FILES ON DISK exports of projections. Printouts, not truth. +``` + +Loss below layer 1 is rendering loss, recoverable by re-derivation. +Artifacts EVOLVE BY SUPERSESSION: no overwrites ever. Forks are legal +(multiple children of one parent); settling a fork = one group- +supersession append naming the family. Git semantics emerge as a theorem. + +The store of record holds: topic logs + artifact blobs + schema registry. +Three things total. Everything else is derived, cached, or ephemeral. + +## 5. Orchestration protocol (six beats) + +``` +1. ASSIGN orchestrator -> agent: package + handle +2. DECLARE agent plans its own steps (plan.declared), WAITS +3. CLEARANCE orchestrator lays all plans side by side: + resource conflicts? duplicate work? contradicting + dependencies? sane estimates? + -> GO per agent, or renegotiate while it's still data +4. EXECUTE cleared agents run, publishing transitions upward +5. REPORT continuous: status, artifacts, token burn +6. RECOVER crash = silence. Last folded envelope = resume pointer. + Completed steps' artifacts survive. Re-run only the + in-flight step (steps should be idempotent). +``` + +Key properties: +- The dependency graph IS the concurrency model. No locks anywhere. + Runnable = all edges satisfied. Parallelism discovered, not configured. +- Resource conflicts are missing edges; caught at CLEARANCE while + cheap, serialized by injecting an edge. +- Orchestrator is planner + economist + router + reviewer + accountant + + coroner. Just the one subscriber that sees every topic and folds. +- Agents plan their OWN work within packages; orchestrator reviews the + whole board before any of it moves. +- Pausing/waiting-for-siblings costs nothing: a subscription with no + matching events yet. Same mechanism as steer and cancel. + +### Agent spawning rule +Agents may spin up THREADS, never other AGENTS. Spawn authority stays +central with the orchestrator. Enforced at API shape (agent handles +expose spawnThread, not spawnAgent) plus runtime check. + +### Cancellation tokens +Every agent carries one; control messages address tokens directly. +Pause/resume/cancel individual agents mid-flight without killing the +orchestration. Tokens are durable-topics subscribers; commands cannot +be missed, only late. + +## 6. Economics + +Every step reports exact token burn (tokensIn/tokensOut/toolCalls) +in its completion envelopes - written at the boundary, not scraped +from provider logs afterward. + +Fan-out decision per package node: +- delegate only if work W comfortably exceeds coordination tax C + (context injection + scaffolding + report-back + synthesis) +- sequential-if-chained: independence must buy wall-clock time +- DO-IT-MYSELF INLINE is a legitimate third option + +Budgets watched live via fold; descope/collapse/cancel interventions +happen mid-flight while cheap. Historical step-cost table accumulates +into an empirical planner: which work types fan out profitably, which +never do. + +## 7. AOP and the seam + +The bus MANUFACTURES the universal seam as a side effect of existing. +All coordination crosses it; therefore all coordination is interceptible. + +Cross-cutting concerns become pipeline stages / wire-taps, not scattered +code: +- telemetry = Wire Tap subscriber on topic:* +- auth = filter stage between publish and deliver; decisions + stamped into headers ({authz: granted, principal}) so + audit trail IS the wire +- metering = header injection at publish boundaries +- retry, dead-lettering, validation, error-mapping = stackable stages + +Decorator discipline (the anti-WCF rule): DECORATORS WIRE, NEVER WORK. +Six-word vocabulary target: @agent @orchestrator @plan @step @on(stage) +plus pipeline stages (@metered @guard). If logic creeps into a decorator, +push it into the handler or the pipeline. No config files, no parallel +configuration universe - declaration lives on the thing it declares. + +Participation rule: anything that matters crosses the bus; internals are +free. Cross a boundary naked and you are unmeasured, unaudited, +uncancellable. + +## 8. Conversation is a topic + +A conversation with an assistant is an orchestration: +- human publishes assignment envelopes (prompts = steers) +- agent publishes plan/step/artifact/completion envelopes +- tool calls are threads under the agent's token +- context window = ephemeral cache; the topic log = durable truth +- compaction = folding the log, not summarizing away history +- retrieval replaces recollection: fetch slices by topic/type/time/ + causality instead of re-injecting whole conversations +- crash/context-death = silence; next instance folds and resumes + +This makes sessions deathless: handoff rituals dissolve because the +log already contains everything. + +## 9. Backlogs and projects (fold patterns, not entities) + +There are no entities with fields - only streams and agreed folds. +A backlog item is a fold pattern over intention/work envelopes on +`project.{name}.backlog`. + +Reading conventions (vocabulary, not schema): +- proposed -> ready -> blocked -> active -> done | dropped (folded) +- dependsOn: read from envelopes; blocked iff any dependency not done +- artifacts: linked by publication + injected {item} headers; roles + evolve by revision-injection (spec -> evidence etc.) +- graduation: activating an item mints an orchestration id; the item + folds to done when that topic completes + +Relationship law: RELATIONSHIPS ARE READ FROM HISTORY, NOT WRITTEN AS +RECORDS. The causal spine is the graph. Edge types come from the +vocabulary of envelope types and header keys (caused-by=parentSeq, +authored-by=author, lives-in=topic, produced=artifacts, supersedes= +supersede envelopes, regards=injected headers). New relationship kinds = +new vocabulary + a fold. The graph grows by vocabulary, never migration. + +Payload law: PAYLOADS HOLD ENTITIES; EDGES ARE ENVELOPES; COLLECTIONS +ARE ALWAYS FOLDS. Any list-of-references field inside a payload is the +smell. + +## 10. Bootstrap ladder + +RUNG 0 - THE SEED (first build; smallest thing two people can run) +- EventBus (exists) + DurableTopicLog: single owned SQLite store, + content-addressed envelopes exactly per section 2, fold/replay, + four query verbs: by-topic, by-type, by-time-range, walk-by-causality +- Conversation mapping live: our own working sessions run as + orchestrations. Deliverable proof: context death loses nothing. +- Discipline: if Rung 0 needs more than a few hundred lines, it is + smuggling Rung 1 concerns. + +RUNG 1 - SELF-HOSTING AGENTS +- tokens, plans/DAGs, decorators. Built BY running Rung 0 sessions. + Development history written in the pipeline from here on. + +RUNG 2 - THE ORCHESTRATOR +- decomposition, clearance review, fan-out, token-economics break-evens. + Built through Rung 1 sessions; multiple parallel workstreams + coordinating their own construction. + +RUNG 3 - CONVERSION CAMPAIGNS +- enumerate the existing application as work packages; convert module + by module THROUGH the pipeline. Existing tables demoted to views over + envelopes, then deleted. retry.ts deleted (aspect replaces it). + +Discipline: never build a rung before standing on the previous one. +No speculative features; each layer earns the next by being used. + +## 11. First artifacts + +- citizen zero: this spec, stored as an artifact in the Seed's own + store at Rung 0 completion, superseded (never edited) thereafter. +- backlog seed: project.neuron items = Rungs 1-3; project.valley items + = pinned investigation threads (ishikawa continuation, testimonies + held open). + +--- + +## Appendix: lineage (nothing here is invented) + +Email Received: chains -> inject-only headers +Hohpe/Woolf EIP -> Process Manager = orchestrator, + Scatter-Gather = fan-out+fold, + Claim Check = artifact refs, + Control Bus = cancellation tokens, + Wire Tap = telemetry, + Dead Letter Channel = failed agents +WCF channel stack -> interceptible pipeline; kept the + structure, refused the config church +git -> fork/supersede artifact semantics +Event sourcing -> the entire persistence ontology +"The world changes; only liars rewrite. Append instead."