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
This commit is contained in:
@@ -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 <universe.db>
|
||||
*/
|
||||
|
||||
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<string>
|
||||
}
|
||||
|
||||
export interface Plan {
|
||||
readonly steps: ReadonlyArray<Step>
|
||||
}
|
||||
|
||||
export const runnable = (plan: Plan, done: ReadonlySet<string>): Array<Step> =>
|
||||
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<Command>): "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<Array<{ ref: string; hash?: string; kind?: string }>>
|
||||
|
||||
// ---------- 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<Plan>
|
||||
/** 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<void>
|
||||
}
|
||||
|
||||
const now = () => Date.now()
|
||||
|
||||
async function readTokenState(seed: Seed, token: string): Promise<"active" | "paused" | "cancelled"> {
|
||||
const control = `agent.${token}.control`
|
||||
const commands: Array<Command> = seed.replay(control).map((e) => JSON.parse(e.body.payloadJSON ?? "{}"))
|
||||
return foldCommands(commands)
|
||||
}
|
||||
|
||||
async function completedSteps(seed: Seed, workTopic: string): Promise<Set<string>> {
|
||||
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<string, unknown>
|
||||
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<typeof MessageHeader.make>[],
|
||||
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<string>
|
||||
}> {
|
||||
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: [] }),
|
||||
})
|
||||
}
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user