diff --git a/packages/neuron/src/kernel/artifacts.ts b/packages/neuron/src/kernel/artifacts.ts new file mode 100644 index 0000000000..6d6aae1a28 --- /dev/null +++ b/packages/neuron/src/kernel/artifacts.ts @@ -0,0 +1,88 @@ +export * as Artifacts from "./artifacts" + +import { Graph } from "./graph" + +/** + * Artifacts are nodes with a schema, a format, and a payload. + * + * Schemas are themselves nodes — first-class, content-addressed, + * superseding each other. An artifact carries a `conforms-to` edge to + * the schema version it was written against. Fetching a schema walks + * to the latest non-superseded version. + * + * Payloads are opaque blobs. The `format` field tells consumers how to + * hydrate: "markdown", "json", "pdf", "html", "png", "schema.json", ... + */ + +export function create(graph: Graph, opts: { + format: string + metadata: Record + payload: string +}): { nodeID: string; hash: string } { + const hash = Bun.hash(opts.payload).toString(36) + const addr = `artifact:${hash}` + const node = graph.ensureNode("artifact", addr) + + graph.edge(node.id, node.id, "metadata", { format: opts.format, ...opts.metadata }) + graph.edge(node.id, node.id, "payload", { blob: opts.payload }) + + return { nodeID: node.id, hash } +} + +/** Create a schema artifact. Same semantics as any other artifact. */ +export function createSchema(graph: Graph, name: string, definition: string): { nodeID: string; hash: string } { + const result = create(graph, { + format: "schema.json", + metadata: { name }, + payload: definition, + }) + return result +} + +/** Supersede an old schema version with a new one. */ +export function supersedeSchema(graph: Graph, oldSchemaNodeID: string, newSchemaNodeID: string) { + graph.edge(newSchemaNodeID, oldSchemaNodeID, "supersedes", {}) +} + +/** + * Resolve the latest schema for an artifact: walk `conforms-to` to the + * referenced schema node, then walk `supersedes` forward to the latest. + */ +export function resolveSchema(graph: Graph, artifactNodeID: string): { definition: string; format: string } | undefined { + const conformsTo = graph.outgoing(artifactNodeID, "conforms-to").at(-1) + if (!conformsTo) return undefined + const edges = graph.outgoing(conformsTo.to_node, "payload") + const payloadEdge = edges.at(-1) + if (!payloadEdge) return undefined + const p = graph.payload<{ blob: string }>(payloadEdge) + if (!p) return undefined + // Check for supersession chain + let latest = conformsTo.to_node + let supersedes = graph.outgoing(latest, "supersedes") + while (supersedes.length > 0) { + latest = supersedes[0].to_node + supersedes = graph.outgoing(latest, "supersedes") + } + const payloadEdges = graph.outgoing(latest, "payload") + const lastPayload = payloadEdges.at(-1) + if (!lastPayload) return undefined + const data = graph.payload<{ blob: string }>(lastPayload) + return data ? { definition: data.blob, format: "schema.json" } : undefined +} + +export function forkFrom(graph: Graph, parentID: string, newContent: string) { + const created = create(graph, { format: "text", metadata: {}, payload: newContent }) + graph.edge(created.nodeID, parentID, "forked-from", {}) + return created +} + +export function supersedes(graph: Graph, winnerID: string, ...losers: string[]) { + for (const loser of losers) graph.edge(winnerID, loser, "supersedes", {}) +} + +export function settles(graph: Graph, winnerID: string, familyIDs: string[], reason?: string) { + for (const id of familyIDs) { + if (id === winnerID) continue + graph.edge(winnerID, id, "settles", { reason }) + } +} diff --git a/packages/neuron/src/kernel/kernel.ts b/packages/neuron/src/kernel/kernel.ts index 6359815f37..422c68c7a9 100644 --- a/packages/neuron/src/kernel/kernel.ts +++ b/packages/neuron/src/kernel/kernel.ts @@ -1,14 +1,18 @@ - +export * as KernelOps from "./kernel" import { Graph } from "./graph" -import type { Node } from "./graph" /** - * Domain operations on the graph. + * Generic process execution over the graph. * - * Knowledge is what you learned. A project organises work. A backlog - * holds intentions. The orchestrator decomposes intent into executable - * steps. All of it lives as nodes and typed edges in one store. + * A process is a node whose payload defines: + * - name + * - input requirements (which edges to read) + * - an executable step (registered by category, resolved at runtime) + * + * The kernel does NOT know what "learn" or "build" mean. It reads the + * process node, resolves its dependencies, runs it, writes results back + * as edges. New processes are data — add a node, not a function. */ export class Kernel { @@ -18,101 +22,78 @@ export class Kernel { this.graph = graph } - // ---- knowledge ---- - - learn(fact: string, source?: string): Node { - const addr = `knowledge:${Date.now().toString(36)}:${Math.random().toString(36).slice(2, 6)}` - const node = this.graph.ensureNode("knowledge", addr) - this.graph.edge(node.id, node.id, "content", { fact, source }) - return node + /** Register a process definition as a node in the graph. */ + defineProcess(name: string, definition: { + description: string + /** Edges to read for input (by kind). */ + reads: string[] + /** Edge kind to write results as. */ + writes: string + /** The actual work. Injected by whoever composes this process. */ + execute: (input: unknown) => Promise + }): void { + const procNode = this.graph.ensureNode("process-definition", `process:${name}`) + const existingEdges = this.graph.incident(procNode.id, "definition") + if (existingEdges.length === 0) { + this.graph.edge(procNode.id, procNode.id, "definition", { ...definition }) + } } - linkKnowledge(knowledgeNodeID: string, targetNodeID: string, relationship: string) { - this.graph.edge(targetNodeID, knowledgeNodeID, relationship, {}) - } - - queryKnowledge(): Array<{ id: string; fact: string; source?: string }> { - return (this.graph as any).db - .query(`SELECT n.id, e.payload FROM nodes n JOIN edges e ON e.from_node = n.id AND e.kind = 'content' WHERE n.kind = 'knowledge' ORDER BY e.at DESC`) - .all() - .map((row: any) => ({ - id: row.id, - fact: JSON.parse(row.payload)?.fact ?? "", - source: JSON.parse(row.payload)?.source, - })) - } - - // ---- backlog ---- - - createBacklog(name: string): Node { - return this.graph.ensureNode("backlog", `backlog:${name}`) - } - - addItem(backlogName: string, text: string): Node { - const backlog = this.graph.getNode(`backlog:${backlogName}`) - if (!backlog) throw new Error(`backlog not found: ${backlogName}`) - const item = this.graph.ensureNode("backlog-item", `item:${crypto.randomUUID()}`) - this.graph.edge(backlog.id, item.id, "contains", { text }) - this.graph.edge(item.id, backlog.id, "status", { state: "proposed" }) - return item - } - - setStatus(itemAddress: string, state: string) { - const item = this.graph.getNode(itemAddress) - if (!item) throw new Error(`item not found: ${itemAddress}`) - this.graph.edge(item.id, item.id, "status", { state }) - } - - backlogItems(backlogName: string): Array<{ id: string; state: string; text: string }> { - const backlog = this.graph.getNode(`backlog:${backlogName}`) - if (!backlog) return [] - return this.graph.outgoing(backlog.id, "contains").map((e) => { - const item = this.graph.getNode(e.to_node) - if (!item) return { id: "", state: "", text: "" } - const statusEdges = this.graph.incident(item.id, "status") - const lastStatus = statusEdges.length > 0 ? JSON.parse(statusEdges[statusEdges.length - 1].payload || "{}") : {} - const payload = this.graph.incident(backlog.id, "contains").find((c) => c.to_node === item.id) - const text = payload ? JSON.parse(payload.payload || "{}").text : "" - return { id: item.id, state: lastStatus?.state ?? "proposed", text } - }).filter((x) => x.id !== "") - } - - // ---- orchestrator ---- - /** - * Decompose intent into steps, execute each, learn from outcomes. - * Steps run in dependency order; each publishes its completion. + * Run a named process against a target node. + * Reads the process definition, gathers input edges, executes, writes output. */ - async orchestrate(intent: string, executors: Map Promise>): Promise> { - const steps = this.decompose(intent) - const done = new Set() - const results: Array<{ step: string; result: string }> = [] + async runProcess(processName: string, targetNodeID: string): Promise { + const defEdge = this.graph + .incident(this.graph.ensureNode("process-definition", `process:${processName}`).id, "definition") + .at(-1) + if (!defEdge) throw new Error(`process "${processName}" not defined`) + const def = this.graph.payload<{ description: string; reads: string[]; execute: (...args: any[]) => Promise }>(defEdge) + if (!def) throw new Error(`process "${processName}" has no definition payload`) - while (done.size < steps.length) { - let progressed = false - for (const step of steps) { - if (done.has(step.id)) continue - if (!step.dependsOn.every((d) => done.has(d))) continue - - const executor = executors.get(step.id) - const result = executor ? await executor(step) : `step ${step.id} completed` - results.push({ step: step.id, result }) - done.add(step.id) - progressed = true - } - if (!progressed && done.size < steps.length) { - throw new Error("dependency cycle in plan") - } + // Gather inputs from the target's incident edges matching the reads list + const inputs: Record = {} + for (const readKind of def.reads ?? []) { + inputs[readKind] = this.graph.outgoing(targetNodeID, readKind).map((e) => ({ + edgeId: e.id, + payload: e.payload ? JSON.parse(e.payload) : undefined, + })) } - return results + + const result = await def.execute(inputs) + + // Write results back as edges on the target node + if (result !== undefined && result !== null) { + this.graph.edge(targetNodeID, targetNodeID, `result:${processName}`, { data: result }) + } + + return result } - private decompose(intent: string): Array<{ id: string; dependsOn: string[] }> { - // Simple sequential decomposition for now. - // Real decomposition will come from the LLM once wired to dialects. - return [ - { id: "step-1", dependsOn: [] }, - { id: "step-2", dependsOn: ["step-1"] }, - ] + // ---- generic node ops ---- + + ensure(kind: string, address: string): Node { + return this.graph.ensureNode(kind, address) + } + + find(kind: string, address: string): Node | undefined { + const n = this.graph.getNode(address) + return n && n.kind === kind ? n : undefined + } + + byKind(kind: string): Node[] { + const db = (this.graph as any).db + if (!db) return [] + return db.query(`SELECT * FROM nodes WHERE kind = ? ORDER BY created_at`).all(kind) ?? [] + } + + edge(from: string, to: string, kind: string, payload?: unknown) { + return this.graph.edge(from, to, kind, payload) } } + +export interface Step { + id: string + description: string + dependsOn: string[] +} diff --git a/packages/neuron/src/kernel/orchestration.ts b/packages/neuron/src/kernel/orchestration.ts new file mode 100644 index 0000000000..6722d2f697 --- /dev/null +++ b/packages/neuron/src/kernel/orchestration.ts @@ -0,0 +1,93 @@ +export * as Orchestration from "./orchestration" + +import { Graph } from "./graph" + +/** + * Every operation is a process. Processes are nodes in the graph. + * One generic executor runs them all. + * + * Results are edges on the target node. New results supersede old + * results from the same process — automatically. Old results stay + * walkable forever. + * + * This covers everything: building artifacts, producing knowledge, + * refining processes, refining knowledge. They are all just + * "run process P against target T" with different definitions. + */ + +export class Orchestrator { + private graph: Graph + + constructor(graph: Graph) { + this.graph = graph + } + + /** Register any process definition. */ + define(name: string, def: { + description: string + reads: string[] + execute: (inputs: Record) => Promise + }): void { + const procNode = this.graph.ensureNode("process-definition", `process:${name}`) + if (this.graph.incident(procNode.id, "definition").length === 0) { + this.graph.edge(procNode.id, procNode.id, "definition", { ...def }) + } + } + + /** + * Run any named process against a target node. + * + * Automatically supersedes prior results from the same process on the + * same target. Old results stay in the graph walkable forever. + */ + async run(processName: string, targetNodeID: string): Promise { + const procAddr = `process:${processName}` + const procNode = this.graph.getNode(procAddr) + if (!procNode) throw new Error(`process not found: ${processName}`) + + // Gather inputs from target's edges matching what this process reads + const inputs: Record = {} + for (const e of this.graph.outgoing(targetNodeID)) { + const p = e.payload ? JSON.parse(e.payload) : undefined + if (!p || typeof p !== "object") continue + for (const readKind of Object.keys(inputs)) { + if (e.kind === readKind && !inputs[readKind].includes(p)) { + inputs[readKind].push(p) + } + } + } + + const result = await this.executeDefinition(processName, inputs) + + // Write result, superseding any prior result from the same process + this.writeResult(targetNodeID, processName, result) + + return result + } + + private writeResult(targetNodeID: string, processName: string, result: unknown): void { + // Supersede prior result from same process on same target + const priorKind = `result:${processName}` + const priors = this.graph.outgoing(targetNodeID, priorKind) + for (const prior of priors) { + this.graph.edge(prior.to_node ?? prior.from_node, prior.from_node, "supersedes", + { process: processName, supersededAt: Date.now() }) + } + + // Write new result + this.graph.edge(targetNodeID, targetNodeID, priorKind, { data: result }) + } + + private async executeDefinition(processName: string, inputs: Record): Promise { + const procAddr = `process:${processName}` + const procNode = this.graph.getNode(procAddr) + if (!procNode) throw new Error(`process not found: ${procAddr}`) + + const defEdges = this.graph.incident(procNode.id, "definition") + const lastDef = defEdges.at(-1) + if (!lastDef?.payload) throw new Error(`no definition for process ${processName}`) + const def = JSON.parse(lastDef.payload) + if (!def.handler) throw new Error(`process ${processName} has no handler`) + return def.handler(inputs) + } +} diff --git a/packages/neuron/src/kernel/process.ts b/packages/neuron/src/kernel/process.ts new file mode 100644 index 0000000000..40e6238432 --- /dev/null +++ b/packages/neuron/src/kernel/process.ts @@ -0,0 +1,148 @@ +export * as Process from "./process" + +import { Graph } from "./kernel/graph" +import { create } from "./kernel/artifacts" + +/** + * The five core processes. Every orchestration runs all five in order. + * Each produces nodes and edges in the graph — the record of what was + * done, what was learned, and what was built. + * + * ORCHESTRATE decompose intent into a step DAG + * EXECUTE run runnable steps; publish every transition + * LEARN fold outcomes into knowledge nodes + * REFINE supersede own procedure with what was learned + * BUILD emit new artifacts from refined understanding + * + * The cycle loops: BUILD feeds the next ORCHESTRATE. + */ + +export const PROCESS_KINDS = ["orchestrate", "execute", "learn", "refine", "build"] as const +export type ProcessKind = typeof PROCESS_KINDS[number] + +interface Step { + id: string + description: string + status: "pending" | "running" | "done" | "failed" +} + +export interface ProcessRecord { + orchestrationID: string + processKind: ProcessKind + nodeID: string + startedAt: number +} + +/** + * Start a full five-process cycle for an intent. + * Returns the orchestration ID so callers can query results later. + */ +export function begin(graph: Graph, sessionID: string, intent: string): string { + const orchID = `orch-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 6)}` + const orchNode = graph.ensureNode("orchestration", `orchestration:${orchID}`) + + // Link the orchestration to its originating conversation/session + const sessionNode = graph.getNode(`session:${sessionID}`) + if (sessionNode) graph.edge(sessionNode.id, orchNode.id, "orchestrates", { intent }) + + // Create a node per process phase, chained in order + let prev = orchNode.id + for (const kind of PROCESS_KINDS) { + const addr = `${kind}:${orchID}` + const node = graph.ensureNode("process", addr) + graph.edge(prev, node.id, kind === "orchestrate" ? "begins" : "follows", { orchestration: orchID, phase: kind }) + prev = node.id + } + + return orchID +} + +/** Mark a phase complete and record its output. */ +export function complete( + graph: Graph, + orchID: string, + kind: ProcessKind, + result: Record, +): void { + const addr = `${kind}:${orchID}` + const node = graph.getNode(addr) + if (!node) throw new Error(`process node not found: ${addr}`) + graph.edge(node.id, node.id, "completed", result) +} + +// ---- ORCHESTRATE ---- + +export interface Step { + id: string + description: string + dependsOn: string[] +} + +export function orchestrate(graph: Graph, orchID: string, steps: Step[]): void { + complete(graph, orchID, "orchestrate", { steps }) + for (const step of steps) { + const stepNode = graph.ensureNode("step", `step:${orchID}:${step.id}`) + for (const dep of step.dependsOn) { + const depNode = graph.ensureNode("step", `step:${orchID}:${dep}`) + graph.edge(depNode.id, stepNode.id, "depends-on", {}) + } + } +} + +// ---- EXECUTE ---- + +export async function execute( + graph: Graph, + orchID: string, + executors: Map Promise>, +): Promise> { + const results: Array<{ step: string; result: string }> = [] + const done = new Set() + const steps = graph.outgoing(graph.getNode(`${"process"}:${orchID}`)?.id ?? "", "depends-on") + + while (done.size < Object.keys(executors).length || done.size < steps.length) { + let progressed = false + for (const [id, executor] of executors) { + if (done.has(id)) continue + const stepNode = graph.getNode(`step:${orchID}:${id}`) + if (!stepNode) continue + const deps = graph.outgoing(stepNode.id, "depends-on") + if (!deps.every((d) => done.has(d.to_node))) continue + + const result = await executor() + results.push({ step: id, result }) + done.add(id) + progressed = true + break + } + if (!progressed) break + } + return results +} + +// ---- LEARN ---- + +export function learn(graph: Graph, orchID: string, facts: string[]): void { + complete(graph, orchID, "learn", { facts }) + for (const fact of facts) { + const addr = `knowledge:${Date.now().toString(36)}:${Math.random().toString(36).slice(2, 6)}` + const kn = graph.ensureNode("knowledge", addr) + graph.edge(kn.id, kn.id, "content", { fact }) + } +} + +// ---- BUILD ---- + +export function buildArtifact(graph: Graph, orchID: string, format: string, metadata: Record, payload: string): { nodeID: string; hash: string } { + const hash = Bun.hash(payload).toString(36) + const addr = `artifact:${hash}` + const node = graph.ensureNode("artifact", addr) + graph.edge(node.id, node.id, "metadata", { format, ...metadata }) + graph.edge(node.id, node.id, "payload", { blob: payload }) + + const procAddr = `build:${orchID}` + const procNode = graph.ensureNode("process", procAddr) + graph.edge(procNode.id, node.id, "produced", {}) + + return { nodeID: node.id, hash } +}