1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
|
// minime.js — use mini-me from Node.js, as a library, by driving `mm -acp`
// over ACP with the official ACP TypeScript SDK, @agentclientprotocol/sdk
// (ticket 188). Same API as ../../nodejs/lib/minime.js, which speaks the
// protocol by hand with no dependency.
//
// import { Agent } from "./lib/minime.js";
//
// const agent = await Agent.start({ config: "agent.yaml" });
// const answer = await agent.complete("Explain this repository"); // not streamed
// for await (const ev of agent.stream("And the tests?")) { // streamed
// if (ev.kind === "text") process.stdout.write(ev.text);
// }
// await agent.close();
//
// One Agent = one `mm -acp` process = one ACP session: the conversation goes
// on from one call to the next, as in an editor. reset() starts a new one.
import { spawn } from "node:child_process";
import { createInterface } from "node:readline";
import { Readable, Writable } from "node:stream";
import path from "node:path";
import { client, ndJsonStream, PROTOCOL_VERSION, RequestError } from "@agentclientprotocol/sdk";
/**
* The events a turn produces, reduced from ACP's `session/update`
* notifications (which carry more detail, meant for editors).
*
* @typedef {{ kind: "text", text: string }} TextEvent a chunk of the answer
* @typedef {{ kind: "thinking", text: string }} ThinkingEvent a chunk of a thinking model's reasoning
* @typedef {{ kind: "tool_call", id: string, title: string, toolKind: string|undefined, input: unknown }} ToolCallEvent
* a tool call announced, before it runs; also what onPermission receives
* @typedef {{ kind: "tool_result", id: string, status: "completed"|"failed", output: string }} ToolResultEvent
* @typedef {{ kind: "done", stopReason: string, usage: object|undefined, text: string }} DoneEvent
* the end of the turn, with the whole answer
* @typedef {TextEvent|ThinkingEvent|ToolCallEvent|ToolResultEvent|DoneEvent} Event
*/
/**
* @typedef {object} Options
* @property {string} config the agent file (agent.yaml); made absolute. Required.
* @property {string} [cwd] the session's working directory (default: process.cwd())
* @property {string} [mm] the mm binary (default: $MM_BIN, then "mm" on the PATH)
* @property {(call: Omit<ToolCallEvent, "kind">) => boolean|string|Promise<boolean|string>} [onPermission]
* asked when the agent wants to run a tool that needs a confirmation (`yolo: false`).
* true = allow once, false = reject once, or an ACP optionId ("allow_always", "never"…).
* Without it, every such call is REJECTED: nothing runs unasked.
* @property {(line: string) => void} [onStderr] mm's stderr lines (banner, warnings)
*/
export class Agent {
/**
* Starts `mm -acp` on an agent file, then opens a session (`initialize`, `session/new`).
*
* @param {Options} options
* @returns {Promise<Agent>}
*
* @example
* const agent = await Agent.start({ config: "agent.yaml", onPermission: (call) => call.toolKind === "read" });
*/
static async start(options) {
const agent = new Agent(options);
try {
await agent.#open();
} catch (err) {
const failure = await agent.#failure(err);
await agent.close();
throw failure;
}
return agent;
}
#child;
#connection;
#session = null; // the SDK's ActiveSession of the current conversation
#cwd;
#onPermission;
#stderrTail = "";
#exited = null; // { code, signal } once mm is gone
#busy = false;
#cancelled = false; // set once session/cancel is sent for the current turn
/** @param {Options} options */
constructor({ config, cwd = process.cwd(), mm = process.env.MM_BIN || "mm", onPermission, onStderr } = /** @type {any} */ ({})) {
if (!config) throw new Error("minime: `config` (the agent file) is required");
this.#cwd = path.resolve(cwd);
this.#onPermission = onPermission;
this.#child = spawn(mm, ["-acp", "-config", path.resolve(config)], { cwd: this.#cwd, stdio: ["pipe", "pipe", "pipe"] });
this.#child.on("error", (err) => this.#gone(new Error(`minime: cannot start ${mm}: ${err.message}`)));
this.#child.on("exit", (code, signal) => {
this.#exited = { code, signal };
this.#gone(new Error(`minime: mm exited (${signal ?? `exit status ${code}`})`));
});
createInterface({ input: this.#child.stderr }).on("line", (line) => {
if (line.trim()) this.#stderrTail = line;
onStderr?.(line);
});
// The client side of ACP: mm's permission requests. The SDK answers any
// other request from mm (fs/*, terminal/*, elicitation) with "method not
// found" — we advertise none of them.
const app = client({ name: "minime-sdk" }).onRequest("session/request_permission", ({ params }) =>
this.#answerPermission(params),
);
this.#connection = app.connect(ndJsonStream(Writable.toWeb(this.#child.stdin), Readable.toWeb(this.#child.stdout)));
}
/** The ACP session id of the current conversation. */
get sessionId() {
return this.#session?.sessionId ?? null;
}
/**
* Runs one turn and returns the whole answer once it is over.
*
* @param {string} prompt
* @param {{ signal?: AbortSignal, onEvent?: (event: Event) => void }} [options]
* aborting `signal` sends `session/cancel`; the turn then ends with stopReason
* "cancelled" and the text so far. `onEvent` sees every event as it comes.
* @returns {Promise<{ text: string, stopReason: string, usage: object|undefined }>}
*
* @example
* const { text } = await agent.complete("What is in this folder?");
*/
async complete(prompt, { signal, onEvent } = {}) {
let last;
for await (const ev of this.stream(prompt, { signal })) {
onEvent?.(ev);
last = ev;
}
return { text: last.text, stopReason: last.stopReason, usage: last.usage };
}
/**
* Runs one turn and yields its events as they come; the last one is
* `{ kind: "done", stopReason, usage, text }`. Leaving the loop early cancels the turn.
*
* @param {string} prompt
* @param {{ signal?: AbortSignal }} [options]
* @returns {AsyncGenerator<Event>}
*
* @example
* for await (const ev of agent.stream("Explain main.go")) {
* if (ev.kind === "text") process.stdout.write(ev.text);
* }
*/
async *stream(prompt, { signal } = {}) {
if (this.#busy) throw new Error("minime: a turn is already running on this agent — wait for it, or use another Agent");
signal?.throwIfAborted();
this.#busy = true;
this.#cancelled = false;
const session = this.#session;
const onAbort = () => this.#cancel();
signal?.addEventListener("abort", onAbort, { once: true });
// The final answer (or the error) also arrives through nextUpdate().
session.prompt(prompt).catch(() => {});
let text = "";
let finished = false;
try {
for (;;) {
let message;
try {
message = await session.nextUpdate();
} catch (err) {
finished = true;
throw await this.#failure(err);
}
if (message.kind === "stop") {
finished = true;
yield { kind: "done", stopReason: message.stopReason, usage: message.response.usage ?? undefined, text };
return;
}
const ev = eventOf(message.update);
if (!ev) continue;
if (ev.kind === "text") text += ev.text;
yield ev;
}
} finally {
signal?.removeEventListener("abort", onAbort);
// A caller that stops iterating early cancels the turn: nothing keeps
// running unread, and the session's queue is left empty for the next turn.
if (!finished && !this.#exited) {
await this.#cancel();
await drain(session);
}
this.#busy = false;
}
}
/** Forgets the conversation: a new ACP session on the same mm process. */
async reset() {
if (this.#busy) throw new Error("minime: cannot reset during a turn");
await this.#newSession();
}
/** Closes mm's stdin and waits for it to exit (kills it after 5 s). Calling it twice is harmless. */
async close() {
this.#session?.dispose();
if (!this.#exited && this.#child.exitCode === null && this.#child.pid !== undefined) {
const exited = new Promise((resolve) => this.#child.once("exit", resolve));
this.#child.stdin.end();
const timer = setTimeout(() => this.#child.kill("SIGKILL"), 5000);
await exited;
clearTimeout(timer);
}
this.#connection.close();
}
async #open() {
// No clientCapabilities: no file system or terminal for mm to call back
// (mm does not use them), and no form support, so ask_user is not offered.
await this.#connection.agent.request("initialize", {
protocolVersion: PROTOCOL_VERSION,
clientCapabilities: {},
clientInfo: { name: "minime-sdk", version: "0.1.0" },
});
await this.#newSession();
}
async #newSession() {
const session = await this.#connection.agent.buildSession(this.#cwd).start();
this.#session?.dispose();
this.#session = session;
}
async #cancel() {
if (this.#cancelled || !this.#session) return;
this.#cancelled = true;
await this.#connection.agent.notify("session/cancel", { sessionId: this.#session.sessionId }).catch(() => {});
}
// A request FROM mm: the permission to run a tool.
async #answerPermission(params) {
const cancelled = { outcome: { outcome: "cancelled" } };
if (this.#cancelled) return cancelled; // ACP: a client that sent session/cancel answers "cancelled"
let choice = false;
try {
choice = this.#onPermission ? await this.#onPermission(toolCall(params.toolCall)) : false;
} catch {
choice = false;
}
const optionId = pickOption(params.options, choice);
return optionId ? { outcome: { outcome: "selected", optionId } } : cancelled;
}
// mm is gone (exited, or never started): every pending request fails with why.
#gone(err) {
this.#connection?.close(err);
}
// One error: the protocol's words (mm puts its own explanation under
// data.error), whether mm exited, and the last line mm wrote on stderr.
// A closed connection usually means mm is exiting: give the process a
// moment to be reaped, so that the error can say so.
async #failure(err) {
if (this.#connection.signal.aborted && !this.#exited && this.#child.pid !== undefined) {
await new Promise((resolve) => {
const timer = setTimeout(resolve, 500);
this.#child.once("exit", () => (clearTimeout(timer), resolve()));
});
}
let message = err instanceof RequestError ? rpcWords(err) : String(err?.message ?? err);
if (!message.startsWith("minime:")) message = `minime: ${message}`;
if (this.#exited && !message.includes("mm exited")) {
message += ` (mm exited: ${this.#exited.signal ?? `exit status ${this.#exited.code}`})`;
}
if (this.#stderrTail) message += ` — ${this.#stderrTail}`;
return new Error(message, { cause: err });
}
}
/**
* Reduces one ACP `session/update` to an Event; undefined for the updates that
* are editor detail (plan, usage_update, available_commands_update…) and for
* tool updates that are not an outcome (in_progress).
*
* @param {any} update
* @returns {Event|undefined}
*/
export function eventOf(update) {
switch (update.sessionUpdate) {
case "agent_message_chunk":
return update.content?.type === "text" ? { kind: "text", text: update.content.text } : undefined;
case "agent_thought_chunk":
return update.content?.type === "text" ? { kind: "thinking", text: update.content.text } : undefined;
case "tool_call":
return { kind: "tool_call", ...toolCall(update) };
case "tool_call_update":
if (update.status === "completed" || update.status === "failed") {
return { kind: "tool_result", id: update.toolCallId, status: update.status, output: outputOf(update) };
}
return undefined;
default:
return undefined;
}
}
/**
* true / false → the "once" options (falling back on the "always" ones); a
* string → that optionId when offered, else a rejection.
*
* @param {Array<{ optionId: string, kind: string }>} options
* @param {boolean|string} choice
* @returns {string|undefined}
*
* @example
* pickOption(params.options, "allow_always");
*/
export function pickOption(options, choice) {
const byKind = (kind) => options.find((o) => o.kind === kind)?.optionId;
if (typeof choice === "string" && options.some((o) => o.optionId === choice)) return choice;
if (choice === true) return byKind("allow_once") ?? byKind("allow_always");
return byKind("reject_once") ?? byKind("reject_always");
}
function toolCall(tc) {
return { id: tc.toolCallId, title: tc.title, toolKind: tc.kind ?? undefined, input: tc.rawInput };
}
// What a tool produced: mm puts it in rawOutput.output; other agents may only
// send it as text content.
function outputOf(update) {
if (typeof update.rawOutput?.output === "string") return update.rawOutput.output;
return (update.content ?? []).map((block) => (block.content?.type === "text" ? block.content.text : "")).join("");
}
// A JSON-RPC error in words: "Internal error (-32603): <mm's explanation>".
function rpcWords(err) {
const detail = typeof err.data?.error === "string" ? `: ${err.data.error}` : "";
return `${err.message} (${err.code})${detail}`;
}
// Reads a session's queue until its turn stops, so the next turn starts clean.
async function drain(session) {
try {
for (;;) if ((await session.nextUpdate()).kind === "stop") return;
} catch {
// the turn ended with an error: nothing left to read
}
}
|
