MagicCode/packages/core/src/workflow.ts

// Registre des workflows autonomes (`maia run`) : un dossier par workflow dans ~/.maia/workflows/<id>/.
//
//   workflow.json   description et état du workflow (écrit par le processus du workflow)
//   agents.json     état de chaque agent : parent + sous-agents (écrit par le processus du workflow)
//   events.jsonl    journal horodaté, une ligne JSON par évènement (ajouts seulement)
//   requests/       <req>.json : demande en attente (autorisation, intervention), écrite par le workflow
//                   <req>.answer.json : réponse, écrite par `maia approve` / `maia reply` (un autre processus)
//   stop            présence = arrêt demandé par `maia stop`
//   traces/         conversation de chaque agent
//   result.md       sortie finale de l'agent parent
//   run.log         sortie du processus en mode --detach
//
// Un seul processus écrit workflow.json, agents.json et events.jsonl : celui du workflow. Les commandes de supervision
// ne font que lire, sauf pour déposer une réponse ou une demande d'arrêt (fichiers séparés). Les écritures JSON sont
// atomiques (fichier temporaire puis renommage) : un lecteur ne voit jamais un fichier à moitié écrit.

import { appendFileSync, existsSync, mkdirSync, readFileSync, readdirSync, renameSync, writeFileSync } from "node:fs";
import { join } from "node:path";

/** États d'un agent (codes stables pour les scripts ; libellés français pour l'affichage). */
export type AgentState = "pending" | "running" | "waiting" | "done" | "error";

export const AGENT_STATE_LABELS: Record<AgentState, string> = {
  pending: "non exécuté",
  running: "en exécution",
  waiting: "en attente",
  done: "terminé",
  error: "en erreur",
};

export type PermissionMode = "wait" | "deny" | "allow";

export interface WorkflowRecord {
  id: string;
  /** Nom donné avec --name, sinon nom de l'agent parent. */
  name: string;
  /** Agent parent (orchestrateur). */
  agent: string;
  agentFile: string;
  /** Dossier de travail (projet). */
  cwd: string;
  task: string;
  model: string;
  permissionMode: PermissionMode;
  pid: number | null;
  createdAt: string;
  endedAt?: string;
  state: AgentState;
  error?: string;
  /** Délai maximal (ms) ; absent : aucun. */
  timeoutMs?: number;
}

export interface AgentRecord {
  name: string;
  role: "parent" | "sous-agent";
  state: AgentState;
  model?: string;
  /** Nombre de lancements (un sous-agent peut être lancé plusieurs fois). */
  runs: number;
  /** Lancements en cours (lancements parallèles du même agent). */
  active: number;
  startedAt?: string;
  endedAt?: string;
  error?: string;
}

export interface WorkflowRequest {
  id: string;
  agent: string;
  kind: "permission" | "intervention";
  message: string;
  /** Intervention : à qui elle s'adresse (utilisateur, équipe…). */
  recipient?: string;
  createdAt: string;
}

export interface WorkflowAnswer {
  approved?: boolean;
  text?: string;
  answeredAt: string;
}

export interface WorkflowEvent {
  at: string;
  agent: string;
  type: string;
  message: string;
}

export interface WorkflowStatus {
  workflow: WorkflowRecord;
  agents: AgentRecord[];
  /** Demandes sans réponse. */
  pending: WorkflowRequest[];
  /** État effectif (processus disparu → error). */
  state: AgentState;
  alive: boolean;
}

export const workflowsRoot = (homeDir: string): string => join(homeDir, ".maia", "workflows");

const now = (): string => new Date().toISOString();

export function writeJsonAtomic(path: string, data: unknown): void {
  const tmp = `${path}.${process.pid}.${Math.random().toString(36).slice(2)}.tmp`;
  writeFileSync(tmp, JSON.stringify(data, null, 2));
  renameSync(tmp, path);
}

export function readJson<T>(path: string): T | null {
  try {
    return JSON.parse(readFileSync(path, "utf8")) as T;
  } catch {
    return null;
  }
}

