neuron/runs/rung1/kernel.ts

/**
 * RUNG 1 - SELF-HOSTING AGENT KERNEL
 *
 * There is exactly one agent. Every agent does the same thing every time:
 *
 *   ORCHESTRATE  decompose intent into a step DAG
 *   EXECUTE      run runnable steps; publish every transition
 *   LEARN        fold outcomes
 *   BUILD        emit new process/knowledge artifacts
 *   REFINE       supersede own procedure with what was learned
 *
 * Cancellation tokens: durable control topic per agent. Commands are
 * ledger entries - never missable, idempotent on replay.
 * Crash recovery: silence is the only failure signal. A fresh kernel
 * folds control + work topics and resumes at the first unfinished step.
 *
 * Isolated executable: bun runs/rung1/kernel.ts <universe.db>
 */

import { Seed } from "../../packages/core/src/seed"
import { EventBusSchema } from "../../packages/core/src/event-bus-schema"

const { Body, MessageHeader, Artifact } = EventBusSchema

// ---------- Plans ----------

export interface Step {
  readonly id: string
  readonly title: string
  readonly dependsOn: ReadonlyArray<string>
}

export interface Plan {
  readonly steps: ReadonlyArray<Step>
}

export const runnable = (plan: Plan, done: ReadonlySet<string>): Array<Step> =>
  plan.steps.filter(
    (s) => !done.has(s.id) && s.dependsOn.every((d) => done.has(d)),
  )

// ---------- Token control (durable, idempotent) ----------

export type Command = { type: "pause" | "resume" | "cancel"; author?: string }

const foldCommands = (commands: Array<Command>): "active" | "paused" | "cancelled" => {
  let state: "active" | "paused" | "cancelled" = "active"
  for (const c of commands) {
    if (c.type === "cancel") state = "cancelled"
    else if (state !== "cancelled") state = c.type === "pause" ? "paused" : "active"
  }
  return state
}

// ---------- Step executor signature ----------

export type StepExecutor = (
  step: Step,
  ctx: { readonly seed: Seed; readonly token: string },
) => Promise<Array<{ ref: string; hash?: string; kind?: string }>>

// ---------- The Loop ----------

export interface AgentSpec {
  readonly token: string
  /** Topic where this agent's work transitions land (durable). */
  readonly workTopic: string
  /** Decompose the current intent into a step DAG. */
  readonly orchestrate: (intent: unknown) => Promise<Plan>
  /** Execute one step. Physics lives here; everything else is protocol. */
  readonly executeStep: StepExecutor
  /** LEARN hook: receive each outcome as it lands. May return knowledge envelopes. */
  readonly learn?: (outcome: {
    readonly step: Step
    readonly ok: boolean
    readonly error?: unknown
    readonly artifacts: ReadonlyArray<{ ref: string; hash?: string; kind?: string }>
  }) => Promise<void>
}

const now = () => Date.now()

async function readTokenState(seed: Seed, token: string): Promise<"active" | "paused" | "cancelled"> {
  const control = `agent.${token}.control`
  const commands: Array<Command> = seed.replay(control).map((e) => JSON.parse(e.body.payloadJSON ?? "{}"))
  return foldCommands(commands)
}

async function completedSteps(seed: Seed, workTopic: string): Promise<Set<string>> {
  return new Set(
    seed.query.byType(workTopic, "plan.step.completed").map((e) => {
      try {
        return (JSON.parse(e.body.payloadJSON ?? "{}") as { stepId?: string }).stepId ?? ""
      } catch {
        return ""
      }
    }),
  )
}

function envelope(input: {
  topic: string
  parentSeq?: number
  author: string
  type: string
  payload?: Record<string, unknown>
  artifacts?: Array<{ ref: string; hash?: string; kind?: string }>
}) {
  return {
    topic: input.topic,
    ...(input.parentSeq === undefined ? {} : { parentSeq: input.parentSeq }),
    author: input.author,
    at: now(),
    headers: [] as ReturnType<typeof MessageHeader.make>[],
    body: Body.make({
      type: input.type,
      ...(input.payload ? { payloadJSON: JSON.stringify(input.payload) } : {}),
      artifacts: (input.artifacts ?? []).map((a) => Artifact.make(a)),
    }),
  }
}

/** Run one full OELBR revolution for an intent. Returns summary. */
export async function runLoop(
  seed: Seed,
  spec: AgentSpec,
  intent: unknown,
): Promise<{
  readonly status: "done" | "paused-out" | "cancelled"
  readonly doneSteps: ReadonlySet<string>
}> {
  const control = `agent.${spec.token}.control`
  const done = await completedSteps(seed, spec.workTopic)

  // ORCHESTRATE - decompose into a DAG (fresh instance reuses ledger's progress)
  let plan = await spec.orchestrate(intent)
  seed.publish({
    ...envelope({ topic: spec.workTopic, author: spec.token, type: "plan.declared",
      payload: { steps: plan.steps } }),
    // parent to latest if resuming mid-conversation
    ...(seed.replay(spec.workTopic).slice(-1)[0] !== undefined
      ? { parentSeq: seed.replay(spec.workTopic).slice(-1)[0]!.seq }
      : {}),
  })

  // EXECUTE - dependency order, honoring the token between steps
  for (;;) {
    const tokenState = await readTokenState(seed, spec.token)
    if (tokenState === "cancelled")
      return { status: "cancelled", doneSteps: done }
    if (tokenState === "paused") {
      await new Promise((r) => setTimeout(r, 50)) // paused: cheap subscription wait
      continue
    }
    const next = runnable(plan, done)[0]
    if (!next) break

    const started = seed.publish({
      ...envelope({ topic: spec.workTopic, author: spec.token, type: "plan.step.started",
        payload: { stepId: next.id, title: next.title } }),
      parentSeq: seed.replay(spec.workTopic).slice(-1)[0]?.seq,
    })

    let artifacts: Array<{ ref: string; hash?: string; kind?: string }> = []
    let failure: unknown
    try {
      artifacts = await spec.executeStep(next, { seed, token: spec.token })
    } catch (err) {
      failure = err
    }

    if (failure !== undefined) {
      seed.publish({
        ...envelope({ topic: spec.workTopic, author: spec.token, type: "plan.step.failed",
          payload: { stepId: next.id, error: String(failure) }, parentSeq: started.seq }),
      })
      if (spec.learn) await spec.learn({ step: next, ok: false, error: failure, artifacts: [] })
      return { status: "cancelled", doneSteps: done } // failed loop ends; lesson recorded
    }

    seed.publish({
      ...envelope({ topic: spec.workTopic, author: spec.token, type: "plan.step.completed",
        payload: { stepId: next.id }, artifacts, parentSeq: started.seq }),
    })
    done.add(next.id)
    if (spec.learn) await spec.learn({ step: next, ok: true, artifacts })
  }

  return { status: "done", doneSteps: done }
}

/** Issue commands to a token. Durable append - cannot be missed. */
export function command(seed: Seed, token: string, cmd: Command): void {
  const control = `agent.${token}.control`
  seed.topic({ id: control, durability: "durable", author: cmd.author ?? "system" })
  seed.publish({
    topic: control,
    author: cmd.author ?? "system",
    at: now(),
    headers: [],
    body: Body.make({ type: `token.${cmd.type}`, payloadJSON: JSON.stringify(cmd), artifacts: [] }),
  })
}