AUDIT / src /lib /agent /GraphOrchestrator.ts
Arypulka98's picture
feat(audit): deploy full backend cluster node (part 2)
cc11e77 verified
Raw
History Blame Contribute Delete
35.9 kB
/**
* GraphOrchestrator.ts — Executes a TaskGraph (FASE 4.1)
*
* Pipeline:
* TaskGraph → [ready nodes] → execute in parallel → mark done →
* → next layer → ... → synthesize → updateMemory
*
* The orchestrator calls the same `executeTool` function used by
* the ReAct loop — no duplication, no regressions.
*
* Designed for iPhone: low memory footprint, no blocking operations,
* concurrency capped at 2 parallel tool calls.
*
* ── Resilience additions (S760) ───────────────────────────────────────────────
*
* S760-A │ Abort cleanup ("Dangling Nodes")
* │ Quando AbortSignal scatta, tutti i nodi in stato "running"
* │ vengono marcati "failed" con msg "abortito" e i loro dipendenti
* │ propagati come "skipped". Lo stato finale è sempre coerente.
*
* S760-B │ Memory offload su IndexedDB ("Memory Pressure" iOS)
* │ Risultati tool > OFFLOAD_THRESHOLD_BYTES (4 KB) vengono scritti
* │ su vfsDb.graphToolResults. In memoria resta solo un ref
* │ "@@dexie:{nodeId}". La synthesis legge da Dexie al momento
* │ della(). Fallsafe: se Dexie fallisce, usa memoria.
*
* S760-C │ Adaptive replanning su nodo critico fallito ("Silent Dep Failures")
* │ Se un nodo con priority < CRITICAL_PRIORITY fallisce, invoca
* │ adaptiveReplan() che costruisce un piano alternativo e lo
* │ injetta nel grafo corrente senza ripartire da zero.
*
* S760-D │ Checkpointing + resume ("State Serialization")
* │ Dopo ogni markNode, il grafo viene serializzato su
* │ vfsDb.taskGraphs. All'avvio, loadIncompleteGraph() ritorna
* │ il grafo parziale (se esistente) permettendo il resume.
* │ Cleanup automatico dopo completamento.
*
* Invarianti rispettate:
* - S330: nessun setInterval (i checkpoint usano await Dexie sincrona)
* - S639: slot _running rilasciato durante backoff — non modified
* - S646: retryPolicy dal contratto del tool — non modified
* - S643: verifier wrapped in try/catch — non modified
* - S637: chiavi composite per tool duplicati — non modified
* - Safari-safe: nessun SharedArrayBuffer, AbortSignal.any() sostituito
* da listener manuale (già presente)
*/
import {
type TaskGraph,
type TaskNode,
buildGraph,
getReadyNodes,
markNode,
graphDone,
} from "./TaskGraph";
import { verifyToolResult, annotateWithProof } from "./verifier";
import { updateMemoryFromGraph, sessionMemory } from "./MemoryUpdater";
import { safeValidateInput, getToolTimeout, TOOL_CONTRACTS } from ".";
// ─── vfsDb import (lazy per evitare circular deps a load time) ────────────────
// Nota: db è VFSDatabase | null — usa db?.table invece di (db as any).table
// Importato in modo lazy nelle funzioni che lo usano (checkpoint + offload)
// così se IndexedDB non è disponibile (test unitari, SSR) il resto funziona.
let _vfsDb: typeof import("../vfsDb").vfsDb | null = null;
async function _getVfsDb() {
if (!_vfsDb) {
try {
const mod = await import("../vfsDb");
_vfsDb = mod.vfsDb;
} catch {
_vfsDb = null;
}
}
return _vfsDb;
}
export type ToolExecutor = (name: string, args: Record<string, unknown>) => Promise<string>;
export interface OrchestratorCallbacks {
onNodeStart?: (node: TaskNode) => void;
onNodeDone?: (node: TaskNode) => void;
onNodeFailed?: (node: TaskNode) => void;
onStatus?: (msg: string) => void;
onProgress?: (done: number, total: number) => void;
}
export interface OrchestrationResult {
success: boolean;
output: string; // aggregated results for synthesizer
toolResults: Map<string, string>;
failedNodes: string[];
skippedNodes: string[]; // S637: nodi saltati per dipendenza fallita upstream
memoryUpdated: boolean;
}
// ─── S760-B: Memory offload constants ────────────────────────────────────────
// Risultati > 4 KB vengono offloadati su Dexie invece di restare in RAM.
// Su iPhone con 3-4 GB RAM condivisa tra app e browser, ogni Map entry
// da 10+ KB è un potenziale contributor al tab kill di Safari.
const OFFLOAD_THRESHOLD_BYTES = 4_096; // 4 KB
const OFFLOAD_REF_PREFIX = "@@dexie:"; // sentinella in toolResults Map
// ─── S760-C: Adaptive replanning threshold ────────────────────────────────────
// Nodi con priority < soglia sono "critici" — il fallimento attiva il replanning.
// La priority viene da executionPlanner (S421). 0 = primissimo nel piano.
// Soglia 3 = i 3 passi più importanti del piano (auth, db, api-router)
// sono critici; quelli secondari (UI, styling, test) non triggerano replan.
const CRITICAL_PRIORITY = 3;
// Fix 1 (S381): concorrenza dinamica basata sul tipo di tool
const LIGHT_TOOLS = new Set([
"get_weather", "get_news", "get_currency", "get_stock",
"search_wikipedia", "math_eval", "get_datetime",
"search_github", "generate_image", "get_ip_info", "get_joke",
]);
const HEAVY_TOOLS = new Set([
"run_code", "execute_shell", "pip_install",
"read_pdf", "create_webpage", "write_file", "iterate_file",
]);
function maxConcurrentFor(nodes: TaskNode[]): number {
const tools = nodes.map(n => n.tool ?? "");
if (tools.every(t => LIGHT_TOOLS.has(t))) return 4;
if (tools.some(t => HEAVY_TOOLS.has(t))) return 1;
return 2;
}
const TICK_MS = 50;
function isTransientError(msg: string): boolean {
const m = msg.toLowerCase();
return (
m.includes("fetch failed") ||
m.includes("network error") ||
m.includes("networkerror") ||
m.includes("failed to fetch") ||
m.includes("econnrefused") ||
m.includes("enotfound") ||
m.includes("socket hang up") ||
m.includes("timeout") ||
m.includes("429") ||
m.includes("rate limit") ||
m.includes("too many requests") ||
m.includes("503") ||
m.includes("502") ||
m.includes("504") ||
m.includes("service unavailable") ||
m.includes("bad gateway") ||
m.includes("gateway timeout")
);
}
// ─── S760-A: Abort cleanup ────────────────────────────────────────────────────
/**
* Marca come "failed" tutti i nodi ancora in stato "running" dopo un abort.
* La propagazione degli skipped viene gestita da markNode() già esistente.
* Deve essere chiamata nel finally del loop principale, SEMPRE.
*/
function _cleanupDanglingNodes(graph: TaskGraph, reason: string): void {
for (const node of graph.nodes.values()) {
if (node.status === "running") {
// markNode propaga automaticamente "skipped" ai dipendenti
markNode(graph, node.id, "failed", undefined, `abortito: ${reason}`);
}
}
}
// ─── S760-B: Tool result offloading ──────────────────────────────────────────
/**
* Scrive il risultato di un tool su Dexie se supera la soglia.
* Ritorna la stringa da usare in toolResults: o il result originale (piccolo)
* o un ref "@@dexie:{nodeId}" (grande, offloadato).
* Fire-and-forget safe: se Dexie fallisce, usa memoria senza crash.
*/
async function _maybeOffloadResult(nodeId: string, result: string): Promise<string> {
// Usa byteLength (UTF-8) se disponibile, altrimenti lunghezza caratteri
const byteSize = (typeof TextEncoder !== "undefined")
? new TextEncoder().encode(result).length
: result.length;
if (byteSize <= OFFLOAD_THRESHOLD_BYTES) return result;
try {
const db = await _getVfsDb();
if (!db) return result; // Dexie non disponibile → usa memoria
// eslint-disable-next-line @typescript-eslint/no-explicit-any
await db?.graphToolResults?.put({ id: nodeId, result, savedAt: Date.now() });
return `${OFFLOAD_REF_PREFIX}${nodeId}`;
} catch {
// Dexie piena o errore → fallback: tronca il risultato in memoria
return result.slice(0, OFFLOAD_THRESHOLD_BYTES) + " [troncato per memoria]";
}
}
/**
* Risolve un ref "@@dexie:{nodeId}" in Dexie, o ritorna la stringa diretta.
* Usata da collectResultsResolved() nella fase di synthesis.
*/
async function _resolveOffloadedResult(refOrResult: string): Promise<string> {
if (!refOrResult.startsWith(OFFLOAD_REF_PREFIX)) return refOrResult;
const nodeId = refOrResult.slice(OFFLOAD_REF_PREFIX.length);
try {
const db = await _getVfsDb();
if (!db) return "[risultato non disponibile]";
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const entry = await db?.graphToolResults?.get(nodeId);
return entry?.result ?? "[risultato scaduto da Dexie]";
} catch {
return "[errore lettura Dexie]";
}
}
/**
* Versione async di che risolve i ref Dexie prima di aggregare.
* Sostituisce(graph) nella fase di synthesis quando l'offload è attivo.
*/
async function collectResultsResolved(graph: TaskGraph): Promise<string> {
const results: string[] = [];
for (const node of graph.nodes.values()) {
if (node.type === "tool_call" && node.result) {
const resolved = await _resolveOffloadedResult(node.result);
results.push(`[${node.label}]\n${resolved}`);
}
}
return results.join("\n\n---\n\n");
}
// ─── S760-C: Adaptive replanning ─────────────────────────────────────────────
/**
* Tenta di costruire un piano alternativo per un nodo critico fallito.
* Non usa LLM esterno per non aggiungere latency — usa una strategia
* deterministica: costruisce un singolo nodo di "recupero" che usa
* un tool alternativo per raggiungere lo stesso obiettivo parziale.
*
* Logica:
* - tool fallito è di tipo "fetch" → prova tool alternativo nello stesso gruppo
* - tool fallito è "write_file" → prova "apply_patch" sullo stesso path
* - altrimenti → aggiunge un nodo "run_code" con il label del nodo fallito
*
* Il nodo alternativo è inserito nel grafo DOPO tutti i nodi done,
* con deps che puntano all'ultimo nodo done (non quello fallito).
* I nodi che dipendevano dal fallito vengono re-wired al nuovo nodo.
*/
function _adaptiveReplan(graph: TaskGraph, failedNodeId: string): boolean {
const failedNode = graph.nodes.get(failedNodeId);
if (!failedNode?.tool) return false;
// Mappa di fallback per tool type
const FETCH_FALLBACKS: Record<string, string> = {
web_search: "web_research",
web_research: "get_news",
read_page: "fetch_url",
fetch_url: "web_search",
search_wikipedia: "web_search",
search_github: "web_search",
};
const WRITE_FALLBACKS: Record<string, string> = {
write_file: "iterate_file",
iterate_file: "write_file",
apply_patch: "write_file",
run_code: "execute_shell",
execute_shell:"run_code",
};
const fallbackTool =
FETCH_FALLBACKS[failedNode.tool] ??
WRITE_FALLBACKS[failedNode.tool] ??
null;
if (!fallbackTool) return false;
// Trova l'ultimo nodo "done" come nuovo anchor delle dipendenze
const lastDoneId = [...graph.nodes.values()]
.filter(n => n.status === "done")
.sort((a, b) => (b.finishedAt ?? 0) - (a.finishedAt ?? 0))[0]?.id;
if (!lastDoneId) return false;
// Crea nodo alternativo
const altId = `replan-${failedNodeId}-${Date.now()}`;
const altNode: TaskNode = {
id: altId,
type: "tool_call",
label: `[Replan] ${failedNode.label} via ${fallbackTool}`,
tool: fallbackTool,
input: { ...(failedNode.input ?? {}), _replan: true },
deps: [lastDoneId],
status: "pending",
priority: (failedNode.priority ?? 999) + 0.5, // subito dopo il fallito
};
graph.nodes.set(altId, altNode);
graph.edges.push({ from: lastDoneId, to: altId });
// Re-wire: i nodi che dipendevano dal fallito ora dipendono dall'alternativo
for (const node of graph.nodes.values()) {
if (node.status === "skipped" && node.deps.includes(failedNodeId)) {
// Riporta "pending" e sostituisce la dipendenza
node.status = "pending";
node.error = undefined;
node.deps = node.deps.map(d => d === failedNodeId ? altId : d);
graph.edges.push({ from: altId, to: node.id });
}
}
return true; // replan riuscito
}
// ─── S760-D: Graph checkpointing ─────────────────────────────────────────────
/**
* Serializza il grafo su Dexie.taskGraphs.
* Fire-and-forget: mai lancia eccezioni, mai blocca il loop.
* Serializza solo campi primitivi — `nodes` è una Map, va convertita.
*/
async function _checkpointGraph(graph: TaskGraph): Promise<void> {
try {
const db = await _getVfsDb();
if (!db) return;
const serialized = JSON.stringify({
id: graph.id,
goal: graph.goal,
createdAt: graph.createdAt,
nodes: [...graph.nodes.entries()].map(([id, node]) => { const { id: _nid, ...rest } = node as unknown as Record<string,unknown>; return { id, ...rest }; }),
edges: graph.edges,
});
// eslint-disable-next-line @typescript-eslint/no-explicit-any
await db?.taskGraphs?.put({
id: graph.id,
graphState: serialized,
goal: graph.goal.slice(0, 200),
timestamp: Date.now(),
status: "incomplete",
});
} catch {
// Dexie piena o schema non ancora aggiornato → silenzio
}
}
/**
* Marca il grafo come completato su Dexie (cleanup automatico post-run).
* In alternativa potrebbe eliminarlo — lo marchiamo "complete" per
* permettere analisi post-hoc senza occupare spazio indefinitamente.
* TTL: gestito da pruneCompletedGraphs() che può essere chiamata on-demand.
*/
async function _markGraphComplete(graphId: string): Promise<void> {
try {
const db = await _getVfsDb();
if (!db) return;
// eslint-disable-next-line @typescript-eslint/no-explicit-any
await db?.taskGraphs?.update(graphId, { status: "complete", completedAt: Date.now() });
} catch { /* non-blocking */ }
}
/**
* Deserializza un grafo da una stringa JSON (salvata da _checkpointGraph).
* Ricostruisce la Map<string, TaskNode> dal formato array serializzato.
*/
function _deserializeGraph(graphState: string): TaskGraph | null {
try {
const raw = JSON.parse(graphState) as {
id: string;
goal: string;
createdAt: number;
nodes: Array<TaskNode & { id: string }>;
edges: Array<{ from: string; to: string }>;
};
return {
id: raw.id,
goal: raw.goal,
createdAt: raw.createdAt,
nodes: new Map(raw.nodes.map(n => [n.id, n])),
edges: raw.edges,
};
} catch {
return null;
}
}
// ─── Public API: load incomplete graph on startup ─────────────────────────────
export interface IncompleteGraphInfo {
graphId: string;
goal: string;
timestamp: number;
}
/**
* Ritorna la lista di grafi incompleti salvati su Dexie.
* Chiamato da main.tsx all'avvio per offrire all'utente il resume.
* Non lancia mai eccezioni.
*/
export async function listIncompleteGraphs(): Promise<IncompleteGraphInfo[]> {
try {
const db = await _getVfsDb();
if (!db) return [];
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const rows = await db?.taskGraphs
?.where("status")
.equals("incomplete")
.sortBy("timestamp") ?? [];
return rows.map((r: { id: string; goal: string; timestamp: number }) => ({
graphId: r.id,
goal: r.goal,
timestamp: r.timestamp,
}));
} catch {
return [];
}
}
/**
* Carica e deserializza un grafo incompleto da Dexie dato il suo ID.
* Usato da GraphOrchestrator.run() in modalità resume.
*/
export async function loadIncompleteGraph(graphId: string): Promise<TaskGraph | null> {
try {
const db = await _getVfsDb();
if (!db) return null;
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const row = await db?.taskGraphs?.get(graphId);
if (!row?.graphState) return null;
return _deserializeGraph(row.graphState);
} catch {
return null;
}
}
/**
* Elimina i grafi completati più vecchi di TTL_MS.
* Chiamabile periodicamente o manualmente dall'utente.
*/
export async function pruneCompletedGraphs(ttlMs = 7 * 24 * 60 * 60_000): Promise<void> {
try {
const db = await _getVfsDb();
if (!db) return;
const cutoff = Date.now() - ttlMs;
// eslint-disable-next-line @typescript-eslint/no-explicit-any
await db?.taskGraphs
?.where("status")
.equals("complete")
.filter((r: { completedAt?: number }) => (r.completedAt ?? 0) < cutoff)
.delete();
} catch { /* non-blocking */ }
}
// ─── S760-D: ignoreGraph + resume singleton ──────────────────────────────────
/**
* Ignora definitivamente un grafo incompleto rimuovendolo da Dexie.
* Chiamato da ResumeGraphBanner quando l'utente clicca "Ignora".
* Fire-and-forget: mai lancia eccezioni.
*/
export async function ignoreGraph(graphId: string): Promise<void> {
try {
const db = await _getVfsDb();
if (!db) return;
// eslint-disable-next-line @typescript-eslint/no-explicit-any
await db?.taskGraphs?.delete(graphId);
} catch { /* non-blocking */ }
}
/**
* Modulo-level ref per il grafo parziale da riprendere.
* Impostato da ResumeGraphBanner via App.tsx onResume.
* Consumato (e azzerato) da graphOrchestrationBlock prima di planToGraph.
*/
let _pendingResumeGraph: TaskGraph | null = null;
/** Imposta il grafo parziale da riprendere al prossimo run(). */
export function setPendingResumeGraph(graph: TaskGraph): void {
_pendingResumeGraph = graph;
}
/** Preleva (e azzera) il grafo in attesa di resume. Usato da graphOrchestrationBlock. */
export function takePendingResumeGraph(): TaskGraph | null {
const g = _pendingResumeGraph;
_pendingResumeGraph = null;
return g;
}
// ─── Node executors ───────────────────────────────────────────────────────────
// Gap-1.2 fix (S723 + S724-EXT): enrich dependent node input with upstream results.
function enrichNodeInput(node: TaskNode, graph: TaskGraph): void {
if ((node as unknown as Record<string, unknown>)._enriched) return;
if (!node.deps?.length) return;
const ups = node.deps
.map(id => graph.nodes.get(id))
.filter((d): d is TaskNode => !!d && d.status === "done" && !!d.result)
// S760-B: un ref "@@dexie:" è un risultato grande — usiamo un placeholder
// invece di fetch asincrona (enrichNodeInput è sincrona per design)
.map(d => d.result!.startsWith(OFFLOAD_REF_PREFIX)
? "[risultato grande — vedi tool precedente]"
: d.result!);
if (!ups.length) return;
(node as unknown as Record<string, unknown>)._enriched = true;
const combined = ups.join("\n---\n").slice(0, 800);
const allText = ups.join(" ");
if (!node.input) return;
if ("query" in node.input) { node.input = { ...node.input, context: combined }; }
else if ("url" in node.input) {
const m = allText.match(/https?:\/\/[^\s"'<>]+/);
if (m) node.input = { ...node.input, url: m[0] };
} else if ("code" in node.input) {
const existing = String(node.input.code ?? "").trim();
const upSlice = combined.slice(0, 400).replace(/`{3}/g, "'''");
node.input = { ...node.input, code: `# --- upstream context ---\n# ${upSlice.replace(/\n/g, "\n# ")}\n# --- end context ---\n\n${existing}` };
} else if ("content" in node.input) {
const existing = String(node.input.content ?? "").trim();
const upSlice = combined.slice(0, 400);
node.input = { ...node.input, content: existing ? `${existing}\n\n# Dati upstream:\n# ${upSlice.replace(/\n/g, "\n# ")}` : upSlice };
} else if ("html" in node.input) {
const safe = combined.slice(0, 300).replace(/-->/g, "-- >");
node.input = { ...node.input, html: `<!-- upstream: ${safe} -->\n${String(node.input.html ?? "")}` };
} else if ("input" in node.input) {
node.input = { ...node.input, input: `${String(node.input.input ?? "")}\n\nContesto upstream:\n${combined.slice(0, 400)}` };
}
}
async function executeToolNode(
node: TaskNode,
executor: ToolExecutor,
signal: AbortSignal,
): Promise<string> {
if (!node.tool) throw new Error(`Nodo tool_call senza tool: ${node.id}`);
const timeout = getToolTimeout(node.tool);
const validatedInput = safeValidateInput(node.tool, node.input ?? {});
const ctrl = new AbortController();
const timer = setTimeout(() => ctrl.abort(), timeout);
const onParentAbort = () => ctrl.abort();
signal.addEventListener("abort", onParentAbort, { once: true });
try {
const result = await executor(node.tool, validatedInput);
clearTimeout(timer);
let annotated: string;
try {
const vr = verifyToolResult(node.tool, validatedInput, result);
annotated = annotateWithProof(node.tool, result, vr);
} catch {
annotated = result;
}
return annotated;
} finally {
clearTimeout(timer);
signal.removeEventListener("abort", onParentAbort);
}
}
async function executeVerifyNode(node: TaskNode, graph: TaskGraph): Promise<string> {
const toolNodeId = node.deps[0];
const toolNode = graph.nodes.get(toolNodeId);
if (!toolNode?.result) return "⚠️ Nessun risultato da verificare";
// S760-B: resolve ref Dexie se necessario
const realResult = toolNode.result.startsWith(OFFLOAD_REF_PREFIX)
? await _resolveOffloadedResult(toolNode.result)
: toolNode.result;
const vr = verifyToolResult(
node.tool ?? toolNode.tool ?? "",
toolNode.input ?? {},
realResult,
);
return vr.verified
? `✅ Verificato: ${vr.proof}`
: `⚠️ Non verificato: ${vr.warning ?? vr.proof}`;
}
async function executeMemoryNode(_node: TaskNode, graph: TaskGraph): Promise<string> {
const result = await updateMemoryFromGraph(graph);
return `✅ Memoria aggiornata: ${result.saved} fatti salvati`;
}
// ─── Main orchestrator ────────────────────────────────────────────────────────
export class GraphOrchestrator {
async run(
graph: TaskGraph,
executor: ToolExecutor,
callbacks: OrchestratorCallbacks = {},
signal?: AbortSignal,
): Promise<OrchestrationResult> {
let _running = 0;
const toolResults = new Map<string, string>();
const failedNodes: string[] = [];
let memoryUpdated = false;
callbacks.onStatus?.(`Avvio grafo: ${graph.goal.slice(0, 60)}…`);
// S760-D: checkpoint iniziale — registra il grafo come "incomplete" subito
// così se la tab viene killata prima del primo nodo completato, il grafo
// è già in Dexie e potrà essere ripreso.
await _checkpointGraph(graph).catch(() => {});
try {
while (!graphDone(graph) && !(signal?.aborted)) {
const ready = getReadyNodes(graph).sort(
(a, b) => (a.priority ?? 999) - (b.priority ?? 999)
);
if (ready.length === 0) break;
const batch = ready.slice(0, maxConcurrentFor(ready) - _running);
if (batch.length === 0) {
await new Promise(r => setTimeout(r, TICK_MS));
continue;
}
const promises = batch.map(async (node) => {
markNode(graph, node.id, "running");
callbacks.onNodeStart?.(node);
_running++;
try {
let result = "";
switch (node.type) {
case "plan":
result = graph.goal;
break;
case "tool_call": {
const _contract = TOOL_CONTRACTS[node.tool as keyof typeof TOOL_CONTRACTS];
const _contractMax = (_contract?.retryPolicy?.maxAttempts ?? 3) - 1;
const _MAX_RETRIES = Math.min(_contractMax, 2);
const _BACKOFF_MS = [300, 900] as const;
enrichNodeInput(node, graph);
let _lastErr: Error | undefined;
for (let _attempt = 0; _attempt <= _MAX_RETRIES; _attempt++) {
if (_attempt > 0) {
_running--;
try {
await new Promise(r => setTimeout(r, _BACKOFF_MS[_attempt - 1]));
} finally {
_running++;
}
if (signal?.aborted) break;
}
try {
result = await executeToolNode(node, executor, signal ?? new AbortController().signal);
_lastErr = undefined;
break;
} catch (err) {
_lastErr = err instanceof Error ? err : new Error(String(err));
if (!isTransientError(_lastErr.message) || signal?.aborted) break;
}
}
if (_lastErr) throw _lastErr;
break;
}
case "verify":
result = await executeVerifyNode(node, graph);
break;
case "memory":
result = await executeMemoryNode(node, graph);
memoryUpdated = true;
break;
case "synthesize":
// S760-B: collectResultsResolved risolve i ref Dexie
result = await collectResultsResolved(graph);
break;
default:
result = "";
}
// S647: hollow success guard
if (node.type === "tool_call" && (!result || !result.trim())) {
result = `⚠️ [${node.tool}] nessun output prodotto`;
}
// S760-B: offload su Dexie se risultato grande
let storedResult = result;
if (node.type === "tool_call" && result && !result.startsWith("⚠️")) {
storedResult = await _maybeOffloadResult(node.id, result);
}
markNode(graph, node.id, "done", storedResult);
if (node.type === "tool_call" && node.tool) {
const _prevCount = [...toolResults.keys()]
.filter(k => k === node.tool || k.startsWith(node.tool + ":")).length;
const _toolKey = _prevCount === 0
? node.tool!
: `${node.tool}:${_prevCount + 1}`;
toolResults.set(_toolKey, storedResult);
sessionMemory.add("tool", _toolKey, result.slice(0, 400)); // S609: 200→400
}
callbacks.onNodeDone?.(node);
} catch (err) {
const msg = err instanceof Error ? err.message : String(err);
markNode(graph, node.id, "failed", undefined, msg);
failedNodes.push(node.id);
callbacks.onNodeFailed?.(node);
// S760-C: adaptive replanning su nodo critico
if (
node.type === "tool_call" &&
(node.priority ?? 999) < CRITICAL_PRIORITY &&
!signal?.aborted
) {
const replanned = _adaptiveReplan(graph, node.id);
if (replanned) {
callbacks.onStatus?.(`↩ Nodo critico fallito — piano alternativo attivato`);
// Rimuovi il nodo dalla lista failedNodes poiché è stato recuperato
const fi = failedNodes.indexOf(node.id);
if (fi !== -1) failedNodes.splice(fi, 1);
}
}
} finally {
_running--;
}
const done2 = [...graph.nodes.values()]
.filter(n => n.status === "done" || n.status === "failed")
.length;
callbacks.onProgress?.(done2, graph.nodes.size);
// S760-D: checkpoint dopo ogni nodo completato/fallito
// fire-and-forget: non blocca il loop
_checkpointGraph(graph).catch(() => {});
// S723: checkpoint memory every 3 nodes
if (done2 > 0 && done2 % 3 === 0 && !memoryUpdated) {
updateMemoryFromGraph(graph).catch(() => {});
}
});
await Promise.all(promises);
}
} finally {
// S760-A: cleanup garantito — SEMPRE eseguito, anche su throw o abort
// Marca i nodi "running" come "failed" e propaga "skipped" ai dipendenti
// così lo stato finale del grafo è sempre coerente.
if (signal?.aborted) {
_cleanupDanglingNodes(graph, "utente ha annullato");
callbacks.onStatus?.("Task annullato");
} else {
// Anche senza abort, un break inaspettato dal while può lasciare running nodes
_cleanupDanglingNodes(graph, "loop terminato inaspettatamente");
}
}
// Memory update se non già fatto via nodo dedicato
if (!memoryUpdated) {
try {
await updateMemoryFromGraph(graph);
memoryUpdated = true;
} catch { /* non-blocking */ }
}
// S760-D: marca grafo come completato (cleanup Dexie)
await _markGraphComplete(graph.id).catch(() => {});
// S760-B: cleanup risultati offloadati su Dexie
// fire-and-forget: rimuove le entry graphToolResults di questa run
_getVfsDb().then(db => {
if (!db) return;
const idsToDelete = [...graph.nodes.values()]
.filter(n => n.result?.startsWith(OFFLOAD_REF_PREFIX))
.map(n => n.id);
if (idsToDelete.length > 0) {
// eslint-disable-next-line @typescript-eslint/no-explicit-any
db?.graphToolResults?.bulkDelete(idsToDelete).catch(() => {});
}
}).catch(() => {});
const skippedNodes = [...graph.nodes.values()]
.filter(n => n.status === "skipped")
.map(n => n.id);
const success = failedNodes.length === 0 && skippedNodes.length === 0;
const output = await collectResultsResolved(graph);
callbacks.onStatus?.("");
return { success, output, toolResults, failedNodes, skippedNodes, memoryUpdated };
}
}
export const graphOrchestrator = new GraphOrchestrator();
// ─── Graph-based planner (invariato) ─────────────────────────────────────────
export interface PlanLike {
goal: string;
complexity: string;
subtasks: Array<{ tool: string; description: string; risk: string }>;
}
let _currentEpPlanOrder: string[] = [];
export function setEpPlanOrder(order: string[]): void {
_currentEpPlanOrder = order;
}
const _FETCH_TOOLS_PG = new Set([
"web_search", "read_page", "get_news", "search_wikipedia",
"fetch_url", "get_stock", "get_currency", "search_github",
"web_research",
]);
const _CODE_TOOLS_PG = new Set([
"run_code", "write_file", "execute_shell",
"create_webpage", "iterate_file",
]);
const _WRITE_SEQ_TOOLS = new Set([
"write_file", "apply_patch", "iterate_file", "create_webpage",
]);
function extractToolArgs(tool: string, description?: string): Record<string, unknown> {
const desc = (typeof description === "string" ? description : "").trim() || tool;
if (["web_search","search_wikipedia","get_news","search_github","fetch_url","read_page","web_research"].includes(tool)) return { query: desc };
if (tool === "run_python") return { code: desc, language: "python" };
if (tool === "execute_shell") return { code: desc, language: "bash" };
if (tool === "run_code") {
const lang =
/\b(bash|shell|sh|zsh)\b/i.test(desc) ? "bash" :
/\b(javascript|js|node(\.?js)?|typescript|ts)\b/i.test(desc) ? "javascript" :
/\b(ruby|rb)\b/i.test(desc) ? "ruby" :
/\b(go|golang)\b/i.test(desc) ? "go" :
"python";
return { code: desc, language: lang };
}
if (tool === "write_file" || tool === "iterate_file") {
const _pm = desc.match(/\b([\w./\-]+\/[\w./\-]+\.[a-zA-Z]{1,10}|[\w\-]+\.[a-zA-Z]{1,10})\b/);
return { path: _pm?.[0] ?? "output.txt", content: desc };
}
if (tool === "apply_patch") {
const _pm = desc.match(/\b([\w./\-]+\/[\w./\-]+\.[a-zA-Z]{1,10}|[\w\-]+\.[a-zA-Z]{1,10})\b/);
return { path: _pm?.[0] ?? "output.txt", patch: desc };
}
if (tool === "create_webpage") return { html: desc };
if (["execute_sql","database_query"].includes(tool)) return { query: desc };
if (["generate_image"].includes(tool)) return { prompt: desc };
if (tool === "get_weather") {
const _cm = desc.match(/\b(?:a|in|per|for|at)\s+([A-Za-z\u00c0-\u00ff][A-Za-z\u00c0-\u00ff\s]{1,24}?)(?:\s*[?,.]|$|\s+(?:oggi|ora|adesso|now|today|domani|tomorrow))/i);
return { city: (_cm?.[1]?.trim() ?? desc.slice(0, 50)).trim() };
}
if (["get_stock","get_currency","get_datetime","get_ip_info","math_eval"].includes(tool)) return { query: desc };
if (tool === "call_api") {
const _urlM = desc.match(/https?:\/\/[^\s"'<>]+/i);
const _methM = desc.match(/\b(GET|POST|PUT|PATCH|DELETE)\b/i);
return { url: _urlM?.[0] ?? "", method: (_methM?.[1]?.toUpperCase() ?? "GET") as "GET"|"POST"|"PUT"|"PATCH"|"DELETE" };
}
if (tool === "send_email") {
const _toM = desc.match(/\b[\w.+\-]+@[\w.\-]+\.[a-z]{2,}\b/i);
const _subjM = desc.match(/(?:oggetto|subject|re)\s*[:\-]\s*([^\n]+)/i);
return { to: _toM?.[0] ?? "", subject: _subjM?.[1]?.trim() ?? desc.slice(0, 60), body: desc };
}
if (tool === "create_pdf") {
const _slug = desc.slice(0, 40).trim().toLowerCase()
.replace(/[^a-z0-9àèìòùáéíóú]+/g, "_").replace(/^_|_$/g, "").slice(0, 35) || "documento";
return { content: desc, filename: `${_slug}.pdf` };
}
return { input: desc };
}
export function planToGraph(plan: PlanLike): TaskGraph {
const _epOrderSnapshot = [..._currentEpPlanOrder];
_currentEpPlanOrder = [];
const steps = plan.subtasks.filter(s => s.tool !== "respond");
let _lastWriteIdx: number | undefined;
return buildGraph({
goal: plan.goal,
steps: steps.map((s, i, arr) => {
let dependsOn: number[] | undefined;
if (i > 0 && _CODE_TOOLS_PG.has(s.tool)) {
for (let j = i - 1; j >= 0; j--) {
if (_FETCH_TOOLS_PG.has(arr[j].tool)) { dependsOn = [j]; break; }
}
}
if (_WRITE_SEQ_TOOLS.has(s.tool)) {
if (_lastWriteIdx !== undefined) {
dependsOn = dependsOn
? [...new Set([...dependsOn, _lastWriteIdx])]
: [_lastWriteIdx];
}
_lastWriteIdx = i;
}
const _epPos = _epOrderSnapshot.length > 0
? (() => {
const label = s.description?.toLowerCase() ?? "";
const idx = _epOrderSnapshot.findIndex(
p => label.includes(p.toLowerCase()) || p.toLowerCase().includes(label.slice(0, 20))
);
return idx === -1 ? 999 : idx;
})()
: 999;
return {
tool: s.tool,
label: (s.description ?? "").slice(0, 150), // S609: 80→150
input: extractToolArgs(s.tool, s.description),
dependsOn,
verify: s.risk === "high",
priority: _epPos,
};
}),
synthesize: true,
memorize: plan.complexity !== "low",
});
}