AgentHarness

Quick Example

import { AgentHarness, any, callModel, loop, until } from '@noetic-tools/core';
import type { ContextData } from '@noetic-tools/core';
import { z } from 'zod';

const searchTool = {
  name: 'search',
  description: 'Search the web for information',
  input: z.object({ query: z.string() }),
  output: z.object({ result: z.string() }),
  execute: async (args: { query: string }) => ({ result: `Results for: ${args.query}` }),
};

const agent = loop({
  id: 'researcher-loop',
  steps: [callModel<ContextData, string, string>({
    id: 'researcher-llm',
    model: 'openai/gpt-4o',
    instructions: 'You are a research assistant.',
    tools: [searchTool],
  })],
  until: any(until.noToolCalls(), until.maxSteps(10)),
});

const harness = new AgentHarness({
  name: 'researcher',
  agentGraph: agent,
  params: {},
  callModelDefaults: { provider: 'openrouter' }, // reads OPENROUTER_API_KEY from the env
});

await harness.execute('Find recent AI news');
const response = await harness.getAgentResponse();
console.log(response.text);

The AgentHarness is the execution engine at the heart of Noetic. It creates contexts, manages per-thread sessions with a message queue, dispatches steps, coordinates channels and context layers, and provides tracing infrastructure.

Calls to execute() enqueue the input on the session identified by options.threadId (a single default thread is used when omitted). Stream accessors are session-scoped — subscribe once and events flow across every turn in that session.

Constructor Options

import { AgentHarness } from '@noetic-tools/core';

const harness = new AgentHarness({
  name: 'my-agent',
  agentGraph: myStep,
  params: { model: 'openai/gpt-4o' },
});

Environment

The environment option groups the surfaces the agent runs against. Every field is optional and defaults to an in-memory adapter, so a zero-config harness is fully sandboxed.

const harness = new AgentHarness({
  name: 'my-agent',
  params: {},
  environment: {
    storage: { adapter, checkpointStore, layerStateStore, stepLedgerRetention },
    fs,
    shell,
    subprocess,
  },
});

execute()

Enqueues input on the session identified by options.threadId (or the default thread). Returns a Promise<void> that resolves once the message is accepted into the queue — not when the model finishes responding. Use the session-scoped accessors to observe the response.

declare const harness: AgentHarness;

await harness.execute('What is 2+2?');
const response = await harness.getAgentResponse();
console.log(response.text);

// Stream text across all turns in the session
for await (const delta of harness.getTextStream()) {
  process.stdout.write(delta);
}

// With options
await harness.execute('Hello', {
  threadId: 'thread-1',
  resourceId: 'user-1',
});

// Submit a second message while the first turn is still running — it queues
// and runs as a new turn after the current turn completes.
await harness.execute('follow-up', { threadId: 'thread-1' });

Rejects with NoeticConfigError code NO_STEP_CONFIGURED if no agentGraph was provided in the constructor.

Session Accessors

Stream accessors are keyed by threadId (via SessionScope), are safe to call before the first execute(), and stay alive across every turn in the session.

Delivery Modes

Each message carries a DeliveryMode. Override per-call via options.deliveryMode, or set defaultDeliveryMode on the harness.

Steering a running turn

A between-rounds or next-item message is a steer: it joins the turn that is running instead of waiting for the next one. When no turn is running, it starts one like next-turn.

await harness.execute('Fix the failing test.');
// While the agent works:
await harness.execute('Use the staging database, not production.', {
  deliveryMode: 'next-item',
});

A model call offers queued steers at three points: when the stream finishes an output item (item_done), before each tool call (before_tool_call), and before each round (before_round). By default a next-item steer applies at item_done, and a between-rounds steer at before_round. Applied steers keep their content parts and role, and land in the history in the order the model saw them.

To change when a steer lands, give a context layer an onSteer hook. It can apply a steer sooner (before a tool call, skipping the rest of the round), hold it longer, rewrite it, or drop it. See Custom Layers.

Steers reach model calls on the session's own context. A spawn child takes them only with steers: 'forward'. With durability on, a turn resumed on a new harness still applies the steers it had been sent; see Durability.

HarnessResponse

