omk-agent-core

omk-agent-core

Stateful agent with tool execution and event streaming. Built on omk-ai.

Installation

npm install omk-agent-core

Quick Start

import { Agent } from "omk-agent-core";
import { getModel } from "omk-ai";

const agent = new Agent({
  initialState: {
    systemPrompt: "You are a helpful assistant.",
    model: getModel("anthropic", "claude-sonnet-4-20250514"),
  },
});

agent.subscribe((event) => {
  if (event.type === "message_update" && event.assistantMessageEvent.type === "text_delta") {
    // Stream just the new text chunk
    process.stdout.write(event.assistantMessageEvent.delta);
  }
});

await agent.prompt("Hello!");

Core Concepts

AgentMessage vs LLM Message

The agent works with AgentMessage, a flexible type that can include:

• Standard LLM messages (user, assistant, toolResult)
• Custom app-specific message types via declaration merging

LLMs only understand user, assistant, and toolResult. The convertToLlm function bridges this gap by filtering and transforming messages before each LLM call.

Message Flow

AgentMessage[] → transformContext() → AgentMessage[] → convertToLlm() → Message[] → LLM
                    (optional)                           (required)

• transformContext: Prune old messages, inject external context
• convertToLlm: Filter out UI-only messages, convert custom types to LLM format

Event Flow

The agent emits events for UI updates. Understanding the event sequence helps build responsive interfaces.

prompt() Event Sequence

When you call prompt("Hello"):

prompt("Hello")
├─ agent_start
├─ turn_start
├─ message_start   { message: userMessage }      // Your prompt
├─ message_end     { message: userMessage }
├─ message_start   { message: assistantMessage } // LLM starts responding
├─ message_update  { message: partial... }       // Streaming chunks
├─ message_update  { message: partial... }
├─ message_end     { message: assistantMessage } // Complete response
├─ turn_end        { message, toolResults: [] }
└─ agent_end       { messages: [...] }

With Tool Calls

If the assistant calls tools, the loop continues:

prompt("Read config.json")
├─ agent_start
├─ turn_start
├─ message_start/end  { userMessage }
├─ message_start      { assistantMessage with toolCall }
├─ message_update...
├─ message_end        { assistantMessage }
├─ tool_execution_start  { toolCallId, toolName, args }
├─ tool_execution_update { partialResult }           // If tool streams
├─ tool_execution_end    { toolCallId, result }
├─ message_start/end  { toolResultMessage }
├─ turn_end           { message, toolResults: [toolResult] }
│
├─ turn_start                                        // Next turn
├─ message_start      { assistantMessage }           // LLM responds to tool result
├─ message_update...
├─ message_end
├─ turn_end
└─ agent_end

Tool execution mode is configurable:

• parallel (default): preflight tool calls sequentially, execute allowed tools concurrently, emit tool_execution_end as soon as each tool is finalized, then emit toolResult messages and turn_end.toolResults in assistant source order
• sequential: execute tool calls one by one, matching the historical behavior

In parallel mode the batch can be scheduled with one of two schedulers:

• dag-v2 (default): builds a deterministic resource-claim DAG per call. Tools declare resourceClaims and an access mode (read or write); the scheduler runs independent calls in source-directed levels, keeps bash, unknown tools, and unclaimed extension tools exclusive, and still preserves source-order result artifacts. A claim-planning failure closes only that call and does not block other claimable calls in the batch.
• waves-v1 (rollback): partitions the batch into ordered waves using partitionToolBatchWaves. Calls in the same wave run concurrently, and waves run one after another in source order. Set toolScheduler: "waves-v1" only when temporarily rolling back DAG scheduling. The OMK CLI also accepts OMK_TOOL_SCHEDULER=waves-v1.

Tool completion events follow tool completion order, but persisted toolResult messages still follow assistant source order.

The mode can be set globally via toolExecution in the agent config, or per-tool via executionMode on AgentTool. If any tool call in a batch targets a tool with executionMode: "sequential", the entire batch executes sequentially regardless of the global setting.

