diff --git a/src/hub/agent.ts b/src/hub/agent.ts deleted file mode 100644 index d25b83c55..000000000 --- a/src/hub/agent.ts +++ /dev/null @@ -1,87 +0,0 @@ -import { v7 as uuidv7 } from "uuid"; -import { Agent as CoreAgent } from "../agent/runner.js"; -import { Channel } from "./channel.js"; -import type { Message } from "./types.js"; - -/** - * Agent — uses pi-agent-core for real inference. - * write() triggers a model run, read() outputs streaming results. - */ -export class Agent { - readonly id: string; - private readonly channel = new Channel(); - private _closed = false; - private readonly agent: CoreAgent; - private queue: Promise = Promise.resolve(); - - constructor(id?: string) { - this.id = id ?? uuidv7(); - this.agent = new CoreAgent({ - logger: { - stdout: this.createChannelStream("[assistant] "), - stderr: this.createChannelStream("[tool] "), - }, - sessionId: this.id, - }); - } - - get closed(): boolean { - return this._closed; - } - - /** Write message to agent (non-blocking, serialized queue) */ - write(content: string): void { - if (this._closed) { - throw new Error("Agent is closed"); - } - - this.queue = this.queue - .then(async () => { - const result = await this.agent.run(content); - if (result.error) { - this.channel.send({ - id: uuidv7(), - content: `[error] ${result.error}`, - }); - } - }) - .catch((err) => { - const message = err instanceof Error ? err.message : String(err); - this.channel.send({ id: uuidv7(), content: `[error] ${message}` }); - }); - } - - /** Continuously read message stream */ - read(): AsyncIterable { - return this.channel; - } - - /** Close agent, stop all reads */ - close(): void { - if (this._closed) return; - this._closed = true; - this.channel.close(); - } - - private createChannelStream(prefix: string): NodeJS.WritableStream { - let buffer = ""; - return { - write: (chunk: any) => { - if (this._closed) return false; - const text = - typeof chunk === "string" - ? chunk - : chunk?.toString?.() ?? String(chunk); - if (!text) return true; - buffer += text; - const parts = buffer.split("\n"); - buffer = parts.pop() ?? ""; - for (const part of parts) { - if (part.length === 0) continue; - this.channel.send({ id: uuidv7(), content: `${prefix}${part}` }); - } - return true; - }, - } as NodeJS.WritableStream; - } -} diff --git a/src/hub/channel.ts b/src/hub/channel.ts deleted file mode 100644 index 5073c3cdc..000000000 --- a/src/hub/channel.ts +++ /dev/null @@ -1,64 +0,0 @@ -/** - * Go channel style async iterable queue. - * Supports multiple writers, single reader, iteration ends after close. - */ -export class Channel implements AsyncIterable { - private buffer: T[] = []; - private closed = false; - - private readers: Array<{ - resolve: (result: IteratorResult) => void; - }> = []; - - get isClosed(): boolean { - return this.closed; - } - - get size(): number { - return this.buffer.length; - } - - /** Send value to channel. Returns false when channel is closed. */ - send(value: T): boolean { - if (this.closed) return false; - - const reader = this.readers.shift(); - if (reader) { - reader.resolve({ value, done: false }); - return true; - } - - this.buffer.push(value); - return true; - } - - /** Close channel, wake up all waiting readers. */ - close(): void { - if (this.closed) return; - this.closed = true; - - for (const reader of this.readers) { - reader.resolve({ value: undefined as T, done: true }); - } - this.readers = []; - } - - [Symbol.asyncIterator](): AsyncIterator { - return { - next: (): Promise> => { - if (this.buffer.length > 0) { - const value = this.buffer.shift()!; - return Promise.resolve({ value, done: false }); - } - - if (this.closed) { - return Promise.resolve({ value: undefined as T, done: true }); - } - - return new Promise>((resolve) => { - this.readers.push({ resolve }); - }); - }, - }; - } -} diff --git a/src/hub/hub.ts b/src/hub/hub.ts index ae6c5cfec..271955837 100644 --- a/src/hub/hub.ts +++ b/src/hub/hub.ts @@ -1,11 +1,11 @@ import type { HubOptions } from "./types.js"; import type { ConnectionState } from "../shared/gateway-sdk/types.js"; -import { Agent } from "./agent.js"; +import { AsyncAgent } from "../agent/async-agent.js"; import { getDeviceId } from "./device.js"; import { GatewayClient } from "../shared/gateway-sdk/client.js"; export class Hub { - private readonly agents = new Map(); + private readonly agents = new Map(); private readonly agentSenders = new Map(); private client: GatewayClient; url: string; @@ -82,7 +82,7 @@ export class Hub { } /** Create new Agent, or rebuild with existing ID */ - createAgent(id?: string): Agent { + createAgent(id?: string): AsyncAgent { if (id) { const existing = this.agents.get(id); if (existing && !existing.closed) { @@ -90,31 +90,31 @@ export class Hub { } } - const agent = new Agent(id); - this.agents.set(agent.id, agent); + const agent = new AsyncAgent({ sessionId: id }); + this.agents.set(agent.sessionId, agent); // Internally consume messages produced by agent void this.consumeAgent(agent); - console.log(`Agent created: ${agent.id}`); + console.log(`Agent created: ${agent.sessionId}`); return agent; } /** Internally read agent output and send via Gateway */ - private async consumeAgent(agent: Agent): Promise { + private async consumeAgent(agent: AsyncAgent): Promise { for await (const msg of agent.read()) { - console.log(`[${agent.id}] ${msg.content}`); - const targetDeviceId = this.agentSenders.get(agent.id); + console.log(`[${agent.sessionId}] ${msg.content}`); + const targetDeviceId = this.agentSenders.get(agent.sessionId); if (targetDeviceId) { this.client.send(targetDeviceId, "message", { - agentId: agent.id, + agentId: agent.sessionId, content: msg.content, }); } } } - getAgent(id: string): Agent | undefined { + getAgent(id: string): AsyncAgent | undefined { return this.agents.get(id); } diff --git a/src/hub/index.ts b/src/hub/index.ts index 54adb7228..4ea50bc63 100644 --- a/src/hub/index.ts +++ b/src/hub/index.ts @@ -1,5 +1,3 @@ -export { Channel } from "./channel.js"; -export { Agent } from "./agent.js"; export { Hub } from "./hub.js"; export { getDeviceId } from "./device.js"; -export type { Message, HubOptions } from "./types.js"; +export type { HubOptions } from "./types.js";