interface HarnessResponse {
  readonly items: ReadonlyArray<Item>;
  readonly usage: { inputTokens: number; outputTokens: number; cachedTokens?: number };
  readonly cost?: number;
  readonly text: string;
  readonly lastLayerUsage?: LastLayerUsage;
}

StreamEvent

Events have a source discriminant: 'sdk' for OpenResponses SSE events, 'framework' for Noetic lifecycle events. Framework events use the harness config.name as prefix.

type StreamEvent = SdkStreamEvent | FrameworkStreamEvent;

Framework events emitted during execution:

callModel steps support an emit option to control step/tool framework event emission — set false to suppress or pass a filter function. Turn-level events are emitted by the session runner and are not gated by step.emit.

Stream Idle Timeout

Each provider call is guarded by a watchdog that aborts the round if no SSE event arrives for 120 seconds. The watchdog re-arms on every stream event, so slow-but-alive responses are never cut short. On timeout, {name}:model_call_stalled is emitted and the surrounding turn fails with turn_aborted { reason: "model stream idle timeout after <N>ms" }, so upstream providers that quietly drop a connection surface as errors rather than hangs.

run()

Low-level step execution. Use this when you need full control over context creation and step dispatch.

const ctx = harness.createContext({ state: { count: 0 } });
const result = await harness.run(myStep, input, ctx);

Before the first run of a root, run() checks what the root's connection handles need against each connection, so a root that needs an operation a connection lacks fails with CONNECTION_UNSATISFIED before anything executes.

When the harness is built with contextLayers + a storage adapter (environment.storage.adapter), run() runs the layers' init() hooks once per context (keyed by ctx.id) before executing — so prior layer state rehydrates from storage, recalls into the model context, and updates persist back. The same memory guarantee execute() gives, without the session/seedSessionHistory ceremony. This also makes parseAndRunWorkflow (which dispatches through run()) memory-aware. The init is idempotent: nested/repeated run() calls never re-init (re-init would clobber accumulated in-memory state).

createContext Options

Channel Methods

Context Methods

The agent harness coordinates context layer lifecycle hooks:

RecallLayerOutput

interface RecallLayerOutput {
  layerId: string;
  items: Item[];
  tokenCount: number;
}

Connection Methods

Durability Methods

When the harness is constructed with a durable environment.storage.adapter (and, typically, a durable environment.subprocess adapter), checkpoint and restore become real crash-recovery hooks — the checkpoint store defaults to one over the adapter. Without storage every call is a no-op — zero-config harnesses keep in-memory semantics.

Cancellation reaches inside the work in flight: blocked channel recv/send calls reject with cancelled, the current provider stream and ACP agent turn are cut short, and the next step boundary throws cancelled. Tokens and cost already spent stay charged to the context. Beyond those points it is cooperative — a runCode body that ignores ctx.aborted between awaits runs to its next step boundary.

Use ctx.abort(reason) for the same signal without layer teardown (it is synchronous and returns void); use cancel when the harness owns the layer lifecycle. Note that harness.abort(scope?) is a different operation: it cancels the current session turn by thread, not a specific context tree.

Graceful stop

A plain cancel cuts the provider stream at once, and the provider still bills the cut-off round. A host that is shutting down can instead give in-flight model calls time to finish:

process.on('SIGTERM', () => {
  void harness.cancel(ctx, { reason: 'SIGTERM', drainMs: 10e3 });
});

With drainMs, nothing new starts beneath ctx: steps, model rounds and tool calls wait instead. callModel rounds and decide calls already in flight get up to drainMs to finish, be charged to the context, and let their step record. Then the cancel runs as without drainMs, and the waiting work throws cancelled. Calls still running at the deadline are cut off the same way a plain cancel cuts them: with durable storage their records stay pending and are priced once the provider has billed them (see Paid Calls).

For a session turn, harness.abort({ threadId, drainMs }) does the same, and getStatus() reports { kind: 'draining', turnId, deadline } while it waits. For a run() execution, getStatus({ executionId: ctx.id }) reports { kind: 'draining', deadline }, then aborting until the run returns. A second cancel or abort during the drain cancels at once.

Both resolve with the drain's report once the cancel is done, so a host can account for the drained calls without listening for events:

const report = await harness.cancel(ctx, { drainMs: 10e3 });
for (const call of report?.finished ?? []) {
  budget.charge(call.cost ?? 0);
}

