@chamvi/core

Core engine for self-hosted, channel-driven AI agents: agents, channels and durable runs.
You write the agent with the AI SDK. The engine queues its runs, checkpoints every step,
pauses for approvals and picks interrupted runs up again, in a plain Node process.

Installation · Usage · Options · API · Notes

Why @chamvi/core?

The AI SDK runs the model and its tool loop, and stops there. An agent that reacts to webhooks and schedules also needs a place to queue the work, a way to survive a pod restart halfway through a run, a guard against posting the same comment twice after a retry, and a way to wait a day for a human to approve a call. That usually means a hosted orchestrator, or a lot of glue code.

@chamvi/core is that layer as a library, with no orchestrator server and no Redis:

Models, messages and tools stay AI SDK types: defineModelAgent() takes a LanguageModel (or a '<provider>/<model id>' string the engine resolves through EngineOptions.models) and a plain ToolSet (tools built with tool(), or defineTool() for tools that read the run context), and nothing is wrapped in a chamvi type.

Table of Contents

• Installation
• Usage
  • Agents and channels
  • Effect tools
  • Approvals
  • Agent state
  • Stores and repositories
• Options
• API
• Notes
• License

Installation

npm install @chamvi/core ai --save

ESM only, Node.js >=26.0.0. ai@^7 (the Vercel AI SDK) is a peer dependency; add the provider package of your model as well, for example @ai-sdk/anthropic.

Usage

One agent, an in-memory store, the development console channel and one manually dispatched run that answers where it came from:

import type { LanguageModel } from 'ai';
import { z } from 'zod';
import {
    Engine,
    consoleChannel,
    createMemoryStore,
    defineEffectTool,
    defineModelAgent,
} from '@chamvi/core';

export async function triageIssue(model: LanguageModel) {
    const labels: string[] = [];
    const lines: string[] = [];

    const triage = defineModelAgent({
        name: 'triage',
        model,
        instructions: 'Label the issue in the trigger payload as bug, feature or question.',
        budget: { maxSteps: 5 },
        tools: {
            // the result is recorded per (run, tool, input): a resumed run reuses it
            addLabel: defineEffectTool({
                description: 'Adds a label to the issue.',
                inputSchema: z.object({ label: z.enum(['bug', 'feature', 'question']) }),
                execute: async ({ label }, { context }) => {
                    labels.push(label);
                    // answers where the run came from, at most once per message
                    await context.channels.findOne()?.emit({ body: `Labelled as ${label}.` });

                    return { added: label };
                },
            }),
        },
    });

    const engine = new Engine({
        store: createMemoryStore(),
        agents: [triage],
        channels: [consoleChannel({
            write: (line) => {
                lines.push(line);
            },
        })],
    });

    await engine.start();

    const { runId } = await engine.dispatch({
        agent: 'triage',
        trigger: 'manual',
        channel: 'console',
        address: 'issue-1',
        payload: { title: 'App crashes on start' },
    });

    await engine.idle();
    await engine.stop();

    return {
        run: await engine.getRun(runId),
        labels,
        lines,
    };
}

The run settles completed, and its output is the text of the model's last answer (with structured output, the parsed value); the console channel wrote Labelled as bug. once. This is the code of test/fixtures/readme-example.ts, which the test suite runs against a scripted model.

Agents and channels

• defineModelAgent() checks the name (^[a-z0-9]+(-[a-z0-9]+)*$, at most 64 characters) and that subscription ids are unique, and fills in the default budget (maxSteps: 20, timeoutMs: 900_000).
• tools is a plain AI SDK ToolSet, defined once at module scope, or a factory (ctx) => ToolSet for dynamic sets (called once per execution). Tools built with defineTool() (or defineEffectTool()) read the run context as context in execute: run (runId, agent, trigger, signal), the agent's state and the run's artifacts. One context per execution, shared by every call and by the factory; treat it as immutable.

const lookup = defineTool({
    inputSchema: z.object({ issue: z.number() }),
    execute: async ({ issue }, { context }) => fetchIssue(issue, { signal: context.run.signal }),
});

• Only tools whose contextSchema is chamvi's toolContextSchema get that context. A third-party tool with its own contextSchema gets its entry of the agent's toolsContext, passed through by reference. An entry the tool's schema rejects fails the run with validation_error, without recording the value. An entry naming a chamvi tool or no tool is an InvalidAgentDefinitionError (thrown by defineModelAgent() for a plain set, a failed run for a factory).

