From 7af37786d80183e335df20d6401cf2da6a4fb347 Mon Sep 17 00:00:00 2001 From: "will.anderson" Date: Sat, 22 Aug 2026 22:49:17 -0500 Subject: [PATCH] feat(neuron): kernel with knowledge, backlog, orchestrator; wire transport proven live --- packages/neuron/src/kernel/aspect.ts | 113 +++++++++----- packages/neuron/src/kernel/kernel.ts | 118 ++++++++++++++ packages/neuron/src/llm/client.ts | 194 ++++++++++++++++++++++++ packages/neuron/src/session/sessions.ts | 110 +++++++------- 4 files changed, 442 insertions(+), 93 deletions(-) create mode 100644 packages/neuron/src/kernel/kernel.ts create mode 100644 packages/neuron/src/llm/client.ts diff --git a/packages/neuron/src/kernel/aspect.ts b/packages/neuron/src/kernel/aspect.ts index 651dc243cd..0ae9f5631b 100644 --- a/packages/neuron/src/kernel/aspect.ts +++ b/packages/neuron/src/kernel/aspect.ts @@ -3,72 +3,102 @@ export * as Aspect from "./aspect" import type { Graph } from "./graph" /** - * The AOP kit. Cross-cutting concerns are stages you decorate onto - * functions - never code scattered inside them. + * The AOP kit - one registry, applied seamlessly to any function. * - * const run = traced("runner.turn", metered("turns", guarded(claims, scope, work))) + * // composition site (once): + * Aspect.use("llm", Aspect.traced("llm"), Aspect.metered("llm")) + * Aspect.use("tool", Aspect.timed()) * - * Every stage records what it did as edges in the graph ledger: - * kind "trace" - a span with duration - * kind "metric" - a named observation with value - * The wire remembers; nothing is scraped after the fact. + * // anywhere, the function opts into its category once: + * export const complete = Aspect.wrap("llm", async (input) => {...}) + * + * Change instrumentation = change the composition list. Zero call sites + * are touched. That is the entire point. */ -export interface Claims { - principal: string - authed: boolean - scopes: string[] +type AsyncFn = (...args: A) => Promise + +/** A stage observes/wraps a call. It may time, log, deny, or pass. */ +export type Stage = (meta: { name: string; category: string }) => ( + fn: AsyncFn, +) => AsyncFn + +interface Entry { + category: string + stage: Stage } -export type AsyncFn = (...args: A) => Promise +const registry: Entry[] = [] -function record(graph: Graph | undefined, kind: string, payload: Record) { - if (!graph) return - graph.ensureNode("telemetry", `telemetry:${process.pid}`) - graph.edge(`telemetry:${process.pid}`, `telemetry:${process.pid}`, kind, { ...payload, at: Date.now() }) +/** Declare which aspects apply to a category. Call once at composition. */ +export function use(category: string, ...stages: Stage[]): void { + for (const stage of stages) registry.push({ category, stage }) } -/** Trace: time the call and record a span edge. */ -export function traced( - graph: Graph | undefined, +/** Wrap any function so its category's aspects apply to every invocation. */ +export function wrap( + category: string, name: string, fn: AsyncFn, ): AsyncFn { - return async (...args: A) => { + const stages = registry.filter((e) => e.category === category).map((e) => e.stage) + return async (...args: A): Promise => { + let chain: AsyncFn = fn + // Applied outermost-last so the first listed stage is outermost. + for (let i = stages.length - 1; i >= 0; i--) { + const inner = chain + chain = stages[i]({ name, category })(inner) + } + return chain(...args) + } +} + +// ---------- built-in stages ---------- + +/** Record duration + outcome of every call into the graph ledger. */ +export function traced(graph?: Graph): Stage { + return () => (fn) => async (...args: any[]) => { const start = Date.now() try { const result = await fn(...args) - record(graph, "trace", { name, ms: Date.now() - start, ok: true }) + if (graph) { + const t = graph.ensureNode("telemetry", `telemetry:${process.pid}`) + graph.edge(t.id, t.id, "trace", { fn: fn.name || "anonymous", ms: Date.now() - start, ok: true }) + } return result } catch (error) { - record(graph, "trace", { name, ms: Date.now() - start, ok: false, error: String(error).slice(0, 200) }) + if (graph) { + const t = graph.ensureNode("telemetry", `telemetry:${process.pid}`) + graph.edge(t.id, t.id, "trace", { + fn: fn.name || "anonymous", + ms: Date.now() - start, + ok: false, + error: String(error).slice(0, 200), + }) + } throw error } } } -/** Meter: count invocations of a named metric. */ -export function metered(graph: Graph | undefined, name: string, fn: AsyncFn): AsyncFn { - return async (...args: A) => { - record(graph, "metric", { name, at: Date.now() }) +/** Count invocations of a named metric into the ledger. */ +export function metered(graph?: Graph): Stage { + return (meta) => (fn) => async (...args: any[]) => { + if (graph) { + const t = graph.ensureNode("telemetry", `telemetry:${process.pid}`) + graph.edge(t.id, t.id, "metric", { name: meta.name, at: Date.now() }) + } return fn(...args) } } -export interface Policy { - claims: Claims - /** Claim requirements, e.g. ["authed:true", "scope:fs.write"]. */ - require?: string[] -} - -/** Guard: one verdict from the principal's claims, stamped on failure too. */ -export function guarded( +/** Deny calls whose claims don't satisfy the requirement list. */ +export function guarded( graph: Graph | undefined, - policy: () => Policy, - fn: AsyncFn, -): AsyncFn { - return async (...args: A) => { - const { claims, require = [] } = policy() + policy: () => { claims: { principal: string; authed: boolean; scopes: string[] }; require: string[] }, +): Stage { + return () => (fn) => async (...args: any[]) => { + const { claims, require } = policy() for (const requirement of require) { const [key, expected] = requirement.split(":") const actual = @@ -78,7 +108,10 @@ export function guarded( ? String(claims.authed) : String((claims as any)[key] ?? "") if (actual !== (expected ?? "true")) { - record(graph, "denied", { requirement, principal: claims.principal }) + if (graph) { + const t = graph.ensureNode("telemetry", `telemetry:${process.pid}`) + graph.edge(t.id, t.id, "denied", { requirement, principal: claims.principal }) + } throw new Error(`guard denied: ${requirement}`) } } diff --git a/packages/neuron/src/kernel/kernel.ts b/packages/neuron/src/kernel/kernel.ts new file mode 100644 index 0000000000..6359815f37 --- /dev/null +++ b/packages/neuron/src/kernel/kernel.ts @@ -0,0 +1,118 @@ + + +import { Graph } from "./graph" +import type { Node } from "./graph" + +/** + * Domain operations on 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. + */ + +export class Kernel { + private graph: Graph + + constructor(graph: Graph) { + 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 + } + + 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. + */ + async orchestrate(intent: string, executors: Map Promise>): Promise> { + const steps = this.decompose(intent) + const done = new Set() + const results: Array<{ step: string; result: string }> = [] + + 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") + } + } + return results + } + + 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"] }, + ] + } +} diff --git a/packages/neuron/src/llm/client.ts b/packages/neuron/src/llm/client.ts new file mode 100644 index 0000000000..49f4c442d4 --- /dev/null +++ b/packages/neuron/src/llm/client.ts @@ -0,0 +1,194 @@ +export * as LLMClient from "./client" + +import type { Dialect } from "../provider/dialects/types" +import type { Graph } from "../kernel/graph" + +/** + * The LLM call is a PIPELINE. You compose stages - each stage observes or + * enriches the call context - and the last stage does the wire work. + * + * const llm = createLLMPipeline(dialect, [traced(), metered()]) + * const result = await llm({ system, messages, tools }) + * + * Every stage sees the same context: request, response, tokens, timing. + * Telemetry stages publish spans/metrics onto the bus; nothing scrapes. + */ + +export interface Usage { + input: number + output: number +} + +export interface ToolCall { + id: string + name: string + args: string +} + +export interface CompletionResult { + text: string + toolCalls: ToolCall[] + finish: string | null + usage: Usage + ms: number +} + +/** A stage wraps the continuation: observe, enrich, short-circuit, or pass. */ +export type Stage = (ctx: CallContext, next: () => Promise) => Promise + +export interface CallContext { + input: { + baseURL: string + authHeaders: Record + model: string + system: string[] + messages: unknown[] + tools?: unknown[] + } + /** Filled by the wire stage. */ + text: string + toolCalls: ToolCall[] + finish: string | null + usage: Usage + ms: number + ok: boolean + error?: string +} + +type WireDialect = Pick + +async function wireStage(ctx: CallContext, dialect: WireDialect): Promise { + const start = Date.now() + const req = dialect.request({ + baseURL: ctx.input.baseURL, + apiKey: "", + authHeaders: ctx.input.authHeaders, + model: ctx.input.model, + system: ctx.input.system, + messages: ctx.input.messages, + tools: ctx.input.tools, + }) + + const res = await fetch(req.url, { + method: "POST", + headers: req.headers, + body: JSON.stringify(req.body), + }) + if (!res.ok || !res.body) { + const detail = await res.text().catch(() => "") + throw new Error(`provider HTTP ${res.status}: ${detail.slice(0, 400)}`) + } + + const parser = dialect.parser() + const reader = res.body.getReader() + const decoder = new TextDecoder() + let carry = "" + + outer: while (true) { + const { done, value } = await reader.read() + if (done) break + carry += decoder.decode(value, { stream: true }) + let nl: number + while ((nl = carry.indexOf("\n")) !== -1) { + const line = carry.slice(0, nl).replace(/\r$/, "") + carry = carry.slice(nl + 1) + if (!line.startsWith("data:")) continue + const payload = line.slice(5).trim() + if (!payload || payload === "[DONE]") continue + + try { + const raw = JSON.parse(payload) + const u = raw.usage ?? raw.response?.usage + if (u) { + ctx.usage.input = + Number(u.input_tokens ?? u.prompt_tokens ?? 0) || ctx.usage.input + ctx.usage.output = + Number(u.output_tokens ?? u.completion_tokens ?? 0) || ctx.usage.output + } + } catch {} + + for (const event of parser.parse(payload)) { + switch (event.type) { + case "text-delta": + ctx.text += event.text + break + case "tool-call": { + const existing = ctx.toolCalls.find((c) => c.id === event.id) + if (existing) existing.args += event.arguments + else ctx.toolCalls.push({ id: event.id, name: event.name, args: event.arguments }) + break + } + case "finish": + ctx.finish = event.reason + break outer + case "error": + throw new Error(`provider error: ${event.message}`) + } + } + } + } + ctx.ms = Date.now() - start +} + +/** + * Compose the pipeline. Order given at composition is order of execution; + * every stage is named code in this file - nothing hidden. + */ +export function createLLMPipeline( + dialect: WireDialect, + stages: Array = [], +): (input: { + baseURL: string + authHeaders: Record + model: string + system: string[] + messages: unknown[] + tools?: unknown[] +}) => Promise { + return async (input) => { + const start = Date.now() + const ctx: CallContext = { + input, + text: "", + toolCalls: [], + finish: null, + usage: { input: 0, output: 0 }, + ms: 0, + ok: true, + } + + // Onion: stages run in order, each may pass to the rest. + const runStage = async (index: number): Promise => { + if (index >= stages.length) { + await wireStage(ctx, dialect) + return + } + await stages[index](ctx, () => runStage(index + 1)) + } + + let failure: unknown + try { + await runStage(0) + } catch (error) { + failure = error + ctx.ok = false + ctx.error = String(error).slice(0, 300) + } + ctx.ms = Date.now() - start + + if (!ctx.ok && failure !== undefined) { + // Errors are data too: rethrow with full context attached. + const err = new Error(`LLM call failed after ${ctx.ms}ms: ${ctx.error}`) + ;(err as any).usage = ctx.usage + throw err + } + + return { + text: ctx.text, + toolCalls: ctx.toolCalls, + finish: ctx.finish, + usage: ctx.usage, + ms: ctx.ms, + } + } +} diff --git a/packages/neuron/src/session/sessions.ts b/packages/neuron/src/session/sessions.ts index 1714749d6e..42236aea6b 100644 --- a/packages/neuron/src/session/sessions.ts +++ b/packages/neuron/src/session/sessions.ts @@ -5,6 +5,8 @@ import * as Conversation from "../conversation" import * as Tools from "../tools" import * as Auth from "../auth" import * as Context from "./context" +import { createLLMPipeline } from "../llm/client" +import * as Aspect from "../kernel/aspect" /** * V2 Session Core - admission is durable and separate from execution. @@ -200,63 +202,65 @@ export function open(input: { const calls = new Map() let finished = false - const req = provider.dialect.request({ - baseURL: provider.baseURL, - apiKey: "", - authHeaders: provider.authHeaders, - model: provider.model, - system, - messages: wireMessages, - tools: provider.dialect.toWireTools(tools), - }) - - const res = await fetch(req.url, { method: "POST", headers: req.headers, body: JSON.stringify(req.body) }) - if (!res.ok || !res.body) { - const detail = await res.text().catch(() => "") - if (process.env.NEURON_DEBUG) - console.error("[wire-dump]", JSON.stringify({ url: req.url, body: req.body }).slice(0, 2000)) - throw new Error(`provider HTTP ${res.status}: ${detail.slice(0, 400)}`) - } - - const parser = provider.dialect.parser() - const reader = res.body.getReader() - const decoder = new TextDecoder() - let carry = "" - outer: while (true) { - const { done, value } = await reader.read() - if (done) break - carry += decoder.decode(value, { stream: true }) - let nl: number - while ((nl = carry.indexOf("\n")) !== -1) { - const line = carry.slice(0, nl).replace(/\r$/, "") - carry = carry.slice(nl + 1) - if (!line.startsWith("data:")) continue - const payload = line.slice(5).trim() - if (!payload || payload === "[DONE]") continue - for (const event of parser.parse(payload)) { - switch (event.type) { - case "text-delta": - text += event.text - input.onDelta?.(event.text) - break - case "reasoning-delta": - break - case "tool-call": { - const existing = calls.get(event.id) - if (existing) existing.args += event.arguments - else calls.set(event.id, { id: event.id, name: event.name, args: event.arguments }) - break + await Aspect.wrap("llm", "llm.turn", async () => { + const req = provider.dialect.request({ + baseURL: provider.baseURL, + apiKey: "", + authHeaders: provider.authHeaders, + model: provider.model, + system, + messages: wireMessages, + tools: provider.dialect.toWireTools(tools), + }) + const res = await fetch(req.url, { + method: "POST", + headers: req.headers, + body: JSON.stringify(req.body), + }) + if (!res.ok || !res.body) { + const detail = await res.text().catch(() => "") + throw new Error(`provider HTTP ${res.status}: ${detail.slice(0, 400)}`) + } + const parser = provider.dialect.parser() + const reader = res.body.getReader() + const decoder = new TextDecoder() + let carry = "" + let finished = false + outer: while (true) { + const { done, value } = await reader.read() + if (done) break + carry += decoder.decode(value, { stream: true }) + let nl: number + while ((nl = carry.indexOf("\n")) !== -1) { + const line = carry.slice(0, nl).replace(/\r$/, "") + carry = carry.slice(nl + 1) + if (!line.startsWith("data:")) continue + const payload = line.slice(5).trim() + if (!payload || payload === "[DONE]") continue + for (const event of parser.parse(payload)) { + switch (event.type) { + case "text-delta": + text += event.text + input.onDelta?.(event.text) + break + case "tool-call": { + const existing = calls.get(event.id) + if (existing) existing.args += event.arguments + else calls.set(event.id, { id: event.id, name: event.name, args: event.arguments }) + break + } + case "finish": + finished = true + break outer + case "error": + throw new Error(`provider error: ${event.message}`) } - case "finish": - finished = true - break outer - case "error": - throw new Error(`provider error: ${event.message}`) } } } - } - + if (finished) return "stop" + return "continue" + })() if (calls.size === 0 || finished) { Conversation.addTurn(graph, conv.nodeId, "neuron", text) return