The beforeToolCall hook runs after tool_execution_start and validated argument parsing. It can block execution. The afterToolCall hook runs after tool execution finishes and before tool_execution_end and final tool result message events are emitted.

Tools can also return terminate: true to hint that the automatic follow-up LLM call should be skipped. The loop only stops early when every finalized tool result in that batch sets terminate: true. Mixed batches continue normally.

Low-level loop callers can set shouldStopAfterTurn to stop gracefully after the current turn completes:

const stream = agentLoop(prompts, context, {
  model,
  convertToLlm,
  shouldStopAfterTurn: async ({ message, toolResults, context, newMessages }) => {
    return shouldCompactBeforeNextTurn(context.messages);
  },
});

shouldStopAfterTurn runs after turn_end is emitted and after the assistant response and any tool executions have completed normally. If it returns true, the loop emits agent_end and exits before polling steering or follow-up queues, and before starting another LLM call. It does not abort the provider stream, does not cancel running tools, and does not alter the assistant message stop reason.

Set maxTurns on either AgentLoopConfig or AgentOptions to bound provider calls in one prompt or continuation run. Each assistant response counts as one turn. At the limit, the current response and tool batch finish normally, then the loop emits agent_end before prepareNextTurn, stop hooks, or steering/follow-up queues run again. The option is unbounded when omitted and must be a positive safe integer when configured.

When you use the Agent class, assistant message_end processing is treated as a barrier before tool preflight begins. That means beforeToolCall sees agent state that already includes the assistant message that requested the tool call.

continue() Event Sequence

continue() resumes from existing context without adding a new message. Use it for retries after errors.

// After an error, retry from current state
await agent.continue();

The last message in context must be user or toolResult (not assistant).

Event Types

Event	Description
agent_start	Agent begins processing
agent_end	Final event for the run. Awaited subscribers for this event still count toward settlement
turn_start	New turn begins (one LLM call + tool executions)
turn_end	Turn completes with assistant message and tool results
message_start	Any message begins (user, assistant, toolResult)
message_update	Assistant only. Includes assistantMessageEvent with delta
message_end	Message completes
tool_execution_start	Tool begins
tool_execution_update	Tool streams progress
tool_execution_end	Tool completes
tool_execution_late_settlement	Tool promise settled after a terminal result was already committed (audit only)

Agent.subscribe() listeners are awaited in registration order. agent_end means no more loop events will be emitted, but await agent.waitForIdle() and await agent.prompt(...) only settle after awaited agent_end listeners finish.

Agent Options

