Architecture

kayak-lab is built as a three-layer architecture: Core, Capabilities, and Projections.

Core Layer

The foundation of the platform. Handles event storage, session lifecycle, and agent execution.

EventStream

Immutable, append-only sequence of events. Events are strictly ordered by sequence number within a session. Different sessions are completely isolated.

const stream = new EventStream();

// Append an event
const event = stream.append({
  session_id: "abc-123",
  event_type: "session.created",
  payload: { state: "active" },
  metadata: { source: "test" },
});

// Read events for a session
const events = stream.getEvents("abc-123");

SessionManager

Manages session lifecycle with a state machine. All transitions emit typed events.

const manager = new SessionManager(eventStream);

// Create a session
const session = await manager.createSession({ description: "My task" });

// Pause and resume
manager.pauseSession(session.id);
manager.resumeSession(session.id);

// Complete
manager.completeSession(session.id);

AgentRuntime

The agent execution loop: input → model → tool cycle. Manages context windows and tool invocations.

const runtime = new AgentRuntime(eventStream, sessionManager, modelManager, toolRegistry);

// Or with the new structured tool calling protocol
const runtime = new AgentRuntime(eventStream, sessionManager, modelManager, legacyToolRegistry, newToolRegistry);

// Start and process input
await runtime.start();
const response = await runtime.processInput("Run ls -la");

Dual-protocol dispatch: AgentRuntime checks newToolRegistry first for structured tools (JSON Schema validated), then falls back to the legacy toolRegistry. This enables incremental migration from legacy tools to the structured protocol.

Tool Calling Module

Structured tool execution protocol with JSON Schema validation, tool registry, authoring, and self-improvement.

The module lives in src/tools/ and provides:

Component	File	Purpose
ToolDefinition	tool-definition.ts	JSON Schema parameter validation for tool invocations
ToolCallingEngine	calling-engine.ts	Executes tools with structured input/output and timeout handling
ToolRegistry	registry.ts	Enable/disable lifecycle, discovery, and lookup of tools
ToolAuthoring	authoring.ts	Proposal/review/accept/reject flow for creating new tools
ToolSelfImprovement	self-improvement.ts	Usage tracking, suggestion generation, and auto-improvement
Types	types.ts	Shared interfaces (ToolHandlerContext, ToolDefinition, etc.)

import { ToolCallingEngine, ToolRegistry } from "./src/tools/mod.ts";

// Register tools with definitions
const registry = new ToolRegistry();
registry.enable("shell", { description: "Execute shell commands", parameters: { /* JSON Schema */ } });

// Execute a tool call
const engine = new ToolCallingEngine(registry);
const result = await engine.invoke({
  tool_name: "shell",
  parameters: { command: "ls -la" },
  tool_call_id: "call-123",
});

Tool authoring flow: Propose → Review → Accept/Reject → Register. Tools go through a review lifecycle before being added to the registry. Self-improvement tracks usage patterns and suggests optimizations.

MCP Module

Model Context Protocol (MCP) integration for connecting to external MCP servers and exposing harness capabilities as an MCP server. The module provides client, server, registry, and search components with a transport abstraction layer.

Component	File	Purpose
MCPClient	client.ts	Connects to external MCP servers, discovers tools, invokes them
MCPServer	server.ts	Exposes harness tools (exposable only) to external MCP clients
MCPRegistry	registry.ts	Manages MCP tools from external servers with state control
MCPSearch	search.ts	Searches MCP tools by name, capability, or category
Transport	transport.ts	Stdio, HTTP, WebSocket transports with factory function
Events	events.ts	MCP event types and type guards
Event Wiring	event-emitter.ts	Wires MCP events to the event stream

import { MCPClient, MCPServer, MCPRegistry, MCPSearch } from "./src/mcp/mod.ts";

// Connect to an external MCP server
const client = new MCPClient({
  name: "my-mcp-server",
  transport: { type: "stdio", command: "mcp-server", args: ["--port", "3000"] },
});
await client.connect();
const tools = await client.discover();

// Register discovered tools
const registry = new MCPRegistry();
for (const tool of tools) {
  registry.register({ tool, serverName: "my-mcp-server", enabled: true, capabilities: [] });
}

// Search for tools
const search = new MCPSearch(registry, new Map([["my-mcp-server", client]]));
const results = search.search({ name: "file" });

// Expose harness tools as MCP server
const server = new MCPServer({
  transport: { type: "http", url: "http://localhost:8080" },
  toolRegistry: {
    getExposableTools: () => toolRegistry.list().filter((t) => t.exposable),
    getTool: (name) => toolRegistry.get(name),
    invokeTool: (name, params) => toolRegistry.invoke(name, params),
  },
});
await server.start();

