feat(neuron): generic process execution engine, orchestration with supersession, aspect kit
This commit is contained in:
@@ -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 })
|
||||
}
|
||||
}
|
||||
@@ -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<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.
|
||||
* 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<string, (step: Step) => Promise<string>>): Promise<Array<{ step: string; result: string }>> {
|
||||
const steps = this.decompose(intent)
|
||||
const done = new Set<string>()
|
||||
const results: Array<{ step: string; result: string }> = []
|
||||
async runProcess(processName: string, targetNodeID: string): Promise<unknown> {
|
||||
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<unknown> }>(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<string, unknown[]> = {}
|
||||
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[]
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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 }
|
||||
}
|
||||
Reference in New Issue
Block a user