The report is { executionId, reason, durationMs, finished, cutOff }, the same data {name}:drain_finished carries. reason is settled when nothing was left, deadline, or stopped. A second stop during the drain resolves with the same report; a stop with no drain resolves undefined. Each call is { kind: 'model_round' | 'decide', model, executionId, round?, dispatchId?, generationId?, cost?, usage? }:

• A finished call the provider billed carries cost and usage exactly as charged to its context, even when its step ended cancelled and left no lastStepMeta. That includes a round rejected and asked again (listed under the same round as its retry) and a call that failed after it was billed. A round is finished once its response is in, even if its stream is still open at the deadline. A call the provider refused, or whose stream a steer cut before its response, has neither.
• A model round carries dispatchId when the harness keeps paid-call records, and generationId once its response headers arrived, so each cutOff call can be matched to its entry in pendingSpend().

{name}:drain_started lists the calls in flight when the drain begins.

The subprocess adapter is the companion surface: it persists handle manifests for every long-lived child so a restarted host can call harness.subprocess.listLive() + harness.restore(executionId) per live child to rejoin the execution. Full durable-execution model, storage layout, and IPC protocol live on the Durability page.

Tracing

AgentConfig

AgentConfig defines the configuration for an agent harness. It is generic over TParams, an arbitrary key-value record that steps and tools access via ctx.harness.config.params.

interface AgentConfig<TParams extends Record<string, unknown> = Record<string, unknown>> {
  name: string;
  storage?: StorageAdapter;
  hooks?: AgentHooks;
  params: TParams;
}

AgentHooks

interface AgentHooks {
  beforeStep?: (step: Step, ctx: Context) => Promise<void>;
  afterStep?: (step: Step, result: unknown, ctx: Context) => Promise<void>;
}

const harness = new AgentHarness({
  name: 'researcher',
  agentGraph: agent,
  params: { model: 'openai/gpt-4o' },
  hooks: {
    async beforeStep(step, ctx) {
      console.log(`Starting step ${step.id} (depth ${ctx.depth})`);
    },
    async afterStep(step, result, ctx) {
      console.log(`Step ${step.id} used ${ctx.tokens.total} tokens`);
    },
  },
});

Resuming a session from disk

harness.seedSessionHistory(threadId, items) populates a fresh session's accumulated items before the next execute() call. Pair it with append-only persistence (one Item per line in a JSONL file) to make a chat survive subprocess restarts:

import { harness } from './my-harness';
import { readPriorItems } from './my-store';

const threadId = 'task-T-abc123';
harness.seedSessionHistory(threadId, await readPriorItems(threadId));
await harness.execute('continue from here', { threadId });

The Noetic CLI uses this pattern for per-task agent runners — see the "Per-Task IPC for Live Chat" section of specs/08-runtime.md for the end-to-end design.

Subprocess Adapter

The harness always holds a SubprocessAdapter. Every runCode, every spawn, and every harness.detachedSpawn routes through harness.subprocess.spawn(...) — in-process vs out-of-process is a property of the adapter, not the step.

import { AgentHarness } from '@noetic-tools/core';
import { createFileStorage, createLocalSubprocessAdapter } from '@noetic-tools/platform-node';

const subprocess = createLocalSubprocessAdapter({
  storage: createFileStorage({ root: `${process.env.HOME}/.noetic/subprocess` }),
});

const harness = new AgentHarness({
  name: 'durable-agent',
  agentGraph: agent,
  params: {},
  environment: { subprocess },                // real OS subprocess with durable manifests
});

Per-step and per-call overrides let you mix execution backends. A step with subprocess: createInMemorySubprocessAdapter() stays in-process even when the harness default is the local adapter; a detachedSpawn(step, input, ctx, { subprocess: otherAdapter }) overrides both. Resolution order is detachedSpawn-overrides.subprocess ?? step.subprocess ?? harness.subprocess.

Related Pages

• Context & Event Log -- the context object the agent harness creates.
• Channels -- channel send/recv methods on the agent harness.
• Context Layers -- context layers managed by the agent harness.
• Observability -- tracing spans created by createSpan.
• Durability -- checkpoint/restore, adapter durability, and the host-restart flow.