diff --git a/src/horde.js b/src/horde.js index d7b4579..7b8c924 100644 --- a/src/horde.js +++ b/src/horde.js @@ -2,13 +2,53 @@ // Modeled on the browser client in reference/backend.js: nonce-tracked actions, // action listeners, orphan handlers for the auth response, auto-reconnect. +// Mirrors the browser Stream: accumulates live content and exposes it as a +// string via getContent() once the server-side stream reaches eof. +export class Stream { + constructor(id) { + this.id = id; + this.partialContent = ""; + this.eof = false; + this.listeners = { data: [], end: [] }; + this.contentPromise = new Promise((resolve) => { + this.on('end', () => resolve(this.partialContent)); + }); + } + + feed(chunk) { + if (this.eof) return; + this.partialContent += chunk; + this.emit('data', chunk); + } + + close() { + if (this.eof) return; + this.eof = true; + this.emit('end', this.partialContent); + } + + getContent() { + return this.contentPromise; + } + + emit(event, ...args) { + for (const fn of this.listeners[event]) fn(...args); + } + + on(event, callback) { + this.listeners[event].push(callback); + if (event === 'data' && this.partialContent) callback(this.partialContent); + if (event === 'end' && this.eof) callback(this.partialContent); + } +} + export default class Horde { constructor({ host, token = null }) { this.host = host; this.token = token; this.socket = null; this.nextNonce = 0; - this.pending = new Map(); // nonce -> { resolve, reject } + this.pending = new Map(); // nonce -> single { resolve, reject } or streaming { run, deferred } this.actionListeners = new Map(); // action -> [callback(packet)] this.orphanHandlers = []; // callbacks for nonce-less messages (auth replies) this.connectListeners = []; @@ -99,7 +139,8 @@ export default class Horde { } clearPending() { - for (const { reject } of this.pending.values()) { + for (const entry of this.pending.values()) { + const reject = entry.streaming ? entry.deferred.reject : entry.reject; reject(new Error('Connection lost')); } this.pending.clear(); @@ -129,16 +170,109 @@ export default class Horde { this.socket.send(JSON.stringify(value)); } - sendAction(action, body = {}) { + // Sends an action. Without a callback this is a single-response request. + // With a callback it is a multiple-result action (like the browser client's + // sendAction): the callback runs once per packet carrying this nonce, and the + // returned promise settles only when the callback calls deferred.resolve() + // (or deferred.reject()). This is the shape used by read-stream and + // prompt-agent, which both emit many packets per request. + sendAction(action, body = {}, callback = null) { const nonce = this.nextNonce++; + + if (callback) { + let resolveInner, rejectInner; + const promise = new Promise((res, rej) => { resolveInner = res; rejectInner = rej; }); + const deferred = { promise, resolve: resolveInner, reject: rejectInner }; + this.pending.set(nonce, { + streaming: true, + run: (packet) => callback(packet, deferred), + deferred, + }); + // Ensure cleanup runs whichever way the action settles, and swallow the + // rejection on the derived chain so it never surfaces as unhandled. + promise.catch(() => {}).finally(() => this.pending.delete(nonce)); + this.send({ action, nonce, body }); + return promise; + } + const promise = new Promise((resolve, reject) => { - this.pending.set(nonce, { resolve, reject }); + this.pending.set(nonce, { streaming: false, resolve, reject }); }); promise.catch(() => {}); // avoid unhandled rejections when nobody awaits this.send({ action, nonce, body }); return promise; } + // Reads a server-side stream into a Stream object. Content arrives as a + // sequence of read-stream packets sharing one nonce; eof closes the stream. + readStream(streamId) { + const stream = new Stream(streamId); + this.sendAction("read-stream", { streamId }, (packet, deferred) => { + if (packet.eof) { + stream.close(); + deferred.resolve(stream.partialContent); + } else if (packet.data != null) { + stream.feed(packet.data); + } + }).catch(() => stream.close()); + return stream; + } + + // Runs a prompt and yields every generated inference message in arrival + // order. prompt-agent is a multiple-result action; assistant/reasoning + // content is an open stream ({streamId}) which is read to completion before + // the message is yielded, so callers always see plain string content. + async *prompt(agentId, body) { + const queue = []; + let notify = null; + let finished = false; + let failure = null; + + const wake = () => { + if (notify) { const resolve = notify; notify = null; resolve(); } + }; + + this.sendAction( + "prompt-agent", + { agentId, ...body }, + (packet, deferred) => { + if (packet.eof) { + finished = true; + deferred.resolve(packet); + wake(); + } else { + queue.push(packet.data); + wake(); + } + } + ).catch((err) => { + failure = err; + finished = true; + wake(); + }); + + while (true) { + if (failure) throw failure; + if (queue.length) { + yield await this.resolveStreams(queue.shift()); + continue; + } + if (finished) return; + await new Promise((resolve) => { notify = resolve; }); + } + } + + // Resolves any open Stream in a message's content to its final text. + async resolveStreams(msg) { + const content = msg && msg.content; + if (content && typeof content === 'object' && content.streamId != null) { + const stream = this.readStream(content.streamId); + const full = await stream.getContent(); + return { ...msg, content: full }; + } + return msg; + } + handleMessage(raw) { let packet; try { @@ -149,14 +283,24 @@ export default class Horde { } if (packet.nonce != null) { - const pending = this.pending.get(packet.nonce); - if (!pending) { + const entry = this.pending.get(packet.nonce); + if (!entry) { console.warn(`Orphan response for nonce ${packet.nonce}`); return; } + + if (entry.streaming) { + // Multiple-result action: keep the entry alive and hand every packet + // (including the terminal eof) to the callback. The callback settles + // the promise once the action is complete. + if (packet.error) entry.deferred.reject(new Error(packet.error)); + else entry.run(packet); + return; + } + this.pending.delete(packet.nonce); - if (packet.error) pending.reject(new Error(packet.error)); - else pending.resolve(packet); + if (packet.error) entry.reject(new Error(packet.error)); + else entry.resolve(packet); return; } diff --git a/src/main.js b/src/main.js index 0b28ddd..56aeded 100755 --- a/src/main.js +++ b/src/main.js @@ -21,6 +21,7 @@ import * as fs from 'fs'; import * as path from 'path'; import { registerHooks } from 'node:module'; import { pathToFileURL, fileURLToPath } from 'url'; +import { createInterface } from 'node:readline/promises'; import Horde from './horde.js'; import run from './run.js'; @@ -45,6 +46,9 @@ if (process.env.HORDE_TOKEN) DEFAULT_CONFIG.token = process.env.HORDE_TOKEN; * 2. Context / API * ------------------------------------------------------------------ */ +// TODO: prompt and confirm don't belong here, it's not runtime related. +// These should be simple library functions in userland or an explicit utils module. + /** * Global execution context exposed to function modules as `Gral`. */ @@ -52,6 +56,30 @@ const Gral = { path: process.cwd(), // The function itself is defined in run.js run, + + /** + * Ask the user a free-form question and resolve with their answer. + * + * @param {string} question Text shown to the user. + * @returns {Promise} + */ + prompt: async function (question) { + const rl = createInterface({ input: process.stdin, output: process.stdout }); + let answer = await rl.question(`${question} `); + rl.close(); + return answer; + }, + + /** + * Ask the user a yes/no question and resolve with their choice. + * + * @param {string} question Text shown to the user. + * @returns {Promise} True for yes/`y`, false otherwise. + */ + confirm: async function (question) { + const answer = await Gral.prompt(`${question} [y/N] `); + return /^y(es)?$/i.test(answer.trim()); + }, }; global.Gral = Gral; diff --git a/src/run.js b/src/run.js index f84aea2..f1c3bf5 100644 --- a/src/run.js +++ b/src/run.js @@ -91,18 +91,37 @@ function extractToolCalls(res) { * @param {?boolean|string[]} [options.write] Writable files/dirs (null = all). * @param {object[]} [options.tools] Extra tools to pass the agent. * @param {?boolean|string[]} [options.defaultTools] Default tool selection. + * @param {?Function} [options.verify] Predicate run after each iteration; a truthy + * return stops the run and resolves to null. + * @param {?Function} [options.onMaxIter] Called once when `maxIter` is exhausted; its + * return value (or null) becomes the result. + * @param {number} [options.maxIter] Maximum tool-call iterations. * @returns {Promise} The agent's final response. + * + * Both callbacks may be synchronous or return a promise and receive a single + * context object: + * + * { step, maxIter, history } + * + * `step` is the number of iterations completed (so `step === maxIter` when + * `onMaxIter` fires). Without `onMaxIter`, exhausting `maxIter` + * throws. */ async function run({ + // prompt agentId, instruction, input, messages, - read = null, - write = null, tools = null, defaultTools = true, + // bounds + read = null, + write = null, + maxIter = MAX_ITERATIONS, + // callbacks verify = null, + onMaxIter = null, } = {}) { const Gral = global.Gral; const root = Gral && Gral.path ? Gral.path : process.cwd(); @@ -118,36 +137,48 @@ async function run({ defaultTools, }); const specs = bundle.map(({ name, description, parameters }) => ({ - type: 'function', - function: { - name, - description, - parameters, - } + name, + description, + parameters, })); const execute = makeExecutor(bundle); const history = Array.isArray(messages) ? messages.slice() : []; - for (let step = 0; step < MAX_ITERATIONS; step++) { - const res = await Gral.horde.sendAction('send-agent-prompt', { - agentId, + for (let step = 0; step < maxIter; step++) { + const messages = []; + const gen = Gral.horde.prompt(agentId, { instruction, input, options: { messages: history, tools: specs, - }, + } }); - const calls = res.toolCalls || []; - if (calls.length === 0) return res.response; + for await (const msg of gen) { + messages.push(msg); + } - if (res.response) { - console.log(res.response); + // Equivalent to the old `res.response` (already a plain string — + // any open {streamId} content is read to eof and resolved for you): + const response = messages + .filter(m => m.role === 'assistant') + .map(m => m.content) + .join(''); + + // Equivalent to the old `res.toolCalls || []`: + const calls = messages + .filter(m => m.role === 'tool') + .map(m => m.content); // ToolCall dicts: { id, tool_name, arguments, result } + + if (calls.length === 0) return response; + + if (response) { + console.log(response); history.push({ role: 'assistant', - content: res.response, + content: response, }); } @@ -168,7 +199,12 @@ async function run({ } } - throw new Error(`run: exceeded ${MAX_ITERATIONS} tool-call iterations`); + if (onMaxIter) { + const result = await onMaxIter({ step: maxIter, maxIter, history }); + return result === undefined ? null : result; + } + + throw new Error(`run: exceeded ${maxIter} tool-call iterations`); } export default run;