feat: route fanout subtasks by flow
This commit is contained in:
parent
a664863485
commit
ed208d56e4
|
|
@ -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<LoopIteration>,
|
||||
executor: FanOutExecutor,
|
||||
synthesizer: (results: LoopIteration[], originalTask: string) => LoopIteration | Promise<LoopIteration>,
|
||||
maxConcurrency?: number,
|
||||
): Promise<FanOutResult> {
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
});
|
||||
});
|
||||
|
|
@ -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<LoopIteration>;
|
||||
|
||||
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<LoopIteration>,
|
||||
executor: FanOutExecutor,
|
||||
synthesizer: (results: LoopIteration[], originalTask: string) => LoopIteration | Promise<LoopIteration>,
|
||||
maxConcurrency: number = 3,
|
||||
): Promise<FanOutResult> {
|
||||
// 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<T>(
|
||||
count: number,
|
||||
maxConcurrency: number,
|
||||
|
|
|
|||
|
|
@ -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";
|
||||
|
|
|
|||
18
src/index.ts
18
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<string, number>,
|
||||
);
|
||||
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);
|
||||
};
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue