Polyweave

Repositories

1

Projects Packages

@polyweave/gatto-agent-harness (1.0.5)

Published 2026-07-11 15:25:33 +00:00 by Dvorak

Installation

Set up this registry in your project .npmrc file:

@polyweave:registry=https://hub.kl1.tenere.ai/api/packages/Polyweave/npm/

To install the package using npm, run the following command:

npm install @polyweave/gatto-agent-harness@1.0.5

or add it to the package.json file:

"@polyweave/gatto-agent-harness": "1.0.5"

For more information on the npm registry, see the documentation.

About this package

@polyweave/gatto-agent-harness

Task-level agent harness for Gatto — push-stream Duplex agent execution loop with tool registry. Integrates with @polyweave/jj-provider for per-BT-node workspace isolation.

Published as @polyweave/gatto-agent-harness.

Synopsis

import { createAgentHarness } from '@polyweave/gatto-agent-harness';

const harness = createAgentHarness({
  maxTurns: 50,
  tools: [myTool1, myTool2],
  jjProvider: myJjProvider,
  durability: { enabled: true, wrapperId: 'agent-1' }
});

harness.source.pipe(myOutputSink);

harness.sink.write({ type: 'PARAM', key: 'workspace', value: '/path/to/workspace' });
harness.sink.write({ type: 'PARAM', key: 'task', value: 'Build the component' });
harness.sink.write({ type: 'START' });

Install

npm install @polyweave/gatto-agent-harness

Why

LLM agent execution requires a structured harness for workspace management, turn-based interaction, tool routing, and state persistence. The gatto-agent-harness provides a task-level agent execution loop built on Polyweave push-streams and observable state from @polyweave/core. An agent is configured via PARAM frames (workspace, task), started with START, receives TEXT frames as user input, routes TOOL_RESULT callbacks, and maintains observable state including history, status, and turn count. Tools are managed through a push-stream-capable tool registry with parameter validation. Frame constructors use the GattoFrameType enum for the Gatto-specific protocol.

API

createAgentHarness(options)

Creates a Gatto agent harness as a push-stream duplex.

createAgentHarness(options?: Object): { source, sink, pipe, write, end, getState }

Name	Type	Required	Default	Description
options.maxTurns	number	No	100	Maximum agent turns before forced end
options.tools	Array	No	[]	Array of tool adapter objects to register
options.jjProvider	any	No	null	JJ provider for workspace isolation (stored, not directly invoked by harness)
options.durability	Object	No	{}	Durability config: { enabled: boolean, wrapperId: string, wrapperType?: string }

Returns: An object with:

• source — push-stream source (emits output frames; backpressure-aware with buffering)
• sink — push-stream sink (receives control/data frames; see frame protocol below)
• pipe(dest) — delegates to source.pipe(dest)
• write(data) — delegates to sink.write(data)
• end(err) — delegates to sink.end(err)
• getState() — returns current observable state snapshot via observableState.getAll()

State shape (from getState()):

{
  workspace: string|null,
  task: string|null,
  history: Array<{ role, content, timestamp, callId?, result?, error? }>,
  status: 'idle'|'thinking'|'executing',
  currentTurn: number,
  initialized: boolean,
  nodeId: string|null,
  sessionId: string|null
}

Frame Protocol

Input frames (sink receives):

Frame Type	Description
PARAM	Sets configuration by key: 'workspace', 'task', 'nodeId', 'sessionId' (stored in observable state); 'maxTurns' (updates opts.maxTurns); 'tool' (registers a tool in the registry). If also META type, forwarded to observableState sink.
START	Initializes the harness. Requires workspace and task to be set via prior PARAM frames — emits ERROR + ends sink if missing. Sets initialized = true, resets turn count, pushes system message to history, emits START + PROGRESS frames.
TEXT / 'TEXT'	User text input. Accumulates to history (role: 'user'), increments currentTurn. If turn count >= maxTurns, emits PROGRESS('max-turns-reached') and ends sink. Otherwise, passes frame through to source. Requires harness to be initialized (ignored silently if not).
DATA / 'DATA'	Routes by frame.data.type: 'tool-result' → _handleToolResult, 'llm-response' → _handleLLMResponse. Others emit ERROR. Requires initialization (emits ERROR otherwise).
TOOL_RESULT / 'TOOL_RESULT'	Shortcut for tool result. Extracts from frame.content or frame.data. Calls _handleToolResult.
END / 'END'	Passes through an END frame to source with phase: 'task-complete'.
ERROR / 'ERROR'	Passes through to source.
META	Forwarded to observableState sink for state subscriptions.

Output frames (source emits):

