|
| 1 | +/** |
| 2 | + * Subtask dispatcher (Phase 3). |
| 3 | + * |
| 4 | + * Executes a validated plan respecting its dependency DAG: |
| 5 | + * - subtasks are grouped into topological "levels" (Kahn's algorithm) |
| 6 | + * - subtasks in the same level have no dependency on each other → run in |
| 7 | + * parallel via the existing ParallelCoordinator (spawnParallel) |
| 8 | + * - a subtask receives ONLY its own prompt plus the compressed results of the |
| 9 | + * subtasks it depends on (context isolation — the token win) |
| 10 | + * |
| 11 | + * The spawn functions are injectable for testing. |
| 12 | + */ |
| 13 | + |
| 14 | +const logger = require("../../logger"); |
| 15 | + |
| 16 | +// Cap how much of a dependency's result we forward, to preserve the |
| 17 | +// context-isolation savings (subagents already return summaries; this bounds |
| 18 | +// pathological cases). |
| 19 | +const MAX_CONTEXT_CHARS = 2000; |
| 20 | + |
| 21 | +/** |
| 22 | + * Group subtasks into dependency levels. Returns array of arrays of ids. |
| 23 | + * Throws if the graph is unresolvable (should not happen — planner validated). |
| 24 | + */ |
| 25 | +function topologicalLevels(subtasks) { |
| 26 | + const byId = new Map(subtasks.map((s) => [s.id, s])); |
| 27 | + const indegree = new Map(subtasks.map((s) => [s.id, 0])); |
| 28 | + const dependents = new Map(subtasks.map((s) => [s.id, []])); |
| 29 | + |
| 30 | + for (const s of subtasks) { |
| 31 | + for (const dep of s.dependsOn) { |
| 32 | + if (!byId.has(dep)) continue; |
| 33 | + indegree.set(s.id, indegree.get(s.id) + 1); |
| 34 | + dependents.get(dep).push(s.id); |
| 35 | + } |
| 36 | + } |
| 37 | + |
| 38 | + const levels = []; |
| 39 | + let frontier = subtasks.filter((s) => indegree.get(s.id) === 0).map((s) => s.id); |
| 40 | + const resolved = new Set(); |
| 41 | + |
| 42 | + while (frontier.length > 0) { |
| 43 | + levels.push(frontier); |
| 44 | + const next = []; |
| 45 | + for (const id of frontier) { |
| 46 | + resolved.add(id); |
| 47 | + for (const child of dependents.get(id)) { |
| 48 | + indegree.set(child, indegree.get(child) - 1); |
| 49 | + if (indegree.get(child) === 0) next.push(child); |
| 50 | + } |
| 51 | + } |
| 52 | + frontier = next; |
| 53 | + } |
| 54 | + |
| 55 | + if (resolved.size !== subtasks.length) { |
| 56 | + throw new Error("Unresolvable subtask graph (cycle or dangling dependency)"); |
| 57 | + } |
| 58 | + return levels; |
| 59 | +} |
| 60 | + |
| 61 | +function compressResult(text) { |
| 62 | + if (typeof text !== "string") return String(text ?? ""); |
| 63 | + if (text.length <= MAX_CONTEXT_CHARS) return text; |
| 64 | + return text.slice(0, MAX_CONTEXT_CHARS) + "\n…[truncated]"; |
| 65 | +} |
| 66 | + |
| 67 | +function buildContextForSubtask(subtask, resultsById) { |
| 68 | + if (!subtask.dependsOn || subtask.dependsOn.length === 0) return null; |
| 69 | + const parts = []; |
| 70 | + for (const dep of subtask.dependsOn) { |
| 71 | + const r = resultsById.get(dep); |
| 72 | + if (r && r.success && r.result) { |
| 73 | + parts.push(`Result of subtask ${dep}:\n${compressResult(r.result)}`); |
| 74 | + } else if (r) { |
| 75 | + parts.push(`Subtask ${dep} did not complete successfully.`); |
| 76 | + } |
| 77 | + } |
| 78 | + return parts.length > 0 ? parts.join("\n\n") : null; |
| 79 | +} |
| 80 | + |
| 81 | +/** |
| 82 | + * Dispatch a validated plan. |
| 83 | + * @param {Object} plan - { subtasks: [...] } |
| 84 | + * @param {Object} [options] |
| 85 | + * @param {string} [options.sessionId] |
| 86 | + * @param {string} [options.cwd] |
| 87 | + * @param {Function} [options.spawnParallel] - (agentTypes[], prompts[], opts) => results[] |
| 88 | + * @returns {Promise<{results: Array, levels: Array, stats: Object}>} |
| 89 | + */ |
| 90 | +async function dispatchPlan(plan, options = {}) { |
| 91 | + const spawnParallel = options.spawnParallel || require("../index").spawnParallel; |
| 92 | + const subtasks = plan.subtasks; |
| 93 | + const byId = new Map(subtasks.map((s) => [s.id, s])); |
| 94 | + const levels = topologicalLevels(subtasks); |
| 95 | + const resultsById = new Map(); |
| 96 | + |
| 97 | + let totalInputTokens = 0; |
| 98 | + let totalOutputTokens = 0; |
| 99 | + let totalSubagents = 0; |
| 100 | + |
| 101 | + for (let li = 0; li < levels.length; li++) { |
| 102 | + const levelIds = levels[li]; |
| 103 | + const levelSubtasks = levelIds.map((id) => byId.get(id)); |
| 104 | + |
| 105 | + const agentTypes = levelSubtasks.map((s) => s.agentType); |
| 106 | + const prompts = levelSubtasks.map((s) => s.prompt); |
| 107 | + const perTaskContext = levelSubtasks.map((s) => buildContextForSubtask(s, resultsById)); |
| 108 | + |
| 109 | + logger.info( |
| 110 | + { level: li, count: levelIds.length, ids: levelIds }, |
| 111 | + "[Decomposition] Dispatching subtask level" |
| 112 | + ); |
| 113 | + |
| 114 | + // spawnParallel shares one options object; pass per-task context by spawning |
| 115 | + // the level as individual parallel calls when contexts differ. |
| 116 | + const levelResults = await runLevel( |
| 117 | + spawnParallel, |
| 118 | + agentTypes, |
| 119 | + prompts, |
| 120 | + perTaskContext, |
| 121 | + options |
| 122 | + ); |
| 123 | + |
| 124 | + levelResults.forEach((res, idx) => { |
| 125 | + const st = levelSubtasks[idx]; |
| 126 | + totalSubagents += 1; |
| 127 | + totalInputTokens += res?.stats?.inputTokens || 0; |
| 128 | + totalOutputTokens += res?.stats?.outputTokens || 0; |
| 129 | + resultsById.set(st.id, { |
| 130 | + id: st.id, |
| 131 | + agentType: st.agentType, |
| 132 | + success: !!res?.success, |
| 133 | + result: res?.success ? res.result : null, |
| 134 | + error: res?.success ? null : res?.error || "unknown error", |
| 135 | + stats: res?.stats || {}, |
| 136 | + }); |
| 137 | + }); |
| 138 | + } |
| 139 | + |
| 140 | + const results = subtasks.map((s) => resultsById.get(s.id)); |
| 141 | + return { |
| 142 | + results, |
| 143 | + levels, |
| 144 | + stats: { |
| 145 | + subagents: totalSubagents, |
| 146 | + inputTokens: totalInputTokens, |
| 147 | + outputTokens: totalOutputTokens, |
| 148 | + }, |
| 149 | + }; |
| 150 | +} |
| 151 | + |
| 152 | +/** |
| 153 | + * Run one level. When subtasks in the level have differing injected contexts we |
| 154 | + * spawn them as separate parallel calls (each with its own mainContext), then |
| 155 | + * await all. When none need context, a single spawnParallel batch is used. |
| 156 | + */ |
| 157 | +async function runLevel(spawnParallel, agentTypes, prompts, perTaskContext, options) { |
| 158 | + const anyContext = perTaskContext.some((c) => c); |
| 159 | + |
| 160 | + if (!anyContext) { |
| 161 | + return spawnParallel(agentTypes, prompts, { |
| 162 | + sessionId: options.sessionId, |
| 163 | + cwd: options.cwd, |
| 164 | + }); |
| 165 | + } |
| 166 | + |
| 167 | + // Mixed/with-context: one spawnParallel call per subtask so each gets its own |
| 168 | + // mainContext, executed concurrently. |
| 169 | + const calls = agentTypes.map((type, i) => |
| 170 | + spawnParallel([type], [prompts[i]], { |
| 171 | + sessionId: options.sessionId, |
| 172 | + cwd: options.cwd, |
| 173 | + mainContext: perTaskContext[i] ? { relevant_context: perTaskContext[i] } : undefined, |
| 174 | + }).then((arr) => arr[0]) |
| 175 | + ); |
| 176 | + return Promise.all(calls); |
| 177 | +} |
| 178 | + |
| 179 | +module.exports = { |
| 180 | + dispatchPlan, |
| 181 | + topologicalLevels, |
| 182 | + buildContextForSubtask, |
| 183 | + compressResult, |
| 184 | + MAX_CONTEXT_CHARS, |
| 185 | +}; |
0 commit comments