@agentick/kernel

Low-level execution primitives for Agentick. Provides procedures, async context management, schema validation, logging, telemetry, and event streaming.

Note: Most applications should use @agentick/core instead. This package is the foundation that core builds upon.

Installation

pnpm add @agentick/kernel

Core Primitives

Procedures

Procedures wrap any async function, generator, or async iterable with middleware, execution tracking, and streaming:

import { createProcedure } from "@agentick/kernel";
import { z } from "zod";

// Async function with schema validation
const fetchUser = createProcedure(
  {
    name: "fetchUser",
    schema: z.object({ id: z.string() }),
  },
  async ({ id }) => {
    return await db.users.find(id);
  },
);

// Execute
const user = await fetchUser({ id: "123" });

// With middleware
const withLogging = fetchUser.use(async (args, ctx, next) => {
  console.log("Fetching user:", args.id);
  const result = await next(args);
  console.log("Found:", result);
  return result;
});

Async Generators

Procedures wrapping generators get automatic context preservation, stream:chunk events, abort handling, and iterator cleanup:

const tokenStream = createProcedure(
  { name: "tokens", handleFactory: false },
  async function* (prompt: string) {
    const response = await fetchSSE(prompt);
    for await (const chunk of response) {
      yield chunk.text;
    }
  },
);

const iter = await tokenStream("tell me a joke");
for await (const token of iter) {
  process.stdout.write(token);
}

Any function returning an AsyncIterable gets the same treatment — it doesn't have to be async function*.

ExecutionHandle

Procedures are callable directly. You can also use .exec():

// Direct call (preferred)
const handle = await fetchUser({ id: "123" });

// Or explicit .exec()
const handle = await fetchUser.exec({ id: "123" });

// Check status
console.log(handle.status); // "pending" | "running" | "completed" | "error"

// Abort if needed
handle.abort();

// Wait for result
const user = await handle.result;

Composition

import { pipe, compose } from "@agentick/kernel";

// Chain procedures left-to-right
const pipeline = pipe(validate, transform, save);

// Or right-to-left (functional style)
const composed = compose(save, transform, validate);

Context (AsyncLocalStorage)

Request-scoped state that flows through async operations automatically:

import { Context } from "@agentick/kernel";

// Create and run within context
Context.run({ user: { id: "123" }, metadata: { traceId: "abc" } }, async () => {
  const ctx = Context.get();
  console.log(ctx.user?.id); // "123"

  // Context propagates through async calls
  await someAsyncOperation(); // Still has access to context
});

// Fork for parallel execution (safe isolation)
await Context.fork({ metadata: { branch: "A" } }, async () => {
  // Child context with overrides
});

Global Event Subscribers

// Subscribe to all context events (useful for DevTools/telemetry)
const unsubscribe = Context.subscribeGlobal((event) => {
  console.log(event.type, event.payload);
});

// Emit events from anywhere
Context.emit("custom:event", { data: "value" });

Schema Validation

Unified handling for Zod 3, Zod 4, and Standard Schema:

import { detectSchemaType, toJSONSchema, validateSchema, parseSchema } from "@agentick/kernel";

// Detect schema type
const type = detectSchemaType(schema); // "zod3" | "zod4" | "standard-schema" | "json-schema"

// Convert to JSON Schema
const jsonSchema = toJSONSchema(myZodSchema);

// Validate (returns result object)
const result = validateSchema(schema, data);
if (result.success) {
  console.log(result.data);
} else {
  console.log(result.issues);
}

// Parse (throws on failure)
const data = parseSchema(schema, input);

Logging

Structured logging with automatic context injection:

import { Logger } from "@agentick/kernel";

// Configure globally
Logger.configure({
  level: "info",
  transport: "pretty", // or "json"
});

// Get context-aware logger
const log = Logger.get();
log.info("Processing request", { userId: "123" });

// Scoped logger
const dbLog = Logger.for("database");
dbLog.debug("Query executed", { query, duration });

Telemetry

Spans and metrics for observability:

import { Telemetry } from "@agentick/kernel";

// Start a trace
const trace = Telemetry.startTrace("handle-request");

// Create spans
const span = Telemetry.startSpan("fetch-user");
span.setAttribute("userId", "123");
try {
  // ... work
} catch (error) {
  span.recordError(error);
} finally {
  span.end();
}

// Metrics
const requestCounter = Telemetry.getCounter("requests_total");
requestCounter.add(1, { method: "POST" });

const latencyHistogram = Telemetry.getHistogram("request_duration_ms");
latencyHistogram.record(42, { endpoint: "/api/users" });

Channels

Bidirectional communication for real-time updates:

import { Channel } from "@agentick/kernel";

const channel = new Channel();

// Subscribe
channel.on("message", (payload) => {
  console.log("Received:", payload);
});

// Publish
channel.emit("message", { text: "Hello" });

// Request/response pattern
const response = await channel.request("getUser", { id: "123" }, 5000);

// Broadcast to all
channel.broadcast("notification", { message: "System update" });

EventBuffer

Type-safe event streaming with replay:

import { EventBuffer } from "@agentick/kernel";