Transport abstraction: The createTransport factory creates the appropriate transport based on configuration. Each transport implements IMCPTransport with connect, disconnect, send, and notify methods.

Event integration: MCP operations emit events to the event stream via wireClientEvents, wireServerEvents, wireRegistryEvents, and wireSearchEvents helpers. Events include mcp.connected, mcp.disconnected, mcp.tool.invocation, mcp.tool.result, mcp.server.started, mcp.server.stopped, and mcp.error.

Memory

Persistent memory subsystem for agent learning and context retention. Stores and retrieves memories across sessions, enabling agents to accumulate knowledge over time.

Memory types:

Type	Purpose	Retention
Episodic	Event logs — what happened in a session	Short-term, event-sourced
Semantic	Learned facts — knowledge acquired over time	Long-term, persistent
Procedural	Learned patterns — how to do things	Long-term, persistent
Working	Active context — current session focus	Session-scoped, volatile

Key modules:

Component	File	Purpose
MemoryProvider	provider.ts	Abstraction layer over storage backends
MemoryStorage	storage.ts	Backend storage interface
MemoryRetrieval	retrieval.ts	Query and ranking of stored memories
MemorySearch	search.ts	Full-text search across memory stores
SharedMemory	shared.ts	Sub-agent context snapshots
MemoryConfig	config.ts	Memory configuration and backend selection
MemoryUpdate	update.ts	Atomic state transitions with event sourcing
MemoryEmitter	emitter.ts	Memory event emission for observability

import { MemoryProvider } from "./src/memory/provider.ts";

const memory = new MemoryProvider(config);
await memory.initialize();

// Store episodic memory
await memory.store({ type: "episodic", content: "User prefers dark mode" });

// Retrieve relevant memories
const results = await memory.search({ query: "user preferences", type: "semantic" });

Provider abstraction: The MemoryProvider decouples storage logic from consumers. Swap between in-memory, file-based, or database-backed storage without changing application code. SharedMemory enables sub-agents to share context snapshots without coupling to parent state.

EventStore

In-memory event persistence with snapshot support. Stores events per session with range queries and replay capabilities. Can be used as a fast/ephemeral store, or swapped with the persistent store for durability.

import { EventStore } from "./src/store/event-store.ts";

const store = new EventStore();

// Store an event
store.store(event);

// Read events
const events = store.getEvents("session-1");
const recent = store.getEventsInRange("session-1", 5, 10);

// Snapshots for fast replay
const snapshot = store.createSnapshot("session-1", { lastEventId: "abc" });

PersistentEventStore

File-based event persistence with JSONL append-only logs, snapshot persistence, and startup recovery. Durably writes events to disk and reconstructs in-memory state on initialization, enabling sessions to survive process restarts and crashes.

import { PersistentEventStore } from "./src/store/persistence.ts";

// Default: uses ./data/events/ directory
const store = new PersistentEventStore({ dataDir: "./data/events" });

// Custom directory
const store = new PersistentEventStore({ dataDir: "/var/lib/kayak/events" });

// Custom backend (e.g., SQLite in the future)
const store = new PersistentEventStore({
  dataDir: "./data/events",
  backend: new SQLitePersistenceBackend("./data/kayak.db"),
});

// Store events — written synchronously to JSONL on disk
store.store(event);

// Read events — served from in-memory cache (rebuilt from disk on startup)
const events = store.getEvents("session-1");
const recent = store.getEventsInRange("session-1", 5, 10);

// Snapshots — persisted to disk as JSON files
const snapshot = store.createSnapshot("session-1", { lastEventId: "abc" });

// Explicit flush (no-op for synchronous writes, available for future buffered backends)
store.flush();

File layout:

File	Purpose
<session_id>.jsonl	Append-only event log (one JSON object per line)
<session_id>.snapshot.json	Latest snapshot for fast recovery

Recovery behavior:

• On startup, PersistentEventStore scans the data directory, loads snapshots, and replays events after the snapshot point.
• Corrupted lines are logged and skipped; valid events continue loading.
• Empty or missing data directory → clean start with no sessions.

Pluggable backends:

The IPersistenceBackend interface allows swapping storage engines without changing callers:

interface IPersistenceBackend {
  write(sessionId: string, line: string): void;
  readLines(sessionId: string): string[];
  writeSnapshot(sessionId: string, data: Snapshot): void;
  readSnapshot(sessionId: string): Snapshot | undefined;
  listSessions(): string[];
  exists(sessionId: string): boolean;
}

The default FilePersistenceBackend uses synchronous Deno file I/O for guaranteed durability per write. Implement this interface for SQLite, PostgreSQL, or other backends.

