feat(neuron): generic process execution engine, orchestration with supersession, aspect kit

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<string, unknown>
+  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 })
+  }
+}

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
+ * A process is a node whose payload defines:
- * holds intentions. The orchestrator decomposes intent into executable
+ *   - name
- * steps. All of it lives as nodes and typed edges in one store.
+ *   - 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 ----
+  /** Register a process definition as a node in the graph. */
-
+  defineProcess(name: string, definition: {
-  learn(fact: string, source?: string): Node {
+    description: string
-    const addr = `knowledge:${Date.now().toString(36)}:${Math.random().toString(36).slice(2, 6)}`
+    /** Edges to read for input (by kind). */
-    const node = this.graph.ensureNode("knowledge", addr)
+    reads: string[]
-    this.graph.edge(node.id, node.id, "content", { fact, source })
+    /** Edge kind to write results as. */
-    return node
+    writes: string
+    /** The actual work. Injected by whoever composes this process. */
+    execute: (input: unknown) => Promise<unknown>
+  }): 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.
+   * Run a named process against a target node.
-   * Steps run in dependency order; each publishes its completion.
+   * Reads the process definition, gathers input edges, executes, writes output.
    */
-  async orchestrate(intent: string, executors: Map<string, (step: Step) => Promise<string>>): Promise<Array<{ step: string; result: string }>> {
+  async runProcess(processName: string, targetNodeID: string): Promise<unknown> {
-    const steps = this.decompose(intent)
+    const defEdge = this.graph
-    const done = new Set<string>()
+      .incident(this.graph.ensureNode("process-definition", `process:${processName}`).id, "definition")
-    const results: Array<{ step: string; result: string }> = []
+      .at(-1)
+    if (!defEdge) throw new Error(`process "${processName}" not defined`)
+    const def = this.graph.payload<{ description: string; reads: string[]; execute: (...args: any[]) => Promise<unknown> }>(defEdge)
+    if (!def) throw new Error(`process "${processName}" has no definition payload`)
 
-    while (done.size < steps.length) {
+    // Gather inputs from the target's incident edges matching the reads list
-      let progressed = false
+    const inputs: Record<string, unknown[]> = {}
-      for (const step of steps) {
+    for (const readKind of def.reads ?? []) {
-        if (done.has(step.id)) continue
+      inputs[readKind] = this.graph.outgoing(targetNodeID, readKind).map((e) => ({
-        if (!step.dependsOn.every((d) => done.has(d))) continue
+        edgeId: e.id,
-
+        payload: e.payload ? JSON.parse(e.payload) : undefined,
-        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")
-      }
     }
-    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[] }> {
+  // ---- generic node ops ----
-    // Simple sequential decomposition for now.
+
-    // Real decomposition will come from the LLM once wired to dialects.
+  ensure(kind: string, address: string): Node {
-    return [
+    return this.graph.ensureNode(kind, address)
-      { id: "step-1", dependsOn: [] },
+  }
-      { id: "step-2", dependsOn: ["step-1"] },
+
-    ]
+  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[]
+}

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<string, unknown[]>) => Promise<unknown>
+  }): 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<unknown> {
+    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<string, unknown[]> = {}
+    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<string, unknown[]>): Promise<unknown> {
+    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)
+  }
+}

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<string, unknown>,
+): 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<string, () => Promise<string>>,
+): Promise<Array<{ step: string; result: string }>> {
+  const results: Array<{ step: string; result: string }> = []
+  const done = new Set<string>()
+  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<string, unknown>, 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 }
+}