/** Le processus existe-t-il encore ? (EPERM : il existe mais appartient à un autre utilisateur.) */
export function isProcessAlive(pid: number | null | undefined): boolean {
  if (!pid) return false;
  try {
    process.kill(pid, 0);
    return true;
  } catch (err) {
    return (err as NodeJS.ErrnoException).code === "EPERM";
  }
}

/** Identifiant lisible et triable : wf_20261026_234112_a3f9. */
export function newWorkflowId(date = new Date()): string {
  const p = (n: number): string => String(n).padStart(2, "0");
  const stamp = `${date.getFullYear()}${p(date.getMonth() + 1)}${p(date.getDate())}_${p(date.getHours())}${p(date.getMinutes())}${p(date.getSeconds())}`;
  return `wf_${stamp}_${Math.random().toString(16).slice(2, 6).padEnd(4, "0")}`;
}

/** État du workflow déduit de celui de ses agents et de la présence du processus. */
export function deriveWorkflowState(workflow: WorkflowRecord, agents: AgentRecord[], alive: boolean): AgentState {
  if (workflow.state === "done" || workflow.state === "error") return workflow.state;
  if (workflow.state === "pending" && !alive) return workflow.pid ? "error" : "pending";
  if (!alive) return "error"; // processus disparu sans avoir terminé
  if (agents.some((a) => a.state === "waiting")) return "waiting";
  return "running";
}

/** Accès en lecture/écriture à un workflow. */
export class WorkflowHandle {
  constructor(readonly dir: string) {}

  get id(): string {
    return this.dir.split(/[\\/]/).pop()!;
  }

  private path(name: string): string {
    return join(this.dir, name);
  }

  readWorkflow(): WorkflowRecord | null {
    return readJson<WorkflowRecord>(this.path("workflow.json"));
  }
  writeWorkflow(record: WorkflowRecord): void {
    writeJsonAtomic(this.path("workflow.json"), record);
  }
  updateWorkflow(patch: Partial<WorkflowRecord>): WorkflowRecord {
    const current = this.readWorkflow();
    if (!current) throw new Error(`workflow ${this.id} illisible`);
    const next = { ...current, ...patch };
    this.writeWorkflow(next);
    return next;
  }

  readAgents(): AgentRecord[] {
    return readJson<AgentRecord[]>(this.path("agents.json")) ?? [];
  }
  writeAgents(agents: AgentRecord[]): void {
    writeJsonAtomic(this.path("agents.json"), agents);
  }

  event(agent: string, type: string, message: string): void {
    const ev: WorkflowEvent = { at: now(), agent, type, message };
    appendFileSync(this.path("events.jsonl"), `${JSON.stringify(ev)}\n`);
  }
  readEvents(): WorkflowEvent[] {
    let raw = "";
    try {
      raw = readFileSync(this.path("events.jsonl"), "utf8");
    } catch {
      return [];
    }
    return raw
      .split("\n")
      .filter(Boolean)
      .flatMap((line) => {
        try {
          return [JSON.parse(line) as WorkflowEvent];
        } catch {
          return [];
        }
      });
  }

  // ----- Demandes (autorisations, interventions) -----

  createRequest(req: Omit<WorkflowRequest, "id" | "createdAt">): WorkflowRequest {
    const dir = this.path("requests");
    mkdirSync(dir, { recursive: true });
    const n = readdirSync(dir).filter((f) => /^r\d+\.json$/.test(f)).length + 1;
    const full: WorkflowRequest = { ...req, id: `r${n}`, createdAt: now() };
    writeJsonAtomic(join(dir, `${full.id}.json`), full);
    return full;
  }
  readRequests(): WorkflowRequest[] {
    const dir = this.path("requests");
    let files: string[] = [];
    try {
      files = readdirSync(dir);
    } catch {
      return [];
    }
    return files
      .filter((f) => /^r\d+\.json$/.test(f))
      .map((f) => readJson<WorkflowRequest>(join(dir, f)))
      .filter((r): r is WorkflowRequest => r !== null)
      .sort((a, b) => Number(a.id.slice(1)) - Number(b.id.slice(1)));
  }
  readAnswer(requestId: string): WorkflowAnswer | null {
    return readJson<WorkflowAnswer>(join(this.path("requests"), `${requestId}.answer.json`));
  }
  /** Écrit par `maia approve` / `maia reply` : une seule réponse par demande. */
  answer(requestId: string, answer: Omit<WorkflowAnswer, "answeredAt">): void {
    const req = this.readRequests().find((r) => r.id === requestId);
    if (!req) throw new Error(`demande ${requestId} introuvable dans le workflow ${this.id}`);
    if (this.readAnswer(requestId)) throw new Error(`la demande ${requestId} a déjà reçu une réponse`);
    writeJsonAtomic(join(this.path("requests"), `${requestId}.answer.json`), { ...answer, answeredAt: now() });
  }
  pendingRequests(): WorkflowRequest[] {
    return this.readRequests().filter((r) => !this.readAnswer(r.id));
  }