const agent = new Agent({
  // Initial state
  initialState: {
    systemPrompt: string,
    // UTF-16 offset ending the stable provider-cacheable prefix.
    systemPromptCacheBoundary?: number,
    // Disable explicit cache affinity/markers for a dynamic replacement.
    systemPromptCacheBoundaryBypass?: boolean,
    model: Model<any>,
    thinkingLevel: "off" | "minimal" | "low" | "medium" | "high" | "xhigh",
    tools: AgentTool<any>[],
    messages: AgentMessage[],
  },

  // Convert AgentMessage[] to LLM Message[] (required for custom message types)
  convertToLlm: (messages) => messages.filter(...),

  // Transform context before convertToLlm (for pruning, compaction)
  transformContext: async (messages, signal) => pruneOldMessages(messages),

  // Steering mode: "one-at-a-time" (default) or "all"
  steeringMode: "one-at-a-time",

  // Follow-up mode: "one-at-a-time" (default) or "all"
  followUpMode: "one-at-a-time",

  // Custom stream function (for proxy backends)
  streamFn: streamProxy,

  // Session ID for provider caching
  sessionId: "session-123",

  // Dynamic API key resolution (for expiring OAuth tokens)
  getApiKey: async (provider) => refreshToken(),

  // Optional positive safe-integer provider-turn budget for each run
  maxTurns: 50,

  // Tool execution mode: "parallel" (default) or "sequential"
  // Also "toolExecution: 'sequential'" forces waves/dag schedulers to run serially.
  toolExecution: "parallel",

  // Tool scheduler: "dag-v2" (default) or legacy "waves-v1" rollback
  toolScheduler: "dag-v2",

  // Maximum concurrent tool calls in one DAG level when using "dag-v2".
  // 0 removes the cap. Default in the OMK CLI is 4.
  maxToolConcurrency: 4,

  // Require extension tools to declare resource claims when using "dag-v2".
  // When true, unclaimed extension tools are treated as exclusive.
  strictExtensionClaims: false,

  // Working directory used for claim resolution and local tool backends.
  cwd: process.cwd(),

  // Custom resource-key resolver for dag-v2. The default Node resolver uses
  // cwd-relative paths and common OMK namespaces. Import it from
  // `omk-agent-core/node` if you need to customize it.
  // resourceKeyResolver: createNodeResourceKeyResolver({ cwd: process.cwd() }),

  // Fallback tool timeout in milliseconds. 0 disables the fallback timer.
  // Individual tools can also set timeoutMs; toolTimeouts[name] overrides this.
  toolTimeoutMs: 0,

  // Per-tool timeout overrides in milliseconds. 0 disables that tool's timer.
  toolTimeouts: {
    read: 30_000,
    bash: 300_000,
  },

  // What to do when a tool promise settles after timeout/abort has already
  // committed a synthetic result. "audit" (default) emits a
  // tool_execution_late_settlement event; "ignore" silently drops it.
  toolExecutionPolicy: {
    lateSettlement: "audit",
  },

  // Preflight each tool call after args are validated. Can block execution.
  beforeToolCall: async ({ toolCall, args, context }) => {
    if (toolCall.name === "bash") {
      return { block: true, reason: "bash is disabled" };
    }
  },

  // Postprocess each tool result before final tool events are emitted.
  afterToolCall: async ({ toolCall, result, isError, context }) => {
    if (toolCall.name === "notify_done" && !isError) {
      return { terminate: true };
    }
    if (!isError) {
      return { details: { ...result.details, audited: true } };
    }
  },

  // Custom thinking budgets for token-based providers
  thinkingBudgets: {
    minimal: 128,
    low: 512,
    medium: 1024,
    high: 2048,
  },
});

Agent State

interface AgentState {
  systemPrompt: string;
  systemPromptCacheBoundary?: number;
  systemPromptCacheBoundaryBypass?: boolean;
  model: Model<any>;
  thinkingLevel: ThinkingLevel;
  tools: AgentTool<any>[];
  messages: AgentMessage[];
  readonly isStreaming: boolean;
  readonly streamingMessage?: AgentMessage;
  readonly pendingToolCalls: ReadonlySet<string>;
  readonly errorMessage?: string;
}

Access state via agent.state. systemPromptCacheBoundary is a UTF-16 offset into systemPrompt; providers that support the metadata may cache or route the prefix before that offset. Set systemPromptCacheBoundaryBypass when a turn replaces the prompt with dynamic content. Missing, invalid, or bypassed boundaries suppress explicit stable-prefix markers and affinity; provider usage is the only proof of a cache hit.

Assigning agent.state.tools = [...] or agent.state.messages = [...] copies the top-level array before storing it. Mutating the returned array mutates the current agent state.

During streaming, agent.state.streamingMessage contains the current partial assistant message.

agent.state.isStreaming remains true until the run fully settles, including awaited agent_end subscribers.

Methods

Prompting

// Text prompt
await agent.prompt("Hello");

// With images
await agent.prompt("What's in this image?", [
  { type: "image", data: base64Data, mimeType: "image/jpeg" }
]);

// AgentMessage directly
await agent.prompt({ role: "user", content: "Hello", timestamp: Date.now() });

// Continue from current context (last message must be user or toolResult)
await agent.continue();

State Management

agent.state.systemPrompt = "New prompt";
agent.state.model = getModel("openai", "gpt-4o");
agent.state.thinkingLevel = "medium";
agent.state.tools = [myTool];
agent.toolExecution = "sequential";
agent.beforeToolCall = async ({ toolCall }) => undefined;
agent.afterToolCall = async ({ toolCall, result }) => undefined;
agent.state.messages = newMessages; // top-level array is copied
agent.state.messages.push(message);
agent.reset();