tools: { search: searchTool },
toolsContext: { search: { apiKey: process.env.SEARCH_KEY } },

• A channel (IChannel: name, optional defaultAddress, start, stop, in, out) is anything a run comes from or talks to; new Engine({ channels }) registers them (names follow the agent-name rule, unique per engine). start() starts every channel after the first recovery (start({ channels: false }) starts none), drain() and stop() stop them first. An agent subscribes through channels: ChannelSubscription[], each naming a registered channel (an unknown one is refused). A channel dispatches through its ChannelContext, which sets channel to its own name.
• Outbound is tool-driven: the engine never sends a message of its own. Inside a tool, ctx.channels.findOne(name) returns a handle on a registered channel with out, or undefined; without a name it is the run's source channel, bound to its source address. ctx.channels.findOneOrFail(name?) throws ChannelNotFoundError (channel_not_found) instead of undefined, and ctx.channels.find() returns every channel with out. handle.emit({ title?, body, key? }, address?) sends through the channel's out; the address defaults to the source address on the source channel, else to the channel's defaultAddress (consoleChannel() has its name, the OpenClaw channel its deliver.to); without either it is a validation_error tool error. An emit is an effect: its key (<runId>:emit:<channel>:<sha256 of address and message>, or your key) is recorded in the effect store, so identical emits of one step and a step redone after a crash send once, and out gets it as message.key to dedupe on; a throwing out records nothing and is a tool error. emit outside a tool call (in the tools factory) throws InvalidRunStateError.
• consoleChannel({ name?, write? }) is the development channel (out only, default name console, default write console.log): one header line per message, the body lines and, for an approval, until when it waits; it never throws.
• The trigger envelope (channel, subscription id, payload; address is where the run came from on the channel) becomes the run's first user message, as a JSON block.
• dispatch({ ..., attachments: [{ mediaType, filename?, data }] }) starts a run with files: each is stored as an artifact of the run before the run is created (a deduplicated dispatch stores nothing), its metadata lands on run.trigger.attachments (ids <runId>.0000, <runId>.0001, ..., so a listing keeps the input order), and the model gets the bytes as file parts of the first message. Checkpoints keep a reference, never the bytes; a missing artifact fails the run with code artifact_not_found.
• The tool context (also the factory's argument) carries artifacts, scoped to the run: put({ mediaType, filename?, data }), get(id) and list(). put() is no effect by itself; call it inside a defineEffectTool to store a file exactly once.
• output takes an AI SDK Output (e.g. Output.object({ schema })): a completed run's output is then the parsed value instead of the text, and a final answer that does not parse fails the run with validation_error (the quoted cause cut to 500 characters; the whole answer stays in the llm step). An answer cut off at the output token limit, or a loop that ended on a tool call nothing executed, fails it with incomplete_response.
• The engine streams every model call: engine.on('run.delta', ({ delta }) => ...) gets { runId, step, kind: 'text' | 'reasoning', text } while the model writes. Deltas are never stored and reach only listeners in the process executing the run.
• A model call that fails mid-stream still ends its step: the llm step (text so far, finish reason error) and a checkpoint are written, then the run fails with the model's error. A stream that closes without the provider's finish part fails the run with incomplete_response and never runs that step's tool calls.

Effect tools

const comment = defineEffectTool({
    inputSchema: z.object({ issue: z.number(), body: z.string() }),
    effectKey: ({ issue }, ctx) => `${ctx.run.runId}:comment:${issue}`,
    // finds a comment an earlier, crashed attempt already posted
    alreadyApplied: async (key, { issue }) => findComment(issue, effectMarker(key)),
    execute: async ({ issue, body }, { context }) => postComment(
        issue,
        `${body}\n${effectMarker(`${context.run.runId}:comment:${issue}`)}`,
    ),
});

const agent = defineModelAgent({
    name: 'responder',
    model,
    instructions: 'Answer new issues.',
    tools: { comment },
});

• The effect key defaults to ${ctx.run.runId}:${toolName}:${sha256(input)}; effectKey(input, ctx) replaces it. effectKey and alreadyApplied get the same tool context as execute (ctx.run, ctx.state, ctx.artifacts); in core 0.1 they got the run context, so ctx.runId is now ctx.run.runId.
• On every call the engine tries the recorded result first, then alreadyApplied, then execute. The result is recorded after execute returns. A throwing execute records nothing, so a retry may apply it.
• effectMarker(key) is a hidden HTML comment to embed in what the tool creates, so that alreadyApplied can find it again.

Approvals

const deploy = defineTool({
    description: 'Deploy a version.',
    inputSchema: z.object({ version: z.string() }),
    execute: async ({ version }, { context }) => {
        // pauses the run until the owner answers; the call then runs again and gets the decision
        const decision = await approve(context, { title: `Deploy ${version}?`, timeoutMs: 60 * 60 * 1000 });
        if (!decision.approved) return `Not deployed: ${decision.reason ?? 'denied'}`;
        return deployVersion(version);
    },
});

const deployer = defineModelAgent({
    name: 'deployer',
    model,
    instructions: 'Deploy the requested version.',
    tools: { deploy },
});

const engine = new Engine({
    store,
    agents: [deployer],
});

// later, from the owner's answer to run.waiting
await engine.approve(runId, { approvalId, approved: true });

• approve(ctx, { title, body?, timeoutMs? }) runs inside a tool call (its execute or the agent's hooks). It puts the run in waiting (run.waiting names the approval, <stepIndex>:<toolCallId>:<n>) and frees its slot; the approval survives restarts. Once answered, the call runs again from the start and approve() returns { approved, reason?, approver? }, so code before it must be idempotent. Other calls of the same step that finished keep their results.
• ctx.channels.findOneOrFail(name).approve({ title, address? }) is the same, with the request delivered through that channel.
• To gate a tool you did not write, call approve() in the agent's beforeToolExecution hook and throw ToolDeniedError for a denial.
• engine.approve(runId, { approvalId, approved, reason?, approver? }) answers the current wait point, for a caller that authorized the approver itself. An answer is single use.
• An approval not answered within its timeoutMs (default 24 hours) is denied with the reason approval timed out.

Untrusted content

Tools can declare trust: { taint: true, guardWhenTainted: true } on defineTool() or defineEffectTool(). taint marks the run before a result or execution error is exposed to the model; guardWhenTainted requires approval or denies execution when the run is already tainted. Each flag accepts a boolean or an (input, ctx) => boolean | Promise<boolean> predicate. A channel with untrusted: true starts tainted runs. Third-party tools are trusted by default; classify them by name with the agent's trust.tools overrides.

trust: {
    onTainted: 'approve', // default; use 'deny' to refuse outbound calls
    tools: {
        externalSearch: { taint: true },
        publish: { guardWhenTainted: true },
    },
},

The first exposure is persisted as run.taint: { at: Date, source: string } before the result reaches the model. It survives checkpoints and restarts; dispatched runs inherit it, a parent that waits on a tainted child is tainted before the outcome reaches its model, and conversation recall conservatively taints the receiving run. Custom stores must preserve this optional field. The database migration also marks pre-existing unfinished runs as tainted because their checkpoints may already contain external content.

Upgrade with downtime, not a rolling deployment: stop all old ingress, workers and approval pollers across every replica and scheduled job, wait for them to exit, then migrate/backfill, then start only new binaries. Do not mix old and new runtimes on the same store. Old pollers can disclose a new automatic approval's private argument-bearing body to its source channel; old workers can create untainted runs after the one-time backfill. Draining executions alone is not sufficient. Custom stores need the same stop/backfill/start sequence. See Multiple replicas.

After exposure, outbound tools require durable approval (or are denied). The gate runs on the final validated input after beforeToolExecution, before effect lookup; changing the arguments requires a new decision. Full arguments stay in the approval inbox/API. Automatic source-channel notifications contain only a generic review request. Direct channel emits are gated too, except sends to the source address on the same channel or a code-configured default address. These are fixed-destination exceptions, not guarantees about the message body.

extract(ctx, { schema, content, instructions?, model? }) calls a separate model with structured output and no tools. Acquire raw content in trusted tool code and return only the extraction; an ordinary untrusted read followed by extraction does not undo taint. Only conservative constrained schemas avoid taint: literal constants (including null), booleans/null, numbers, fixed enums, closed objects, arrays bounded to 10 items, and patterned strings bounded to 200 characters. Free text, references, combinators and unknown schema features taint the run. Results are effect-cached; extraction never clears existing taint. The caller still owns authorization and semantic validation of extracted values.

Limits: this is not a prompt-injection-proof sandbox. Enabled shell execution can bypass outbound-tool gates, including through scripts and nested processes. Shell command policies and the optional permission classifier are separate work. State provenance, provider-executed tools, sandbox network egress and disclosure to the model provider are outside this boundary. Source replies/default-address sends and unclassified third-party tools remain trusted capabilities. GitHub read tools send model-selected queries, paths and refs to the GitHub API without an outbound gate; classify those tools as outbound with trust.tools if the API operator or its logs must not receive arbitrary run data without approval.

Invokers and tool access

A run can carry its invoker: who started it, as an authenticator resolved it (@chamvi/http sets it from the request's principal). toolPolicy decides per tool whether the run may use it; a refused tool is not offered to the model.

const ops = defineModelAgent({
    name: 'ops',
    model,
    instructions: 'Operate the service.',
    tools: { deploy, rollback },
    // offers a tool only when the invoker holds tool:ops:<tool>; cron and webhook runs get none
    toolPolicy: requireToolPermission(),
    hooks: {
        // release managers deploy without asking
        async beforeToolExecution({ toolName, input }, ctx) {
            const { invoker } = ctx.run.trigger;
            if (toolName !== 'deploy' || (invoker && hasPermission(invoker.permissions, 'release:manage'))) return input;
            const decision = await approve(ctx, { title: 'Deploy?' });
            if (!decision.approved) throw new ToolDeniedError('Deploy denied.');
            return input;
        },
    },
});

const { runId } = await engine.dispatch({
    agent: 'ops',
    trigger: 'http',
    channel: 'http',
    payload: {},
    invoker: { id: 'u-42', kind: 'user', permissions: ['tool:ops:deploy'] },
});

// later, once the run waits: the approver is recorded on the run's wait step
const { approvalId } = (await engine.getRun(runId))!.waiting!;
await engine.approve(runId, { approvalId, approved: true, approver: { id: 'u-7', kind: 'user', permissions: [] } });

• The invoker ({ id, kind, name?, permissions }) is a snapshot taken at dispatch. A run keeps it across approvals and restarts; revoking a grant later does not reach a run that already holds it.
• With an invoker, the idempotencyKey is scoped to it (<agent>:<channel>:<kind>:<id>): only the same invoker's retry is deduplicated, another invoker with the same key gets its own run.
• Tool code sees it as context.run.trigger.invoker in tools and the factory and ctx.run.trigger.invoker in effect hooks and the agent hooks.
• The policy covers state tools too. A call to a refused tool that still arrives fails as a tool error and the run goes on: the step records it as tool_not_allowed (ToolNotAllowedError), whether the model called it anyway (the model sees the AI SDK's unavailable-tool error) or an approved call from an older history reached the engine's re-check.
• Permissions are :-separated: * alone grants everything, a * segment matches one segment, and segment counts must match. requireToolPermission({ withoutInvoker: 'allow' }) lets runs without an invoker use every tool.

Agent state

const digest = defineModelAgent({
    name: 'digest',
    model,
    instructions: 'Track open loops across days.',
    state: { loops: z.object({ title: z.string(), due: z.string().optional() }) },
    tools: ({ state }) => ({
        countLoops: tool({
            inputSchema: z.object({}),
            execute: async () => (await state.collection('loops').list()).length,
        }),
    }),
});

• state.get and state.set are a key-value store per agent. Each declared collection holds documents, validated by its Standard Schema on upsert.
• The model also gets each collection as state_<collection>_list, state_<collection>_upsert and state_<collection>_remove tools; upsert and remove are effect tools.

Stores and repositories

createMemoryStore() keeps everything in process memory, for development and tests. The factory assembles separate capability repositories (runs, steps, checkpoints, idempotency keys, locks, effects, state, artifacts, agent revisions and optional conversations) into a plain IStore aggregate, not a facade with forwarding methods, and proves their contracts with the shared contract suite from @chamvi/test-kit:

import { describeRepositoryContract } from '@chamvi/test-kit';

describeRepositoryContract((clock) => createMemoryStore({ clock }));

The factory receives a ControllableClock and must use it for every expiry decision. The suite runs on Vitest.

Repository queries use findOne and find, with matching findOneOrFail and findOrFail methods when absence should throw. Collection results keep their existing shape: runs.find returns { data, total }, while artifacts.find returns an array. OrFail checks the selected result, so an empty page throws even when its total is nonzero. State values such as null, false and 0 are present. Domain mutations keep their specific names, such as claim and beginTurn.

Context

A tool reaches a shared instance (an API client, the GitHub App, a workspace manager) through a typed key instead of a closure. The engine keeps the value by reference; inject(ctx, key) reads it in a tool's execute, in the tools factory and in the effect hooks, and nothing provided is ever stored or serialized (ToolContext is { run, state, artifacts, channels } and gains no field for a provided value).

import { defineContextKey, defineTool, inject } from '@chamvi/core';

const DEPLOY = defineContextKey<IDeployClient>('deploy');

const deploy = defineTool({
    description: 'Deploys a version.',
    inputSchema: z.object({ version: z.string() }),
    execute: async ({ version }, { context }) => inject(context, DEPLOY).deploy(version),
});

engine.provide(DEPLOY, client); // once per key, after new Engine()

A missing value throws ContextNotProvidedError (context_not_provided, naming the key only): a tool error in execute and the effect hooks, a failed run (before any model call) in the tools factory. tryInject(ctx, key) returns undefined instead. provide refuses undefined, a key not from defineContextKey() and a second value for one key.

Options

new Engine(options):

defineModelAgent(definition):

Structured instructions

Strings remain shorthand for one system message. Native message types are available directly from core:

import { defineModelAgent } from '@chamvi/core';
import type { AgentInstructions, SystemModelMessage, UserModelMessage } from '@chamvi/core';

const policy: SystemModelMessage = { role: 'system', content: 'Summarize the report.' };
const example: UserModelMessage = { role: 'user', content: [{ type: 'text', text: 'All checks passed.' }] };
const instructions: AgentInstructions = [policy, example, { role: 'assistant', content: 'Ready to release.' }];
const agent = defineModelAgent({ name: 'reporter', instructions });

System content is a string; user/assistant content and providerOptions use their native SDK types. Leading system blocks go through SDK instructions; the remaining messages stay in order before the run transcript, including any later system blocks. First-class skills append a separate metadata system block. These configured messages are transient: they hydrate artifact markers for model/tool context, stay out of checkpoints and conversation history, and never become pending replay calls. Recalled conversation messages follow this configured prefix, and approval resume uses the captured conversation snapshot. Resumed executions rebuild them from the current definition. Stored transcript system roles are rejected before replay/model work. Manual SkillCatalog.instructions remains a string for explicit composition.

See the instructions guide for the typed example and role-specific content limits.

API

Types: Agent, AgentDefinition, ToolContext, ContextKey, IContextProvider, ToolPolicy, AgentHooks, ToolCallInfo, TriggerEnvelope, Invoker, Run, RunStatus, Step, Checkpoint, WaitPoint, Budget, Usage, ApprovalRequestInput, ApprovalDecision, ApprovalAnswer, IStore and its capability interfaces, MemoryStoreOptions, Artifact, ArtifactInput, IRunArtifacts, IChannel, ChannelSubscription, ChannelContext, ChannelMessage, ChannelSendOptions, ChannelHandle, EmitMessage, ConsoleChannelOptions, IAgentState, RunEvent, EngineOptions, ModelRef, IModelResolver, IClock.

Test helpers (ControllableClock, describeRepositoryContract) live in @chamvi/test-kit.

Notes

EngineOptions.skills supplies defaults for every agent; AgentDefinition.skills adds scoped extras. Both accept the generic SkillInput protocol from core. Import skill data types, gitSkills, loadSkillBundle, custom adapters (defineSkill, defineSkillSource) and catalogs from optional @chamvi/skills. Its factory values carry the runtime automatically, with no module registration. Core coordinates preparation, cancellation and ordinary contributed-tool execution; the optional package owns validation, contents, catalogs and caches. Core works without skills installed.

• A tool error does not fail the run. It goes back to the model as a tool error result. Throw FatalToolError to fail the run; the step budget caps retry loops.
• Budgets settle runs. A run stopped by maxSteps or maxTokens while the model still wanted to call tools fails with code BudgetExceeded; a run that hits its timeout fails with code Timeout.
• Effects are recorded after they happen. A crash between a tool's side effect and the store write repeats the effect on resume, unless alreadyApplied finds it.
• Several approvals in one step are handled one at a time. The run waits on the first in call order; the other suspended calls stay pending and ask again after the resume, before any further model call.
• createMemoryStore() is not durable. Everything is gone with the process; use a persistent store to survive restarts or to run replicas.
• stop() stops polling and the channels. It resolves once every channel stopped; runs in progress continue, so use drain() to shut down.

License

Published under Apache License 2.0.