  // ----- Arrêt -----

  requestStop(): void {
    writeFileSync(this.path("stop"), now());
  }
  stopRequested(): boolean {
    return existsSync(this.path("stop"));
  }

  // ----- Résultat et traces -----

  writeResult(text: string): void {
    writeFileSync(this.path("result.md"), text);
  }
  readResult(): string | null {
    try {
      return readFileSync(this.path("result.md"), "utf8");
    } catch {
      return null;
    }
  }
  writeTrace(name: string, data: unknown): string {
    const dir = this.path("traces");
    mkdirSync(dir, { recursive: true });
    const file = join(dir, `${name}.json`);
    writeJsonAtomic(file, data);
    return file;
  }

  status(): WorkflowStatus | null {
    const workflow = this.readWorkflow();
    if (!workflow) return null;
    const agents = this.readAgents();
    const alive = isProcessAlive(workflow.pid);
    return { workflow, agents, pending: this.pendingRequests(), state: deriveWorkflowState(workflow, agents, alive), alive };
  }
}

/** Le registre global : ~/.maia/workflows. */
export class WorkflowRegistry {
  readonly root: string;
  constructor(homeDir: string) {
    this.root = workflowsRoot(homeDir);
  }

  create(record: Omit<WorkflowRecord, "id" | "createdAt" | "state"> & { id?: string }): WorkflowHandle {
    mkdirSync(this.root, { recursive: true });
    let id = record.id ?? newWorkflowId();
    while (!record.id && existsSync(join(this.root, id))) id = newWorkflowId();
    const handle = new WorkflowHandle(join(this.root, id));
    mkdirSync(handle.dir, { recursive: true });
    handle.writeWorkflow({ ...record, id, createdAt: now(), state: "pending" });
    handle.writeAgents([]);
    return handle;
  }

  ids(): string[] {
    try {
      return readdirSync(this.root)
        .filter((d) => d.startsWith("wf_") && existsSync(join(this.root, d, "workflow.json")))
        .sort();
    } catch {
      return [];
    }
  }

  get(id: string): WorkflowHandle {
    return new WorkflowHandle(join(this.root, id));
  }

  /**
   * Retrouve un workflow : identifiant complet, morceau unique de l'identifiant, nom (le plus récent), ou « last ».
   * Lève une erreur explicite si rien ne correspond ou si c'est ambigu.
   */
  find(ref: string): WorkflowHandle {
    const ids = this.ids();
    if (ids.length === 0) throw new Error("aucun workflow enregistré");
    if (ref === "last") return this.get(ids.at(-1)!);
    if (ids.includes(ref)) return this.get(ref);
    const byId = ids.filter((id) => id.includes(ref));
    if (byId.length === 1) return this.get(byId[0]!);
    const byName = ids.filter((id) => this.get(id).readWorkflow()?.name === ref);
    if (byName.length > 0) return this.get(byName.at(-1)!);
    if (byId.length > 1) throw new Error(`« ${ref} » est ambigu : ${byId.join(", ")}`);
    throw new Error(`workflow « ${ref} » introuvable`);
  }

  /** États de tous les workflows (les plus récents en dernier). */
  list(): WorkflowStatus[] {
    return this.ids()
      .map((id) => this.get(id).status())
      .filter((s): s is WorkflowStatus => s !== null);
  }
}