Session and Thinking Budgets

agent.sessionId = "session-123";

agent.thinkingBudgets = {
  minimal: 128,
  low: 512,
  medium: 1024,
  high: 2048,
};

Control

agent.abort();           // Cancel current operation
await agent.waitForIdle(); // Wait for completion

Events

const unsubscribe = agent.subscribe(async (event, signal) => {
  if (event.type === "agent_end") {
    // Final barrier work for the run
    await flushSessionState(signal);
  }
});
unsubscribe();

Steering and Follow-up

Steering messages let you interrupt the agent while tools are running. Follow-up messages let you queue work after the agent would otherwise stop.

agent.steeringMode = "one-at-a-time";
agent.followUpMode = "one-at-a-time";

// While agent is running tools
agent.steer({
  role: "user",
  content: "Stop! Do this instead.",
  timestamp: Date.now(),
});

// After the agent finishes its current work
agent.followUp({
  role: "user",
  content: "Also summarize the result.",
  timestamp: Date.now(),
});

const steeringMode = agent.steeringMode;
const followUpMode = agent.followUpMode;

agent.clearSteeringQueue();
agent.clearFollowUpQueue();
agent.clearAllQueues();

Use clearSteeringQueue, clearFollowUpQueue, or clearAllQueues to drop queued messages.

When steering messages are detected after a turn completes:

• All tool calls from the current assistant message have already finished
• Steering messages are injected
• The LLM responds on the next turn

Follow-up messages are checked only when there are no more tool calls and no steering messages. If any are queued, they are injected and another turn runs.

Custom Message Types

Extend AgentMessage via declaration merging:

declare module "omk-agent-core" {
  interface CustomAgentMessages {
    notification: { role: "notification"; text: string; timestamp: number };
  }
}

// Now valid
const msg: AgentMessage = { role: "notification", text: "Info", timestamp: Date.now() };

Handle custom types in convertToLlm:

const agent = new Agent({
  convertToLlm: (messages) => messages.flatMap(m => {
    if (m.role === "notification") return []; // Filter out
    return [m];
  }),
});

Tools

Define tools using AgentTool:

import { Type } from "typebox";

const readFileTool: AgentTool = {
  name: "read_file",
  label: "Read File",  // For UI display
  description: "Read a file's contents",
  parameters: Type.Object({
    path: Type.String({ description: "File path" }),
  }),
  // Override execution mode for this tool (optional).
  // "sequential" forces the entire batch to run one at a time.
  // "parallel" allows concurrent execution with other tool calls.
  // If omitted, the global toolExecution config applies.
  executionMode: "sequential",

  // Resource claims for dag-v2 scheduling. Return "exclusive" to run alone,
  // or a claim list. Omit to let the tool run exclusively (or as an unclaimed
  // extension tool when strictExtensionClaims is false).
  resourceClaims: (args, context) => [
    { kind: "path", key: args.path, access: "read" },
  ],

  // Per-tool timeout in milliseconds. 0 disables the timer for this tool.
  timeoutMs: 30_000,

  execute: async (toolCallId, params, signal, onUpdate) => {
    const content = await fs.readFile(params.path, "utf-8");

    // Optional: stream progress
    onUpdate?.({ content: [{ type: "text", text: "Reading..." }], details: {} });

    // Optional: add `terminate: true` here to skip the automatic follow-up LLM call
    // when every finalized tool result in the batch does the same.
    return {
      content: [{ type: "text", text: content }],
      details: { path: params.path, size: content.length },
    };
  },
};

agent.state.tools = [readFileTool];

Error Handling

Throw an error when a tool fails. Do not return error messages as content.

execute: async (toolCallId, params, signal, onUpdate) => {
  if (!fs.existsSync(params.path)) {
    throw new Error(`File not found: ${params.path}`);
  }
  // Return content only on success
  return { content: [{ type: "text", text: "..." }] };
}