Frame Type	Description
START	Emitted after initialization. Metadata: { phase: 'task-execution', workspace, task, nodeId, sessionId }
PROGRESS	Emitted for: initialization ({ operation: 'initialized', workspace, task, maxTurns }), tool completion ({ operation: 'tool-completed', callId, status }), max turns reached ({ operation: 'max-turns-reached', currentTurn, maxTurns })
TEXT	Passed through from input after history accumulation
TOOL_CALL	Emitted during LLM response processing (via _handleLLMResponse). Each tool call produces a TOOL_CALL frame and a corresponding tool registration call. Format: { type: 'DATA', data: { type: 'tool-call', id: callId, tool: name, arguments: {...} } }
THINKING	Emitted during LLM response if data.thinking present
END	Emitted on sink.end(): `{ reason: 'completed'
ERROR	Emitted for: missing workspace/task at START, harness not initialized, unknown frame type, unknown data type, catch-all errors in sink.write

Internal Flow

1. Configuration phase: Send PARAM frames to set workspace, task, nodeId, sessionId, maxTurns, and register tools.
2. Initialization: Send START. The harness validates workspace/task exist, pushes a system message to history, emits START and PROGRESS frames.
3. Interaction loop: Send TEXT frames as user input. The harness accumulates history, increments turns, and passes through. Send DATA frames with type: 'llm-response' to process LLM responses (extracts thinking text, assistant content, and tool calls).
4. Tool execution: LLM responses with toolCalls array trigger TOOL_CALL frames to source and tool execution via the tool registry. Results come back as TOOL_RESULT/DATA frames.
5. Termination: Send END or reach max turns. END emits final frames and marks source ended.

Error Conditions

The harness emits ERROR frames (never throws) for:

• Missing workspace (START handler): 'Workspace not configured. Send PARAM(workspace, path) first.' — also calls sink.end(new Error('Missing workspace'))
• Missing task (START handler): 'Task not configured. Send PARAM(task, description) first.' — also calls sink.end(new Error('Missing task'))
• Not initialized (DATA/TEXT handlers): 'Harness not initialized. Send START frame first.'
• Unknown frame type (sink default): 'Unknown frame type: {frame.type}'
• Unknown data type (DATA handler): 'Unknown data type: {data.type}'
• DATA frame missing type: 'DATA frame missing type'
• Catch-all (sink error handler): wraps thrown errors with { operation: 'agent-harness.sink.write', frame }
• Sink.end with error: wraps in ERROR frame with { phase: 'shutdown' }

---

createToolRegistry(initialTools)

Creates a push-stream-capable tool registry with parameter validation and async execution.

createToolRegistry(initialTools?: Array): { source, sink, pipe, write, end, paused, ended, register, has, list, get, getToolOpenAIFormat }

Name	Type	Required	Default	Description
initialTools	Array	No	[]	Array of tool objects to register initially

Tool object shape:

{
  id: string,           // Tool identifier
  execute: (args, context, callback) => void,  // Async execution callback
  description?: string,  // For OpenAI format
  parameters?: { type: 'object', properties: Object, required?: string[] }  // For validation + OpenAI format
}

Returns: An object with:

• source / sink — push-stream sources/sinks (backpressure-aware with buffering)
• pipe(dest) — delegates to source.pipe(dest)
• write(data) — delegates to sink.write(data)
• end(err) — delegates to sink.end(err)
• paused — getter, source paused state
• ended — getter, source ended state
• register(tool) — registers a tool. Silently ignores tools without id or execute function.
• has(toolId) — returns boolean
• list() — returns array of registered tool IDs
• get(toolId) — returns tool object or undefined
• getToolOpenAIFormat() — returns tools in OpenAI function-calling format

Frame Protocol

Input frames (sink receives):

Frame Type	Description
DATA with data.type === 'tool-call'	Executes a tool. Validates required parameters if tool.parameters.required exists. Calls tool.execute(args, context, callback). Emits TOOL_RESULT on completion or error DATA frame for missing tool / missing required fields.
PARAM with key === 'register-tool'	Registers the value as a new tool

Error conditions: Emits ERROR frames (DATA type with error) for:

• Tool not found: { error: "Tool '{toolId}' not found", status: 'error' }
• Missing required field: { error: "Missing required field: {field}", status: 'error' }
• Catch-all in sink.write, emits ERROR frame

---

GattoFrameType

Enum extending FrameType from @polyweave/core with Gatto-specific types:

{
  ...FrameType,
  DATA: 'DATA',
  THINKING: 'THINKING',
  TASK_START: 'TASK_START',
  TASK_COMPLETE: 'TASK_COMPLETE',
  AGENT_INITIALIZED: 'AGENT_INITIALIZED',
  MAX_TURNS_REACHED: 'MAX_TURNS_REACHED'
}

---

Frame Constructors

All constructors accept an optional opts object with id, timestamp (default Date.now()), and type-specific metadata.

createStartFrame(metadata?)

Returns { type: FrameType.START, metadata, timestamp: Date.now() }

createEndFrame(metadata?)

Returns { type: FrameType.END, metadata, timestamp: Date.now() }

createParamFrame(key, value, metadata?)

Returns { type: FrameType.PARAM, key, value, metadata, timestamp: Date.now() }

createDataFrame(data, metadata?)

Returns { type: GattoFrameType.DATA, data, metadata, timestamp: Date.now() }

createProgressFrame(progress, metadata?)

Returns { type: FrameType.PROGRESS, progress, metadata, timestamp: Date.now() }

createErrorFrame(error, metadata?)

Returns { type: FrameType.ERROR, error: string, stack?: string, metadata, timestamp: Date.now() }. If error is an Error instance, extracts .message and .stack.

createTextFrame(text, metadata?)

Returns { type: FrameType.TEXT, content: { text }, metadata, timestamp: Date.now() }

createToolCallFrame(id, tool, args, metadata?)

Returns { type: GattoFrameType.DATA, data: { type: 'tool-call', id, tool, arguments: args }, metadata, timestamp: Date.now() }

createToolResultFrame(callId, result, error?, metadata?)

Returns { type: GattoFrameType.DATA, data: { type: 'tool-result', callId, result, error, status: 'success'|'error' }, metadata, timestamp: Date.now() }

createThinkingFrame(content, metadata?)

Returns { type: GattoFrameType.DATA, data: { type: 'thinking', content }, metadata, timestamp: Date.now() }

Examples

// Subscribe to state changes via getState polling
setInterval(() => {
  const state = harness.getState();
  console.log('Status:', state.status, 'Turn:', state.currentTurn);
}, 1000);

// Full agent lifecycle
import { createAgentHarness } from '@polyweave/gatto-agent-harness';

const harness = createAgentHarness({
  maxTurns: 25,
  tools: [shellTool, fileTool],
  durability: { enabled: true, wrapperId: 'task-42' }
});

harness.source.pipe(createFrameSink(frame => {
  if (frame.type === GattoFrameType.DATA && frame.data?.type === 'tool-call') {
    console.log('Tool called:', frame.data.tool);
  }
}));

harness.sink.write({ type: 'PARAM', key: 'workspace', value: wsRoot });
harness.sink.write({ type: 'PARAM', key: 'task', value: 'Fix the build errors' });

harness.sink.write({ type: 'START' }); // initializes and begins

// Simulate input text
harness.sink.write({ type: 'TEXT', content: { text: 'Please fix the TypeScript errors' } });

// Simulate LLM response
harness.sink.write({
  type: 'DATA',
  data: {
    type: 'llm-response',
    content: 'I will fix the errors now.',
    thinking: 'Analyzing the error log...',
    toolCalls: [{ tool: 'shell', arguments: { command: 'npx tsc --noEmit' } }]
  }
});

// Programmatically register a tool
harness.sink.write({ type: 'PARAM', key: 'tool', value: {
  id: 'echo',
  description: 'Echoes back input',
  parameters: { type: 'object', properties: { text: { type: 'string' } }, required: ['text'] },
  execute: (args, ctx, cb) => cb(null, { echoed: args.text })
}});

Tests

npm test

Runs: node --test test/*.test.js

Covers: harness initialization and PARAM configuration, START flow (workspace+task validation), TEXT frame history accumulation and turn counting, maxTurns enforcement with forced end, DATA routing (tool-result, llm-response, unknown types), LLM response processing (thinking, assistant content, tool calls), tool registry: registration, lookup, parameter validation, async execution, OpenAI format generation, frame constructors (all 10), sink end/shutdown with error wrapping, and property-based testing via js-rigor.

Requirements

• Node.js >= 22.0.0
• ESM only ("type": "module")
• Peer dependencies: @polyweave/core
• Dependencies: @polyweave/jj-provider, @polyweave/tools

Caveats

• The harness is driven by push-stream frames rather than autonomous execution. External scheduling is required.
• JJ provider integration is stored but not directly invoked by the harness — integration is handled externally.
• Durability relies on createObservableState from @polyweave/core. State is persisted only if the observable state wrapper supports it (via options.durability.enabled).
• getState() returns a snapshot of all observable state properties. There is no subscribe() method exposed on the harness itself — use state directly via the observable pattern if needed.

Dependencies

Dependencies

ID	Version
@polyweave/harness-base	*

Development Dependencies

ID	Version
@rigor/core	*

Peer Dependencies

ID	Version
@polyweave/core	*

Keywords

polyweave gatto agent harness llm push-stream

Details

npm

2026-07-11 15:25:33 +00:00

5

John Dvorak

SEE LICENSE IN LICENSE

latest

8.5 KiB

Assets (1)

gatto-agent-harness-1.0.5.tgz 8.5 KiB

Versions (3) View all

1.0.5 2026-07-11

1.0.2 2026-07-10

1.0.1 2026-07-10