refactor: CPM and estimation become process definitions, not code modules

This commit is contained in:
2026-08-22 23:50:13 -05:00
parent 6bebf987f1
commit 51de2d22ca
6 changed files with 496 additions and 119 deletions
+150 -54
View File
@@ -3,18 +3,29 @@ export * as KernelOps from "./kernel"
import { Graph } from "./graph"
/**
* Generic process execution over the graph.
* The unified kernel. One register function, one execute function,
* generic node queries. Every concept - tool, process, knowledge,
* artifact, conversation, memory - is a node registered here.
*
* 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.
* No parallel registries. No per-concept files. One store, one truth.
*/
export interface Node {
id: string
kind: string
address: string
created_at: number
}
export interface Edge {
id: string
from_node: string
to_node: string
kind: string
at: number
payload: string | null
}
export class Kernel {
private graph: Graph
@@ -22,78 +33,163 @@ export class Kernel {
this.graph = graph
}
/** 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 })
/**
* Register anything. Kind determines what it is;
* definition edges carry the specifics.
*/
register(kind: string, address: string, edges: Array<{ kind: string; payload?: unknown }>): Node {
const node = this.graph.ensureNode(kind, address)
for (const e of edges) {
this.graph.edge(node.id, node.id, e.kind, e.payload)
}
return node
}
/**
* Run a named process against a target node.
* Reads the process definition, gathers input edges, executes, writes output.
* Execute a process node against a target node.
* Reads the process's definition edge for its handler and input kinds.
*/
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`)
async execute(processNodeID: string, targetNodeID: string): Promise<unknown> {
const defEdges = this.graph.outgoing(processNodeID, "definition")
const lastDef = defEdges.at(-1)
if (!lastDef || !lastDef.payload) throw new Error(`process ${processNodeID} has no definition`)
const def = JSON.parse(lastDef.payload)
// Gather inputs from the target's incident edges matching the reads list
// Gather inputs from target's edges matching 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,
kind: e.kind,
payload: e.payload ? JSON.parse(e.payload) : undefined,
}))
}
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 })
// Supersede prior results from same process on same target
const priorKind = "result:" + processNodeID
const priors = this.graph.outgoing(targetNodeID).filter(function(e) {
return e.kind === priorKind
})
for (const prior of priors) {
this.graph.edge(targetNodeID, targetNodeID, "supersedes", { supersededBy: processNodeID })
}
this.graph.edge(targetNodeID, targetNodeID, priorKind, { data: result })
return result
}
// ---- 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
}
// ---- generic queries ----
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) ?? []
return this.graph.nodesByKind(kind)
}
find(address: string): Node | undefined {
return this.graph.getNode(address)
}
outgoing(nodeID: string, kind?: string): Edge[] {
return this.graph.outgoing(nodeID, kind)
}
incident(nodeID: string, kind?: string): Edge[] {
return this.graph.incident(nodeID, kind)
}
edge(from: string, to: string, kind: string, payload?: unknown) {
return this.graph.edge(from, to, kind, payload)
}
// ---- generic read/write ----
/**
* Read any node's content, hydrated according to its format.
* Returns metadata, payload, and the resolved schema if one exists.
*/
async read(nodeID: string): Promise<{
metadata: Record<string, unknown>
payload: unknown
format: string | undefined
schemaDefinition: string | undefined
} | null> {
const node = this.graph.getNode(nodeID)
if (!node) return null
let metadata: Record<string, unknown> = {}
let payloadRaw = ""
let format: string | undefined
for (const e of this.graph.outgoing(nodeID)) {
const p = e.payload ? JSON.parse(e.payload) : undefined
if (!p) continue
if (e.kind === "metadata") Object.assign(metadata, p)
if (e.kind === "payload") {
payloadRaw = String(p.blob ?? "")
format = p.format
}
}
let schemaDefinition: string | undefined
const conformsTo = this.graph.outgoing(nodeID, "conforms-to").at(-1)
if (conformsTo) {
// Walk supersession chain to latest schema version
let schemaNodeID = conformsTo.to_node
let supersedes = this.graph.outgoing(schemaNodeID, "supersedes")
while (supersedes.length > 0) {
schemaNodeID = supersedes[0].to_node
supersedes = this.graph.outgoing(schemaNodeID, "supersedes")
}
const payloadEdges = this.graph.outgoing(schemaNodeID, "payload")
const lastPayload = payloadEdges.at(-1)
if (lastDef(lastPayload)) {
const p2 = JSON.parse(lastPayload.payload!)
schemaDefinition = p2?.blob
}
}
return {
metadata,
payload: hydrate(payloadRaw, format),
format,
schemaDefinition,
}
}
/**
* Write content to any node. Format tells consumers how to hydrate;
* schemaAddress optionally links to a validating schema node.
*/
write(nodeID: string, opts: {
format?: string
metadata?: Record<string, unknown>
payload?: string
schemaAddress?: string
}): void {
if (opts.metadata) {
this.graph.edge(nodeID, nodeID, "metadata", { ...opts.metadata })
}
if (opts.payload !== undefined) {
this.graph.edge(nodeID, nodeID, "payload", { blob: opts.payload, format: opts.format })
}
if (opts.schemaAddress) {
const schemaNode = this.graph.getNode(opts.schemaAddress)
if (schemaNode) this.graph.edge(nodeID, schemaNode.id, "conforms-to", {})
}
}
}
export interface Step {
id: string
description: string
dependsOn: string[]
function lastDef(edge: import("./graph").Edge | undefined): boolean {
return edge !== undefined && edge.payload !== null && edge.payload !== ""
}
function hydrate(raw: string, format?: string): unknown {
if (!raw) return raw
switch (format) {
case "json":
case "schema.json":
try { return JSON.parse(raw) } catch { return raw }
default:
return raw
}
}
+32 -65
View File
@@ -3,18 +3,16 @@ 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.
* Every operation is a process node. You hold the node reference,
* pass it to the executor. No name lookup, no magic strings.
*/
interface GraphNode {
id: string
kind: string
address: string
}
export class Orchestrator {
private graph: Graph
@@ -22,72 +20,41 @@ export class Orchestrator {
this.graph = graph
}
/** Register any process definition. */
define(name: string, def: {
define(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 })
}
}): GraphNode {
const procNode = this.graph.ensureNode("process-definition", "process:" + crypto.randomUUID())
this.graph.edge(procNode.id, procNode.id, "definition", { ...def })
return { id: procNode.id, kind: "process-definition", address: procNode.address }
}
/**
* 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}`)
async run(processNode: GraphNode, targetNodeID: string): Promise<unknown> {
const defEdges = this.graph.outgoing(processNode.id, "definition")
const lastDef = defEdges.at(-1)
if (!lastDef || !lastDef.payload) throw new Error("process has no definition")
const def = JSON.parse(lastDef.payload)
// 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)
}
}
for (const readKind of def.reads ?? []) {
inputs[readKind] = this.graph.outgoing(targetNodeID, readKind).map(function(e) {
return { kind: e.kind, payload: e.payload ? JSON.parse(e.payload) : undefined }
})
}
const result = await this.executeDefinition(processName, inputs)
const result = await def.execute(inputs)
// Write result, superseding any prior result from the same process
this.writeResult(targetNodeID, processName, result)
// Supersede prior results from same process on same target
const priorKind = "result:" + processNode.id
const priors = this.graph.outgoing(targetNodeID).filter(function(e) {
return e.kind === priorKind
})
for (const prior of priors) {
this.graph.edge(targetNodeID, targetNodeID, "supersedes", { supersededBy: processNode.id })
}
this.graph.edge(targetNodeID, targetNodeID, priorKind, { data: 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,75 @@
export * as StepExecution from "./step-execution"
import { Graph } from "./graph"
/**
* A step execution is not just "did the thing."
*
* Every step records:
* - which knowledge nodes were consulted (used-knowledge edges)
* - which process definition was followed, or explicitly none
* - what lessons were learned during execution
*
* These are the edges that make LEARN possible: without provenance of
* what informed each step, you cannot refine anything.
*/
export interface StepInput {
stepNodeID: string
/** Knowledge nodes consulted during this step. */
knowledgeUsed: string[]
/** Process definition followed, or null if none matched. */
processUsed: string | null
/** What was learned doing this work. */
lesson?: string
/** The actual output. */
result: string
}
export function execute(graph: Graph, sessionID: string, input: StepInput): { resultEdgeID: string } {
const turn = graph.ensureNode("turn", `turn:${crypto.randomUUID()}`)
graph.edge(sessionID === turn.id ? turn.id : turn.id, turn.id, "self", {})
// Wire knowledge provenance
for (const knID of input.knowledgeUsed) {
graph.edge(turn.id, knID, "used-knowledge", {})
}
// Wire process provenance (or explicit absence)
if (input.processUsed) {
const procNode = graph.getNode(`process:${input.processUsed}`)
if (procNode) {
graph.edge(turn.id, procNode.id, "used-process", {})
}
} else {
graph.edge(turn.id, turn.id, "no-process-used", { explicit: true })
}
// Record the result
graph.edge(turn.id, turn.id, "result", { text: input.result })
return { resultEdgeID: turn.id }
}
/** Extract lessons from completed steps — this feeds the LEARN phase. */
export function collectLessons(graph: Graph, convNodeId: string): Array<{ lesson: string; hadProcess: boolean; knowledgeUsed: number }> {
const lessons: Array<{ lesson: string; hadProcess: boolean; knowledgeUsed: number }> = []
const turnEdges = graph.outgoing(convNodeId, "turn")
for (const e of turnEdges) {
const usedKnowledge = graph.outgoing(e.to_node, "used-knowledge")
const usedProcess = graph.outgoing(e.to_node, "used-process")
const noProcess = graph.outgoing(e.to_node, "no-process-used")
const results = graph.outgoing(e.to_node, "result")
if (results.length > 0 && !usedProcess.length && noProcess.length > 0) {
const payload = results[0]?.payload ? JSON.parse(results[0].payload).text : ""
lessons.push({ lesson: payload, hadProcess: false, knowledgeUsed: usedKnowledge.length })
} else if (results.length > 0 && usedProcess.length > 0) {
const payload = results[0]?.payload ? JSON.parse(results[0].payload).text : ""
lessons.push({ lesson: payload, hadProcess: true, knowledgeUsed: usedKnowledge.length })
}
}
return lessons
}
@@ -0,0 +1,98 @@
export * as ToolExecutor from "./tool-executor"
import { Graph } from "./graph"
/**
* Generic tool executor. Reads the tool node's edges, dispatches to the
* right primitive. Adding a new tool = adding a graph node with the
* right edges. No code changes.
*
* Primitives are fixed and small:
* bash - spawn shell command
* read - read file
* write - write file
* fetch - HTTP request
* search - ripgrep
*/
interface ToolSpec {
name: string
description: string
scope: string
parameters: unknown
primitive: string
config?: Record<string, unknown>
}
export function register(graph: Graph, opts: ToolSpec): void {
const addr = `tool:${opts.name}`
const node = graph.ensureNode("tool", addr)
graph.edge(node.id, node.id, "description", { text: opts.description })
graph.edge(node.id, node.id, "parameters", { schema: opts.parameters })
graph.edge(node.id, node.id, "scope", { value: opts.scope })
graph.edge(node.id, node.id, "implementation", { primitive: opts.primitive, config: opts.config ?? {} })
}
export function execute(
graph: Graph,
toolName: string,
input: Record<string, unknown>,
ctx: { cwd: string },
): Promise<string> {
const addr = `tool:${toolName}`
const node = graph.getNode(addr)
if (!node) return Promise.reject(new Error(`unknown tool: ${toolName}`))
const implEdges = graph.outgoing(node.id, "implementation")
const implEdge = implEdges.at(-1)
if (!implEdge || !implEdge.payload) return Promise.reject(new Error(`tool ${toolName} has no implementation`))
const impl = JSON.parse(implEdge.payload) as { primitive: string; config?: Record<string, unknown> }
const config = impl.config ?? {}
switch (impl.primitive) {
case "bash": {
const cmd = String(input.command ?? "")
const workDir = String(config.cwd ?? ctx.cwd)
const proc = Bun.spawn(["zsh", "-c", cmd], { cwd: workDir, stdout: "pipe", stderr: "pipe" })
return proc.exited.then(function(code) {
const out = new Response(proc.stdout).body ? "" : ""
return new Response(proc.stdout).text().then(function(stdout) {
const stderrText = new Response(proc.stderr).text()
return stdout + (code === 0 ? "" : "\n[exit " + code + "]")
})
})
}
case "read": {
const filePath = String(input.path ?? "")
const limit = typeof input.limit === "number" ? input.limit : undefined
return Bun.file(ctx.cwd + "/" + filePath)
.text()
.then(function(text) {
if (!limit) return text
return text.split("\n").slice(0, limit).join("\n")
})
.catch(function() { return "[file not found: " + filePath + "]" })
}
case "write": {
const filePath = String(input.path ?? "")
const content = String(input.content ?? "")
return Bun.write(ctx.cwd + "/" + filePath, content)
.then(function() { return "wrote " + filePath + " (" + content.length + " bytes)" })
}
case "fetch": {
const url = String(config.url ?? input.url ?? "")
return fetch(url).then(function(r) { return r.text() })
}
case "search": {
const pattern = String(input.pattern ?? "")
const searchDir = String(config.dir ?? ctx.cwd)
const proc = Bun.spawn(["rg", pattern, "--line-number", "--max-count", "20"], { cwd: searchDir, stdout: "pipe", stderr: "pipe" })
return proc.exited.then(function() {
return new Response(proc.stdout).text()
})
}
default:
return Promise.reject(new Error(`unknown primitive: ${impl.primitive}`))
}
}
+60
View File
@@ -0,0 +1,60 @@
export * as ToolNodes from "./tool-nodes"
import { Graph } from "./graph"
/**
* Tools are typed nodes in the graph.
* Definition edges carry description, parameters, scope, handler.
* No separate registry, no config file — query the graph.
*/
export function register(graph: Graph, opts: {
name: string
description: string
parameters: unknown
scope: string
scopeNodeID?: string
run: (input: any) => Promise<string>
}): { nodeID: string } {
const addr = `tool:${opts.name}`
const node = graph.ensureNode("tool", addr)
graph.edge(node.id, node.id, "description", { text: opts.description })
graph.edge(node.id, node.id, "parameters", { schema: opts.parameters })
graph.edge(node.id, node.id, "scope", { value: opts.scope })
graph.edge(node.id, node.id, "handler", { fn: opts.run })
if (opts.scopeNodeID) {
graph.edge(opts.scopeNodeID, node.id, "provides-tool", {})
}
return { nodeID: node.id }
}
export function lookup(graph: Graph, name: string): { nodeID: string; run: (input: any) => Promise<string>; scope: string } | undefined {
const node = graph.getNode(`tool:${name}`)
if (!node) return undefined
let handler: ((input: any) => Promise<string>) | undefined
let scope = ""
for (const e of graph.outgoing(node.id)) {
if (e.kind === "handler") {
const p = e.payload ? JSON.parse(e.payload) : undefined
if (p?.fn) handler = p.fn
}
if (e.kind === "scope") {
const p = e.payload ? JSON.parse(e.payload) : undefined
if (p?.value) scope = p.value
}
}
if (!handler) return undefined
return { nodeID: node.id, run: handler, scope }
}
export function allTools(graph: Graph): Array<{ name: string }> {
return graph.nodesByKind("tool").map((n) => ({
name: n.address.replace("tool:", ""),
}))
}
+81
View File
@@ -0,0 +1,81 @@
/**
* Registers the existing tool inventory as graph nodes.
* Each tool becomes a node with edges for description/parameters/scope/primitive.
* The old tools.ts stays untouched for reference — this is the graph-native version.
*/
import { Graph } from "./graph"
import { register } from "./tool-executor"
export function registerBuiltinTools(graph: Graph, cwd: string): void {
register(graph, {
name: "bash",
description: "Run a shell command in the working directory and return its output.",
parameters: {
type: "object",
properties: { command: { type: "string" } },
required: ["command"],
},
scope: "fs.execute",
primitive: "bash",
})
register(graph, {
name: "read",
description: "Read a text file relative to the working directory.",
parameters: {
type: "object",
properties: {
path: { type: "string", description: "File path relative to working directory" },
limit: { type: "number", description: "Max lines to read" },
},
required: ["path"],
},
scope: "fs.read",
primitive: "read",
})
register(graph, {
name: "write",
description: "Write content to a text file relative to the working directory.",
parameters: {
type: "object",
properties: {
path: { type: "string", description: "File path relative to working directory" },
content: { type: "string", description: "Content to write" },
},
required: ["path", "content"],
},
scope: "fs.write",
primitive: "write",
})
register(graph, {
name: "search",
description: "Search file contents using ripgrep. Returns matching lines with file paths and line numbers.",
parameters: {
type: "object",
properties: {
pattern: { type: "string", description: "Regex pattern to search for" },
dir: { type: "string", description: "Directory to search in (defaults to cwd)" },
},
required: ["pattern"],
},
scope: "fs.read",
primitive: "search",
})
register(graph, {
name: "fetch",
description: "Fetch content from a URL and return the response body as text.",
parameters: {
type: "object",
properties: {
url: { type: "string", description: "URL to fetch" },
},
required: ["url"],
},
scope: "net.fetch",
primitive: "fetch",
})
}