Thrown errors are caught by the agent and reported to the LLM as tool errors with isError: true.

Return terminate: true from execute() or afterToolCall to hint that the agent should stop after the current tool batch. This only takes effect when every finalized tool result in the batch is terminating. The hint is runtime-only; emitted toolResult transcript messages remain standard LLM tool results.

Tool Timeouts and Late Settlement

Every tool call is guarded by a timeout. The effective timeout for a call is resolved in order:

• AgentTool.timeoutMs (per tool)
• AgentLoopConfig.toolTimeouts[name] (per-name override)
• AgentLoopConfig.toolTimeoutMs (fallback)
• No timer when all of the above are 0 or unset

When a timeout wins the race, the agent closes the tool call and commits a synthetic terminal result with a timeout disposition. The same happens for abort or blocked calls. If the provider itself ends with error or aborted while emitting complete, unambiguous tool calls, the agent closes those calls without executing them. Terminal abort paths stop before hooks, queue polling, or another provider request. The transcript retains each terminal result and its disposition.

If the real tool promise settles after the terminal result has already been committed, the agent emits a tool_execution_late_settlement event (when toolExecutionPolicy.lateSettlement is "audit", the default). The event is advisory: it does not change the transcript that the LLM sees.

agent.subscribe((event) => {
  if (event.type === "tool_execution_late_settlement") {
    console.log("Late settlement:", event.toolCallId, event.toolName);
  }
});

Tool Result Dispositions and Transcript Integrity

Every terminal tool result carries a details.omk envelope with a ToolCallDisposition:

• completed - Normal success
• failed - Tool threw an error
• blocked - beforeToolCall blocked execution
• aborted - Caller aborted the run
• timeout - Tool exceeded its timeout
• skipped - Tool was skipped (e.g. duplicate call after transcript repair)

The loop enforces transcript integrity before tool execution, every continuation, and every provider request. It checks for:

• Duplicate tool-call IDs
• Orphan tool results (no matching call)
• Duplicate results for the same call
• Interleaved non-result messages between a call and its result
• Missing results for finalized tool calls

Unambiguous missing-only tail results are repaired by synthesizing skipped/error results. Corrupt or ambiguous transcripts fail closed with an error rather than fabricating an assistant message; tools from an ambiguous assistant turn never execute. This is why thrown errors should be real errors, not strings returned as content.

Proxy Usage

For browser apps that proxy through a backend:

import { Agent, streamProxy } from "omk-agent-core";

const agent = new Agent({
  streamFn: (model, context, options) =>
    streamProxy(model, context, {
      ...options,
      authToken: "...",
      proxyUrl: "https://your-server.com",
    }),
});

Low-Level API

For direct control without the Agent class:

import { agentLoop, agentLoopContinue } from "omk-agent-core";

const context: AgentContext = {
  systemPrompt: "You are helpful.",
  systemPromptCacheBoundary: "You are helpful.".length,
  systemPromptCacheBoundaryBypass: false,
  messages: [],
  tools: [],
};

const config: AgentLoopConfig = {
  model: getModel("openai", "gpt-4o"),
  convertToLlm: (msgs) => msgs.filter(m => ["user", "assistant", "toolResult"].includes(m.role)),
  toolExecution: "parallel",  // overridden by per-tool executionMode if set
  beforeToolCall: async ({ toolCall, args, context }) => undefined,
  afterToolCall: async ({ toolCall, result, isError, context }) => undefined,
};

const userMessage = { role: "user", content: "Hello", timestamp: Date.now() };

for await (const event of agentLoop([userMessage], context, config)) {
  console.log(event.type);
}

// Continue from existing context
for await (const event of agentLoopContinue(context, config)) {
  console.log(event.type);
}

These low-level streams are observational. They preserve event order, but they do not wait for your async event handling to settle before later producer phases continue. If you need message processing to act as a barrier before tool preflight, use the Agent class instead of raw agentLoop() or agentLoopContinue().

License

MIT