From ed208d56e42eda9caa644f6d4316e24869e3f7df Mon Sep 17 00:00:00 2001 From: artale Date: Mon, 15 Jun 2026 13:25:27 +0200 Subject: [PATCH] feat: route fanout subtasks by flow --- src/fable5/compound-stack.ts | 4 +- src/fable5/dynamic-workflows.test.ts | 52 +++++++++++++++++++++++ src/fable5/dynamic-workflows.ts | 63 ++++++++++++++++++++++++++-- src/fable5/index.ts | 12 +++++- src/index.ts | 18 ++++++-- 5 files changed, 138 insertions(+), 11 deletions(-) create mode 100644 src/fable5/dynamic-workflows.test.ts diff --git a/src/fable5/compound-stack.ts b/src/fable5/compound-stack.ts index 7addc79..5bccaca 100644 --- a/src/fable5/compound-stack.ts +++ b/src/fable5/compound-stack.ts @@ -3,7 +3,7 @@ import { IndependentVerifier, type VerificationResult } from "./independent-veri import { FiveStageStateFile, type FiveStageState } from "./state-file-5stage.js"; import { GoalPattern, type GoalDefinition } from "./goal-pattern.js"; import { WorktreeManager } from "./worktree-isolation.js"; -import { DynamicWorkflows, type AdversarialResult, type FanOutResult, type FusionPanelResult } from "./dynamic-workflows.js"; +import { DynamicWorkflows, type AdversarialResult, type FanOutExecutor, type FanOutResult, type FusionPanelResult } from "./dynamic-workflows.js"; import { ProxyClient } from "../pai/proxy-client.js"; import { SkillSync } from "../pai/skill-sync.js"; import type { LoopIteration } from "../core/types.js"; @@ -376,7 +376,7 @@ export class CompoundStack { async fanOut( task: string, subTasks: string[], - executor: (subTask: string, index: number) => LoopIteration | Promise, + executor: FanOutExecutor, synthesizer: (results: LoopIteration[], originalTask: string) => LoopIteration | Promise, maxConcurrency?: number, ): Promise { diff --git a/src/fable5/dynamic-workflows.test.ts b/src/fable5/dynamic-workflows.test.ts new file mode 100644 index 0000000..1bda773 --- /dev/null +++ b/src/fable5/dynamic-workflows.test.ts @@ -0,0 +1,52 @@ +import { describe, expect, it } from "vitest"; +import { DynamicWorkflows } from "./dynamic-workflows.js"; +import type { LoopIteration } from "../core/types.js"; + +const iterationFor = (label: string, index: number): LoopIteration => ({ + number: index + 1, + phase: "execute", + plan: `Execute: ${label}`, + executed: `Done: ${label}`, + observation: `Result: ${label}`, + reflection: `Done ${index + 1}`, + refinement: "", + metrics: { durationMs: 100, successRate: 0.9, qualityScore: 0.8, improvementDelta: 0.05 }, + timestamp: new Date().toISOString(), +}); + +describe("DynamicWorkflows fan-out routing", () => { + it("classifies sub-tasks into direct/planning/review flow buckets", async () => { + const wf = new DynamicWorkflows(); + const subtasks = ["Add tests", "Write architecture plan", "Run final review checks"]; + + const result = await wf.fanOutAndSynthesize( + "Harden login flow for service", + subtasks, + (subTask, index, routeContext) => { + expect(routeContext).toBeDefined(); + expect(routeContext?.index).toBe(index); + expect(routeContext?.subTask).toBe(subTask); + + return iterationFor(subTask, index); + }, + (results, _task): LoopIteration => ({ + number: 999, + phase: "refine", + plan: `Synthesis for ${results.length}`, + executed: results.map((r) => r.executed).join(" | "), + observation: "Synthesis done", + reflection: "Synthesis complete", + refinement: "", + metrics: { durationMs: 40, successRate: 0.95, qualityScore: 0.9, improvementDelta: 0.2 }, + timestamp: new Date().toISOString(), + }), + ); + + expect(result.routeContexts).toHaveLength(3); + expect(result.routeContexts[0]?.flow).toBe("direct"); + expect(result.routeContexts[1]?.flow).toBe("planning"); + expect(result.routeContexts[2]?.flow).toBe("review"); + expect(result.iterations).toHaveLength(3); + expect(result.synthesisVerdict).toBeDefined(); + }); +}); diff --git a/src/fable5/dynamic-workflows.ts b/src/fable5/dynamic-workflows.ts index 8713433..5d622c5 100644 --- a/src/fable5/dynamic-workflows.ts +++ b/src/fable5/dynamic-workflows.ts @@ -1,6 +1,6 @@ import type { LoopIteration, GodModeRaceResult, PerturbationResult } from "../core/types.js"; import { IndependentVerifier, type VerificationResult } from "./independent-verifier.js"; -import { ModelRouter, type TaskComplexity } from "./model-router.js"; +import { ModelRouter, type TaskComplexity, type RoutingResponse } from "./model-router.js"; import { type FusionPanelSlug, type FusionResult, @@ -13,6 +13,22 @@ import { Parseltongue } from "../upgrades/parseltongue.js"; // ── Types ────────────────────────────────────────────────── export type WorkflowPattern = "fan-out-synthesize" | "adversarial-verify" | "loop-until-done" | "fusion-panel" | "godmode-race" | "parseltongue-test"; +export type FanOutSubtaskFlow = "direct" | "planning" | "review"; + +export interface FanOutSubtaskRouteContext { + task: string; + subTask: string; + index: number; + flow: FanOutSubtaskFlow; + reason: string; + route: RoutingResponse; +} + +export type FanOutExecutor = ( + subTask: string, + index: number, + routeContext?: FanOutSubtaskRouteContext, +) => LoopIteration | Promise; export interface WorkflowStep { id: string; @@ -31,6 +47,7 @@ export interface WorkflowDefinition { export interface FanOutResult { iterations: LoopIteration[]; + routeContexts: FanOutSubtaskRouteContext[]; synthesized: LoopIteration; synthesisVerdict: VerificationResult; } @@ -96,15 +113,21 @@ export class DynamicWorkflows { async fanOutAndSynthesize( task: string, subTasks: string[], - executor: (subTask: string, index: number) => LoopIteration | Promise, + executor: FanOutExecutor, synthesizer: (results: LoopIteration[], originalTask: string) => LoopIteration | Promise, maxConcurrency: number = 3, ): Promise { // Phase 1: Fan out — run all sub-tasks (with concurrency limit) + const routeContexts: FanOutSubtaskRouteContext[] = new Array(subTasks.length); const iterations = await this.mapWithConcurrency( subTasks.length, maxConcurrency, - (index) => Promise.resolve(executor(subTasks[index], index)), + (index) => { + const subTask = subTasks[index]; + const routeContext = this.routeFanOutSubtask(task, subTask, index); + routeContexts[index] = routeContext; + return Promise.resolve(executor(subTask, index, routeContext)); + }, ); // Phase 2: Synthesize @@ -113,7 +136,7 @@ export class DynamicWorkflows { // Phase 3: Verify the synthesis const synthesisVerdict = this.verifier.verify(synthesized, task); - return { iterations, synthesized, synthesisVerdict }; + return { iterations, routeContexts, synthesized, synthesisVerdict }; } /** @@ -292,6 +315,38 @@ export class DynamicWorkflows { return await pt.testGate(input, gateFn, options); } + private routeFanOutSubtask(task: string, subTask: string, index: number): FanOutSubtaskRouteContext { + const lower = `${subTask} ${task}`.toLowerCase(); + const isPlanning = /\b(plan|design|architecture|strategy|roadmap|approach|proposal|scope|spec|mvp)\b/i.test(lower); + const isReview = /\b(review|audit|check|verify|validation|assessment|safety|test|diff)\b/i.test(lower); + + const flow: FanOutSubtaskFlow = isPlanning + ? "planning" + : isReview + ? "review" + : "direct"; + + const complexity = flow === "direct" ? "simple" : "medium"; + const domain: "code" | "planning" | "review" = + flow === "planning" ? "planning" : flow === "review" ? "review" : "code"; + + const route = this.modelRouter.route({ + task: `${task} ${subTask}`, + domain, + complexity, + requiresVision: false, + }); + + return { + task, + subTask, + index, + flow, + reason: flow === "direct" ? "Direct/simple fanout -> cheap flow" : `Plan/review fanout -> routed as ${flow}`, + route, + }; + } + private async mapWithConcurrency( count: number, maxConcurrency: number, diff --git a/src/fable5/index.ts b/src/fable5/index.ts index 5284a44..5c68aa3 100644 --- a/src/fable5/index.ts +++ b/src/fable5/index.ts @@ -17,7 +17,17 @@ export { CompoundStack } from "./compound-stack.js"; export type { StackConfig, StackStatus } from "./compound-stack.js"; export { DynamicWorkflows } from "./dynamic-workflows.js"; -export type { WorkflowPattern, WorkflowStep, WorkflowDefinition, FanOutResult, AdversarialResult, LoopUntilDoneResult } from "./dynamic-workflows.js"; +export type { + WorkflowPattern, + WorkflowStep, + WorkflowDefinition, + FanOutResult, + FanOutSubtaskFlow, + FanOutSubtaskRouteContext, + FanOutExecutor, + AdversarialResult, + LoopUntilDoneResult, +} from "./dynamic-workflows.js"; export { AgentChain } from "./agent-chains.js"; export type { ChainStage, ChainConfig, StageArtifact, ChainResult } from "./agent-chains.js"; diff --git a/src/index.ts b/src/index.ts index 439e7b3..9c0e1c5 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1658,14 +1658,14 @@ fable const subTasks = opts.subtasks.split(",").map((s) => s.trim()); const wf = new DynamicWorkflows(); - const executor = (subTask: string, _i: number): LoopIteration => ({ + const executor = (subTask: string, _i: number, routeContext?: { flow?: "direct" | "planning" | "review"; reason?: string }): LoopIteration => ({ number: _i + 1, phase: "execute" as const, plan: `Executing sub-task: ${subTask}`, executed: `Completed: ${subTask}`, observation: `Sub-task results: ${subTask}`, - reflection: `Sub-task ${_i + 1} done`, - refinement: "", + reflection: `Sub-task ${_i + 1} done${routeContext?.flow ? ` (${routeContext.flow})` : ""}`, + refinement: routeContext?.reason ?? "", metrics: { durationMs: 100, successRate: 0.9, qualityScore: 0.7, improvementDelta: 0.05 }, timestamp: new Date().toISOString(), }); @@ -1689,6 +1689,16 @@ fable console.log(` Sub-tasks: ${subTasks.length}`); console.log(` Iterations: ${result.iterations.length}`); console.log(` Synthesis verdict: ${result.synthesisVerdict.verdict}`); + if (result.routeContexts.length > 0) { + const counts = result.routeContexts.reduce( + (acc, rc) => { + acc[rc.flow] = (acc[rc.flow] ?? 0) + 1; + return acc; + }, + {} as Record, + ); + console.log(` Route flows: direct=${counts.direct ?? 0}, planning=${counts.planning ?? 0}, review=${counts.review ?? 0}`); + } console.log(` `); if (result.synthesisVerdict.gaps.length > 0) { console.log(` Gaps:`); @@ -2594,7 +2604,7 @@ security if (/\.(test|spec)\.(ts|js)$/i.test(file)) return; const rel = path.relative(root, file).replace(/\\/g, "/"); // ponytail: default scan ignores our own intentional prompt-injection generator; use --include-fixtures for audit mode. - if (!opts.includeFixtures && rel === "src/upgrades/parseltongue.ts") return; + if (!opts.includeFixtures && rel.endsWith("upgrades/parseltongue.ts")) return; if (/\.(md|txt|ts|js|json|yaml|yml)$/i.test(file)) files.push(file); };