Spaces:
Sleeping
Sleeping
| /** | |
| * 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", | |
| }); | |
| } | |