Schema Registry

Event schema versioning and migration. Registers schema versions per event type and migrates events on read.

const registry = new SchemaRegistry();

// Register schema with migration
registry.register("session.created", 2, schemaV2, (event) => migrateV1toV2(event));

// EventStore uses registry for automatic migration
const store = new EventStore(undefined, registry);

Health System

Component health reporting with parallel checks and Kubernetes-compatible endpoints.

import { HealthRegistry, createHealthHandler } from "./src/core/health.ts";

const registry = new HealthRegistry();
registry.register("event-store", () => checkEventStore(store));
registry.register("capabilities", () => checkCapabilities(registry));

// Run all checks in parallel (1s timeout each)
const result = await registry.check();
// result.status: "healthy" | "degraded" | "unhealthy"

// HTTP endpoints
const handler = createHealthHandler(registry);
// GET /health  → full status (200/503)
// GET /ready   → readiness (200/503)
// GET /alive   → liveness (200)

Configuration Management

YAML-based config loading with env var overrides and secret masking.

import { loadConfig, validateConfig, maskSecrets } from "./src/core/config.ts";

// Load from config directory (precedence: env > file > defaults)
const config = await loadConfig("./config");

// Validate raw config
const result = validateConfig(rawObj);
if (!result.valid) {
  console.error(result.errors);
}

// Mask secrets in output
console.log(maskSecrets(config));
// { persistence: { dataDir: "/data" }, capabilities: { github: { token: "***" } } }

Hot-reload: ConfigWatcher watches the config file for changes with debounced file watching to prevent reload storms. On change, the new config is validated before replacing the active one — if validation fails, the previous config is retained and the error is logged, enabling automatic rollback to a known-good state.

Env var overrides: KAYAK_PERSISTENCE_DATA_DIR → config.persistence.dataDir

Rate Limiting

Token bucket rate limiter for external API calls. Smooths burst traffic while enforcing average rate.

import { TokenBucket, RateLimiter } from "./src/core/rate-limiter.ts";

const bucket = new TokenBucket({
  capacity: 100,        // max burst
  refillRate: 10,       // tokens per interval
  refillIntervalMs: 1000,
});

const limiter = new RateLimiter(bucket);

// Wrap async function — throws on limit exceeded
const limitedFetch = limiter.wrap(fetch);
await limitedFetch("https://api.example.com");

// Or wait for tokens
const waitingFetch = limiter.wrapWithWait(fetch);
await waitingFetch("https://api.example.com"); // blocks until tokens available

Bounded Queue

Queue with configurable overflow policies for backpressure handling.

import { BoundedQueue } from "./src/core/bounded-queue.ts";

const queue = new BoundedQueue<Event>({
  maxSize: 1000,
  policy: "drop-oldest",  // or: drop-newest, block, reject
});

queue.push(event);
const next = queue.shift();

Overflow policies:

Policy	Behavior
drop-oldest	Remove oldest item when full
drop-newest	Discard new item when full
block	Wait until space available
reject	Throw error when full

Reliability Patterns

Circuit breaker, retry, and fallback for fault tolerance.

import { CircuitBreaker } from "./src/core/circuit-breaker.ts";
import { withRetry } from "./src/core/retry.ts";
import { executeWithFallback } from "./src/core/fallback.ts";

// Circuit breaker — opens after N failures
const breaker = new CircuitBreaker({ failureThreshold: 5, recoveryTimeMs: 30_000 });

// Retry with backoff
const result = await withRetry(fn, { maxRetries: 3, baseDelayMs: 100 });

// Fallback — try primary, fall back on failure
const result = await executeWithFallback(primaryFn, fallbackFn, breaker);

Error Taxonomy

Typed error hierarchy with error codes and retryability.

import { AppError, ValidationError, TimeoutError, RateLimitError } from "./src/core/errors.ts";

// All errors extend AppError
throw new ValidationError("Invalid input", { field: "name" });
throw new TimeoutError("Request timed out", { timeoutMs: 5000 });
throw new RateLimitError("Rate limited", { retryAfterMs: 60_000 });

// Error codes are unique strings
// AppError carries context, module, and operation

Capability Layer

Abstract interfaces for external systems. Capabilities are pluggable and independently testable.

Capability	Interface	Implementation
Shell	IShellCapability	Real — Deno.Command with safety constraints
Sandbox	ISandboxRuntime	Real — Docker/gVisor with hardened flags
Git	IGitCapability	Stubbed — simulated data
GitHub	IGitHubCapability	Stubbed — simulated data
Kubernetes	IKubernetesCapability	Stubbed — simulated data

Capabilities follow a common pattern:

interface ICapability {
  readonly definition: CapabilityDefinition;
  initialize(context: CapabilityContext): Promise<void>;
  dispose(): Promise<void>;
}

interface CapabilityResult<T> {
  success: boolean;
  data?: T;
  error?: string;
}

Projection Layer

UI surfaces subscribe to the event stream and render events. Projections are independent — multiple can run simultaneously.

Projection Protocol

Subscription management with event filtering, pause/resume, and reconnection support.

const protocol = new ProjectionProtocol(eventStream);

// Subscribe to a session
const sub = protocol.subscribe("abc-123", (event) => {
  console.log(`[${event.event_type}]`, event.payload);
}, {
  filter: { event_types: ["tool.execution.started", "tool.execution.completed"] },
});

// Pause and resume
protocol.pause(sub.id);
protocol.resume(sub.id);

Terminal Projection

ANSI-colored event rendering for CLI surfaces.

const terminal = new TerminalProjection(protocol);
await terminal.start("abc-123");

WebSocket Projection

Real-time event delivery to connected clients via WebSocket. Supports subscription management, gap recovery, and backpressure.

import { WebSocketProjectionServer } from "./src/projection/websocket-server.ts";

const server = new WebSocketProjectionServer(eventStore, { port: 8080 });
await server.start();

// Clients connect via WebSocket, receive:
// - Welcome message with version and capabilities
// - Events matching their subscription filter
// - Gap recovery on reconnect

Heartbeat lifecycle: The WebSocket server maintains connection health via configurable heartbeat pings. heartbeatIntervalMs controls how often pings are sent; pongTimeoutMs determines how long to wait for a pong response before closing a stale connection.

Per-session event ordering: Events are inserted into each session's pendingEvents queue via binary search on sequence number, ensuring correct ordering even when events arrive out of order from concurrent sources.

Backpressure: Each client session uses a BoundedQueue for its send queue. When the queue fills, the configured overflow policy (drop-oldest, drop-newest, block, reject) applies, preventing slow consumers from blocking the event stream.

Client messages: subscribe, unsubscribe, reconnect, pongServer messages: welcome, event, error, ping

Review CLI

Standalone code review tool in review/. Runs custom checks and delegates to external CLI tools, producing prioritized findings. Not part of the agent runtime — executes directly via deno task review.

Check registry pattern: registry.ts scans review/checks/ and review/delegate/ at startup. Each file must export a default ReviewCheck object. No configuration or wiring required — add a file, it runs.

Component	File	Purpose
Finding	types.ts	A single finding: title, body, priority (1–3), confidence, file path, line range
ReviewCheck	types.ts	Interface: name, description, run(ctx) → Finding[]
ReviewContext	types.ts	Pre-computed file metadata shared across all checks
buildContext	context.ts	Scans src/ once; builds file map, test pairing, dependency graph
loadChecks	registry.ts	Auto-discovers custom checks from review/checks/
loadDelegates	registry.ts	Auto-discovers tool delegates from review/delegate/
formatFindings	formatter.ts	Groups findings by file, renders with priority colors and summary
main	cli.ts	Parses --only/--skip/--fail-on, runs preflight + checks, outputs results

CLI flags:

Flag	Effect
--only <name>	Run only the named check(s)
--skip <name>	Skip the named check(s) or preflight
--fail-on <priority>	Exit 1 if any finding has priority ≤ threshold (CI gate)

Custom check example:

import type { Finding, ReviewCheck, ReviewContext } from "../types.ts";

const check: ReviewCheck = {
  name: "my-check",
  description: "What this check does",
  async run(ctx: ReviewContext): Promise<Finding[]> {
    const findings: Finding[] = [];
    for (const [, entry] of ctx.files) {
      // scan and produce findings
    }
    return findings;
  },
};

export default check;

Delegate example (wraps an external CLI tool):

import type { Finding, ReviewCheck } from "../types.ts";

const check: ReviewCheck = {
  name: "my-tool",
  description: "Runs an external tool",
  async run(): Promise<Finding[]> {
    const cmd = new Deno.Command("my-tool", { args: [...], stdout: "piped" });
    const output = await cmd.output();
    // parse output into Finding[]
    return findings;
  },
};

export default check;

Data Flow

Design Principles

1. Events are the source of truth — All state is derivable from events. No hidden mutable state.
2. UI independence — Agent runtime never knows about UI surfaces. Projections are disposable.
3. Provider independence — Model providers are abstracted. Swap without code changes.
4. Capability independence — External systems are abstracted. Test with mocks.
5. Session isolation — Events from different sessions never mix. Concurrent sessions are safe.
6. Recovery by replay — Any session state can be reconstructed from its event stream.