type MyEvent =
  | { type: "start"; id: string }
  | { type: "progress"; percent: number }
  | { type: "complete"; result: unknown };

const buffer = new EventBuffer<MyEvent>();

// Subscribe with type narrowing
buffer.on("progress", (event) => {
  console.log(`${event.percent}% complete`);
});

// Late subscribers can replay
buffer.onReplay("start", (event) => {
  console.log("Started:", event.id);
});

// Async iteration
for await (const event of buffer) {
  console.log(event.type);
}

// Push events
buffer.push({ type: "start", id: "123" });
buffer.push({ type: "progress", percent: 50 });
buffer.push({ type: "complete", result: { success: true } });

// Close when done
buffer.close();

Guards

Gate procedure execution with access control checks:

import { createGuard, GuardError } from "@agentick/kernel";

// Simple predicate — deny throws GuardError
const adminOnly = createGuard(
  (envelope) => envelope.context.user?.roles?.includes("admin") ?? false,
);

// With config — custom reason and guard type
const roleGuard = createGuard(
  {
    name: "role-guard",
    guardType: "role",
    reason: (envelope) => `User ${envelope.context.user?.id} lacks required role`,
  },
  (envelope) => envelope.context.user?.roles?.includes("admin") ?? false,
);

// Throw GuardError directly for full control
const customGuard = createGuard({ name: "acl-guard" }, (envelope) => {
  if (!hasPermission(envelope.context.user)) {
    throw GuardError.role(["admin", "moderator"]);
  }
  return true;
});

// Apply to any procedure via .use()
const secured = fetchUser.use(adminOnly);

Guards are middleware — they compose with .use() like any other middleware but are purpose-built for allow/deny decisions.

GuardError

import { GuardError, isGuardError } from "@agentick/kernel";

// Factories
GuardError.role(["admin"]); // "Requires one of roles [admin]"
GuardError.denied("Custom reason"); // Custom denial

// Type guard
if (isGuardError(error)) {
  error.code; // "GUARD_DENIED"
  error.guardType; // "role", "custom", etc.
  error.details; // { roles: [...], guard: "name", ... }
}

Key Patterns

Middleware Pipelines

import { createPipeline } from "@agentick/kernel";

const authPipeline = createPipeline([
  async (args, ctx, next) => {
    if (!ctx.user) throw new Error("Unauthorized");
    return next(args);
  },
  async (args, ctx, next) => {
    ctx.metadata.startTime = Date.now();
    return next(args);
  },
]);

// Apply to any procedure
const securedFetch = fetchUser.use(authPipeline);

Immutable Composition

All procedure methods return new instances:

const base = createProcedure({ name: "base" }, handler);
const withTimeout = base.withTimeout(5000);
const withContext = withTimeout.withContext({ tenant: "acme" });

// Original unchanged
console.log(base === withTimeout); // false

API Reference

Procedures

| Export | Description | | ----------------------------------- | ------------------------------------------ | | createProcedure(options, handler) | Create a procedure (function or generator) | | generatorProcedure(options, fn) | Create a generator procedure with this | | createHook(options, handler) | Create a hook procedure | | pipe(...procedures) | Chain left-to-right | | compose(...procedures) | Chain right-to-left | | createPipeline(middleware) | Bundle middleware |

Guards

| Export | Description | | --------------------------- | --------------------------- | | createGuard(fn) | Create guard from predicate | | createGuard(config, fn) | Create guard with config | | GuardError | Access denied error class | | GuardError.role(roles) | Role-based denial factory | | GuardError.denied(reason) | Custom denial factory | | isGuardError(error) | Type guard for GuardError |

Context

| Export | Description | | ------------------------------------ | --------------------------- | | Context.create(overrides?) | Create root context | | Context.run(context, fn) | Run within context | | Context.fork(overrides, fn) | Fork for parallel execution | | Context.get() / Context.tryGet() | Access current context | | Context.emit(type, payload) | Emit event | | Context.subscribeGlobal(handler) | Subscribe to all events |

Schema

| Export | Description | | ------------------------------- | --------------------------- | | detectSchemaType(schema) | Identify schema type | | toJSONSchema(schema) | Convert to JSON Schema | | validateSchema(schema, value) | Validate with result object | | parseSchema(schema, value) | Parse or throw |

Logging & Telemetry

| Export | Description | | ------------------------------ | ------------------------ | | Logger.configure(options) | Configure logging | | Logger.get() | Get context-aware logger | | Logger.for(name) | Get scoped logger | | Telemetry.startSpan(name) | Create span | | Telemetry.getCounter(name) | Create counter metric | | Telemetry.getHistogram(name) | Create histogram metric |

Streaming

| Export | Description | | ----------------- | ------------------------------------- | | Channel | Pub/sub with request/response | | EventBuffer<T> | Type-safe event streaming with replay | | mapStream(s,fn) | Transform items in an async stream | | tapStream(s,fn) | Side effects without modifying stream | | mergeStreams(s) | Merge multiple streams (race) | | isAsyncIterable | Type guard for async iterables |

License

MIT