377 lines
15 KiB
TypeScript
377 lines
15 KiB
TypeScript
// `maia run AGENT.md` : workflow autonome = un agent parent (orchestrateur) sans interface + ses sous-agents.
|
||
// L'état de chaque agent est tenu à jour dans le registre ~/.maia/workflows/<id>/ (voir core/src/workflow.ts).
|
||
|
||
import type { ChildProcess } from "node:child_process";
|
||
import {
|
||
AGENT_STATE_LABELS,
|
||
DEFAULT_AGENT_TASK,
|
||
WorkflowRegistry,
|
||
createSession,
|
||
finalAssistantText,
|
||
requestFinalSummary,
|
||
runTurn,
|
||
type AgentRecord,
|
||
type AgentState,
|
||
type AgentUI,
|
||
type LLMProvider,
|
||
type PermissionMode,
|
||
type Tool,
|
||
type WorkflowHandle,
|
||
} from "@maia/core";
|
||
import {
|
||
INTERVENTION_TOOL_NAME,
|
||
initializeOrchestrator,
|
||
loadOrchestrator,
|
||
resolveSubAgentModel,
|
||
} from "@maia/agent";
|
||
import { createProvider, loadModelChoices, loadSettings, type MaiaSettings } from "@maia/providers";
|
||
import { createSubAgentLauncher } from "./subagent-parent";
|
||
|
||
/** Délai maximal d'un sous-agent dans un workflow : celui du workflow (un sous-agent peut attendre une autorisation longtemps). */
|
||
const MAX_TIMER_MS = 2_147_483_647;
|
||
export const DEFAULT_POLL_MS = 1000;
|
||
|
||
export interface WorkflowRunOptions {
|
||
/** AGENT.md de l'agent parent. */
|
||
file: string;
|
||
task?: string;
|
||
model?: string;
|
||
name?: string;
|
||
permissionMode?: PermissionMode;
|
||
timeoutMs?: number;
|
||
/** Workflow déjà créé (mode --detach : le processus détaché reprend l'identifiant). */
|
||
workflowId?: string;
|
||
/** Dossier de lancement (sert de dossier de travail si le AGENT.md n'est pas dans un dossier .maia/agents). */
|
||
launchDir: string;
|
||
homeDir: string;
|
||
}
|
||
|
||
export interface WorkflowRunDeps {
|
||
settings?: MaiaSettings;
|
||
/** Modèles disponibles (sinon : interrogation de LiteLLM). */
|
||
availableModels?: string[];
|
||
createProvider?(model: string, settings: MaiaSettings): LLMProvider;
|
||
/** Processus d'un sous-agent (tests). */
|
||
spawnChild?: () => ChildProcess;
|
||
/** Progression lisible (une ligne par évènement). */
|
||
log?(line: string): void;
|
||
/** Annonce de l'identifiant dès sa création. */
|
||
onCreated?(id: string): void;
|
||
pollMs?: number;
|
||
}
|
||
|
||
export interface WorkflowRunResult {
|
||
id: string;
|
||
state: AgentState;
|
||
exitCode: number;
|
||
result: string;
|
||
error?: string;
|
||
}
|
||
|
||
const time = (): string => new Date().toTimeString().slice(0, 8);
|
||
|
||
/** États des agents d'un workflow, enregistrés à chaque changement. */
|
||
class AgentBook {
|
||
private readonly agents = new Map<string, AgentRecord & { waits: number }>();
|
||
constructor(private readonly handle: WorkflowHandle) {}
|
||
|
||
private save(): void {
|
||
this.handle.writeAgents([...this.agents.values()].map(({ waits: _w, ...a }) => a));
|
||
}
|
||
private get(name: string, role: AgentRecord["role"] = "sous-agent"): AgentRecord & { waits: number } {
|
||
let a = this.agents.get(name);
|
||
if (!a) {
|
||
a = { name, role, state: "pending", runs: 0, active: 0, waits: 0 };
|
||
this.agents.set(name, a);
|
||
}
|
||
return a;
|
||
}
|
||
declare(name: string, role: AgentRecord["role"], model?: string): void {
|
||
const a = this.get(name, role);
|
||
if (model) a.model = model;
|
||
this.save();
|
||
}
|
||
start(name: string, model: string, role: AgentRecord["role"] = "sous-agent"): void {
|
||
const a = this.get(name, role);
|
||
a.model = model;
|
||
a.runs++;
|
||
a.active++;
|
||
a.state = a.waits > 0 ? "waiting" : "running";
|
||
a.startedAt ??= new Date().toISOString();
|
||
delete a.endedAt;
|
||
delete a.error;
|
||
this.save();
|
||
}
|
||
end(name: string, ok: boolean, error?: string): void {
|
||
const a = this.get(name);
|
||
a.active = Math.max(0, a.active - 1);
|
||
if (!ok && error) a.error = error;
|
||
if (a.active === 0) {
|
||
a.state = ok && !a.error ? "done" : "error";
|
||
a.endedAt = new Date().toISOString();
|
||
}
|
||
this.save();
|
||
}
|
||
wait(name: string, delta: 1 | -1): void {
|
||
const a = this.get(name);
|
||
a.waits = Math.max(0, a.waits + delta);
|
||
if (a.active > 0) a.state = a.waits > 0 ? "waiting" : "running";
|
||
this.save();
|
||
}
|
||
/** Fin du workflow : un agent encore actif est interrompu. */
|
||
closeAll(reason: string): void {
|
||
for (const a of this.agents.values()) {
|
||
if (a.state === "running" || a.state === "waiting") {
|
||
a.state = "error";
|
||
a.error = reason;
|
||
a.active = 0;
|
||
a.waits = 0;
|
||
a.endedAt = new Date().toISOString();
|
||
}
|
||
}
|
||
this.save();
|
||
}
|
||
}
|
||
|
||
async function resolveModel(
|
||
requested: string | undefined,
|
||
settings: MaiaSettings,
|
||
available: string[] | undefined,
|
||
log: (line: string) => void,
|
||
): Promise<string> {
|
||
const catalog = available ?? (await loadModelChoices(settings)).choices.map((c) => c.value);
|
||
if (requested) {
|
||
const asSpec = requested.includes(":") ? requested : `litellm:${requested}`;
|
||
if (catalog.length === 0) return asSpec; // catalogue indisponible : le modèle demandé est pris tel quel
|
||
const r = resolveSubAgentModel(requested, asSpec, catalog);
|
||
if (r.warning) log(`Modèle « ${requested} » introuvable parmi les modèles disponibles : utilisé tel quel (${asSpec}).`);
|
||
return r.spec;
|
||
}
|
||
const first = catalog[0];
|
||
if (!first) throw new Error("Aucun modèle indiqué (--model, champ model de l'agent, DefaultModel du settings.json) et aucun modèle disponible auprès de LiteLLM.");
|
||
log(`Aucun modèle indiqué : ${first} (premier modèle disponible). Fixez « DefaultModel » dans ~/.maia/settings.json pour le choisir.`);
|
||
return first;
|
||
}
|
||
|
||
/** Exécute un workflow jusqu'au bout. Ne lève pas d'erreur : l'échec est porté par le résultat (et le registre). */
|
||
export async function runWorkflow(opts: WorkflowRunOptions, deps: WorkflowRunDeps = {}): Promise<WorkflowRunResult> {
|
||
const log = deps.log ?? (() => undefined);
|
||
const registry = new WorkflowRegistry(opts.homeDir);
|
||
let handle: WorkflowHandle | null = opts.workflowId ? registry.get(opts.workflowId) : null;
|
||
|
||
// Échec avant même la création du workflow (fichier, modèle…) : il est quand même enregistré, en erreur
|
||
const failEarly = (message: string): WorkflowRunResult => {
|
||
if (!handle) {
|
||
handle = registry.create({
|
||
name: opts.name ?? opts.file,
|
||
agent: opts.file,
|
||
agentFile: opts.file,
|
||
cwd: opts.launchDir,
|
||
task: opts.task ?? "",
|
||
model: opts.model ?? "",
|
||
permissionMode: opts.permissionMode ?? "wait",
|
||
pid: process.pid,
|
||
});
|
||
deps.onCreated?.(handle.id);
|
||
}
|
||
handle.updateWorkflow({ state: "error", error: message, endedAt: new Date().toISOString(), pid: process.pid });
|
||
handle.event("workflow", "error", message);
|
||
log(`[${time()}] Workflow en erreur : ${message}`);
|
||
return { id: handle.id, state: "error", exitCode: 1, result: "", error: message };
|
||
};
|
||
|
||
let settings: MaiaSettings;
|
||
let def: Awaited<ReturnType<typeof loadOrchestrator>>;
|
||
let model: string;
|
||
try {
|
||
settings = deps.settings ?? (await loadSettings(opts.homeDir));
|
||
def = await loadOrchestrator(opts.file, opts.launchDir);
|
||
model = await resolveModel(opts.model ?? def.agent.model ?? settings.DefaultModel, settings, deps.availableModels, log);
|
||
} catch (err) {
|
||
return failEarly((err as Error).message);
|
||
}
|
||
const parent = def.agent.name;
|
||
const permissionMode: PermissionMode = opts.permissionMode ?? def.permissionMode ?? "wait";
|
||
const timeoutMs = opts.timeoutMs ?? def.timeoutMs;
|
||
const task = opts.task?.trim() || DEFAULT_AGENT_TASK;
|
||
|
||
const record = {
|
||
name: opts.name ?? parent,
|
||
agent: parent,
|
||
agentFile: def.file,
|
||
cwd: def.cwd,
|
||
task,
|
||
model,
|
||
permissionMode,
|
||
pid: process.pid,
|
||
...(timeoutMs ? { timeoutMs } : {}),
|
||
};
|
||
if (handle) handle.updateWorkflow({ ...record, state: "running" });
|
||
else {
|
||
handle = registry.create(record);
|
||
handle.updateWorkflow({ state: "running" });
|
||
}
|
||
const wf: WorkflowHandle = handle;
|
||
deps.onCreated?.(wf.id);
|
||
const book = new AgentBook(wf);
|
||
const say = (agent: string, type: string, message: string): void => {
|
||
wf.event(agent, type, message);
|
||
log(`[${time()}] ${agent} › ${message}`);
|
||
};
|
||
for (const w of def.warnings) say(parent, "warn", w);
|
||
say("workflow", "start", `Workflow ${wf.id} démarré : agent ${parent}, modèle ${model}, autorisations « ${permissionMode} », dossier ${def.cwd}`);
|
||
|
||
const controller = new AbortController();
|
||
const stopped: { reason: string | null } = { reason: null };
|
||
const abort = (reason: string): void => {
|
||
if (stopped.reason) return;
|
||
stopped.reason = reason;
|
||
controller.abort(new Error(reason));
|
||
};
|
||
const pollMs = deps.pollMs ?? DEFAULT_POLL_MS;
|
||
const stopTimer = setInterval(() => {
|
||
if (wf.stopRequested()) abort("arrêté (maia stop)");
|
||
}, pollMs);
|
||
const timeoutTimer = timeoutMs ? setTimeout(() => abort(`délai maximal dépassé (${Math.round(timeoutMs / 60_000)} min)`), Math.min(timeoutMs, MAX_TIMER_MS)) : undefined;
|
||
|
||
/** Attend la réponse à une demande (fichier déposé par maia approve / reply) ; null si le workflow est arrêté. */
|
||
const waitAnswer = (requestId: string): Promise<{ approved?: boolean; text?: string } | null> =>
|
||
new Promise((resolve) => {
|
||
const tick = (): void => {
|
||
const answer = wf.readAnswer(requestId);
|
||
if (answer) return resolve(answer);
|
||
if (controller.signal.aborted) return resolve(null);
|
||
setTimeout(tick, pollMs);
|
||
};
|
||
tick();
|
||
});
|
||
|
||
/** Autorisation d'outil selon le mode : allow / deny immédiat, wait = demande enregistrée et attente de `maia approve`. */
|
||
const askPermission = async (agent: string, message: string): Promise<boolean> => {
|
||
if (permissionMode === "allow") {
|
||
say(agent, "permission", `autorisé automatiquement (mode allow) : ${message}`);
|
||
return true;
|
||
}
|
||
if (permissionMode === "deny") {
|
||
say(agent, "permission", `refusé (mode deny) : ${message}`);
|
||
return false;
|
||
}
|
||
const req = wf.createRequest({ agent, kind: "permission", message });
|
||
book.wait(agent, 1);
|
||
say(agent, "waiting", `EN ATTENTE ${req.id} : ${message} → maia approve ${wf.id} ${req.id} (ou --deny)`);
|
||
const answer = await waitAnswer(req.id);
|
||
book.wait(agent, -1);
|
||
const ok = answer?.approved === true;
|
||
say(agent, "permission", `${req.id} ${ok ? "autorisé" : "refusé"}${answer ? "" : " (workflow arrêté)"}`);
|
||
return ok;
|
||
};
|
||
|
||
const interventionTool: Tool = {
|
||
spec: {
|
||
name: INTERVENTION_TOOL_NAME,
|
||
description:
|
||
"Demande l'intervention d'un humain (l'utilisateur ou un tiers) et attend sa réponse : le workflow passe « en attente » jusqu'à " +
|
||
"`maia reply`. À utiliser quand une contrainte l'exige ou qu'une décision humaine est nécessaire. Retourne la réponse.",
|
||
inputSchema: {
|
||
type: "object",
|
||
properties: {
|
||
message: { type: "string", description: "Ce qui bloque et ce qui est attendu (décision, action, information)" },
|
||
recipient: { type: "string", description: "À qui s'adresse la demande (ex. utilisateur, équipe sécurité)" },
|
||
},
|
||
required: ["message"],
|
||
},
|
||
},
|
||
permission: "auto",
|
||
describe: (input) => String(input.message ?? "").slice(0, 120),
|
||
async execute(input) {
|
||
const message = typeof input.message === "string" ? input.message.trim() : "";
|
||
if (!message) throw new Error('Paramètre "message" manquant : décris ce qui est attendu.');
|
||
const recipient = typeof input.recipient === "string" && input.recipient.trim() ? input.recipient.trim() : undefined;
|
||
const req = wf.createRequest({ agent: parent, kind: "intervention", message, ...(recipient ? { recipient } : {}) });
|
||
book.wait(parent, 1);
|
||
say(parent, "waiting", `EN ATTENTE ${req.id} (intervention${recipient ? ` : ${recipient}` : ""}) : ${message} → maia reply ${wf.id} ${req.id} "réponse"`);
|
||
const answer = await waitAnswer(req.id);
|
||
book.wait(parent, -1);
|
||
if (!answer) throw new Error("Workflow arrêté pendant l'attente de l'intervention.");
|
||
say(parent, "reply", `${req.id} : réponse reçue`);
|
||
return `Réponse${recipient ? ` (${recipient})` : ""} : ${answer.text ?? (answer.approved === false ? "refusé" : "accepté")}`;
|
||
},
|
||
};
|
||
|
||
const traceCount = new Map<string, number>();
|
||
const launchSubAgent = createSubAgentLauncher({
|
||
cwd: def.cwd,
|
||
homeDir: opts.homeDir,
|
||
settings,
|
||
getModel: () => model,
|
||
getSessionName: () => null,
|
||
store: null,
|
||
confirm: (message, agent) => askPermission(agent ?? parent, message),
|
||
info: (m) => say(parent, "info", m),
|
||
warn: (m) => say(parent, "warn", m),
|
||
timeoutMs: MAX_TIMER_MS,
|
||
...(deps.spawnChild ? { spawn: deps.spawnChild } : {}),
|
||
onAgentState: (e) => {
|
||
if (e.state === "running") book.start(e.agent, e.model);
|
||
else book.end(e.agent, e.state === "done", e.error);
|
||
say(e.agent, "state", AGENT_STATE_LABELS[e.state] + (e.error ? ` : ${e.error}` : ""));
|
||
},
|
||
onTranscript: (e) => {
|
||
const n = (traceCount.get(e.agent) ?? 0) + 1;
|
||
traceCount.set(e.agent, n);
|
||
wf.writeTrace(n === 1 ? e.agent : `${e.agent}-${n}`, { agent: e.agent, model: e.model, startedAt: e.startedAt, ok: e.ok, messages: e.messages });
|
||
},
|
||
});
|
||
|
||
// L'agent parent n'a pas d'interface : progression dans le journal, autorisations selon le mode
|
||
const ui: AgentUI = {
|
||
onThinking() {},
|
||
onText() {},
|
||
onTextEnd() {},
|
||
onToolCall: (name, summary) => say(parent, "tool", `${name} ${summary}`),
|
||
onToolResult: (name, output, isError) => {
|
||
if (isError) say(parent, "tool-error", `${name} : ${output.slice(0, 300)}`);
|
||
},
|
||
confirm: (message) => askPermission(parent, message),
|
||
};
|
||
|
||
let init: Awaited<ReturnType<typeof initializeOrchestrator>> | null = null;
|
||
let session: ReturnType<typeof createSession> | null = null;
|
||
try {
|
||
init = await initializeOrchestrator({
|
||
def,
|
||
homeDir: opts.homeDir,
|
||
launchSubAgent,
|
||
extraTools: [interventionTool],
|
||
log: (m) => say(parent, "info", m),
|
||
warn: (m) => say(parent, "warn", m),
|
||
});
|
||
book.start(parent, model, "parent");
|
||
for (const name of init.mentioned) book.declare(name, "sous-agent");
|
||
const provider = (deps.createProvider ?? createProvider)(model, settings);
|
||
session = createSession({ provider, init, ui, cwd: def.cwd });
|
||
|
||
const end = await runTurn(session, task, controller.signal, { maxIterations: init.maxIterations });
|
||
if (stopped.reason) throw new Error(stopped.reason);
|
||
let result = end === "done" ? finalAssistantText(session.messages) : "";
|
||
if (!result) result = await requestFinalSummary(session, controller.signal);
|
||
wf.writeResult(result);
|
||
book.end(parent, true);
|
||
book.closeAll("interrompu à la fin du workflow");
|
||
wf.updateWorkflow({ state: "done", endedAt: new Date().toISOString() });
|
||
say("workflow", "done", "Workflow terminé");
|
||
return { id: wf.id, state: "done", exitCode: 0, result };
|
||
} catch (err) {
|
||
const message = stopped.reason ?? (err as Error).message;
|
||
book.end(parent, false, message);
|
||
book.closeAll(message);
|
||
wf.updateWorkflow({ state: "error", error: message, endedAt: new Date().toISOString() });
|
||
say("workflow", "error", `Workflow en erreur : ${message}`);
|
||
return { id: wf.id, state: "error", exitCode: stopped.reason?.startsWith("arrêté") ? 130 : 1, result: "", error: message };
|
||
} finally {
|
||
clearInterval(stopTimer);
|
||
clearTimeout(timeoutTimer);
|
||
if (session) wf.writeTrace(parent, { agent: parent, model, messages: session.messages });
|
||
await init?.close().catch(() => undefined);
|
||
}
|
||
}
|