feat(engine): custom workflow nodes run as their column's agent
U3: per-run IR resolution feeds the core column-agent resolver at the runCustomNode seam; override supersedes node agent/model/persona wholesale, defer fills bare nodes only; adoption and fallback are audited via logEntry; raw-CLI nodes log a skip. Also fixes the customInstructions persona drift — node-level executor:"agent" persona injection now uses the typed soul/instructionsText fields (KTD-6).
This commit is contained in:
gsxdsm
@@ -0,0 +1,217 @@
|
||||
// Column-agent custom-node resolution (plan U3, R2/R3/R4/R8, KTD-2/KTD-3/KTD-6).
|
||||
//
|
||||
// `runGraphCustomNode` synthesizes a `WorkflowStep` and runs it on the proven
|
||||
// WorkflowStep machinery. The seam wiring (maybeExecuteWorkflowGraph) resolves
|
||||
// the per-node column-agent binding and threads it in as a parameter. These
|
||||
// tests call `runGraphCustomNode` directly with that binding and assert the
|
||||
// synthesized step's model/persona plus the audit log entries — mirroring the
|
||||
// established executor harness (executor-workflow-step-scope.test.ts): build a
|
||||
// real TaskExecutor over a mock store and spy on `executeWorkflowStep` /
|
||||
// `executeScriptWorkflowStep` to capture the synthesized step.
|
||||
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import "./executor-test-helpers.js";
|
||||
import { TaskExecutor } from "../executor.js";
|
||||
import { createMockStore, resetExecutorMocks } from "./executor-test-helpers.js";
|
||||
import type { WorkflowColumnAgent } from "@fusion/core";
|
||||
|
||||
function makeAgent(overrides: Record<string, unknown> = {}) {
|
||||
return {
|
||||
id: "agent-col",
|
||||
name: "Column Agent",
|
||||
soul: "I am the senior reviewer.",
|
||||
instructionsText: "Always be thorough.",
|
||||
runtimeConfig: { executorProvider: "anthropic", executorModelId: "claude-col" },
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
function makeExecutor(store: ReturnType<typeof createMockStore>, agent: unknown | null) {
|
||||
const agentStore = {
|
||||
getAgent: vi.fn().mockResolvedValue(agent),
|
||||
};
|
||||
const executor = new TaskExecutor(store as any, "/tmp/test", { agentStore } as any);
|
||||
return { executor, agentStore };
|
||||
}
|
||||
|
||||
/** Spy both session-running paths; return the captured synthesized step. */
|
||||
function spyStep(executor: TaskExecutor) {
|
||||
const captured: { step?: any } = {};
|
||||
vi.spyOn(executor as any, "executeWorkflowStep").mockImplementation(async (...args: any[]) => {
|
||||
captured.step = args[1];
|
||||
return { success: true, output: "ok" };
|
||||
});
|
||||
vi.spyOn(executor as any, "executeScriptWorkflowStep").mockImplementation(async (...args: any[]) => {
|
||||
captured.step = args[1];
|
||||
return { success: true, output: "ok" };
|
||||
});
|
||||
return captured;
|
||||
}
|
||||
|
||||
function loggedLines(store: ReturnType<typeof createMockStore>): string[] {
|
||||
return store.logEntry.mock.calls.map((call: any[]) => String(call[1] ?? ""));
|
||||
}
|
||||
|
||||
const OVERRIDE: WorkflowColumnAgent = { agentId: "agent-col", mode: "override" };
|
||||
const DEFER: WorkflowColumnAgent = { agentId: "agent-col", mode: "defer" };
|
||||
|
||||
describe("runGraphCustomNode column-agent resolution (plan U3)", () => {
|
||||
beforeEach(() => {
|
||||
resetExecutorMocks();
|
||||
});
|
||||
|
||||
it("override column: node with own cfg.agentId runs as column agent (model+persona) and logs substitution+mode", async () => {
|
||||
const store = createMockStore();
|
||||
store.getTask.mockResolvedValue({ id: "FN-001", worktree: "/tmp/wt", log: [] } as any);
|
||||
const { executor, agentStore } = makeExecutor(store, makeAgent());
|
||||
const captured = spyStep(executor);
|
||||
|
||||
const node = {
|
||||
id: "review",
|
||||
kind: "prompt",
|
||||
column: "review",
|
||||
config: {
|
||||
executor: "agent",
|
||||
agentId: "node-own-agent",
|
||||
modelProvider: "openai",
|
||||
modelId: "gpt-node",
|
||||
prompt: "Review the diff.",
|
||||
},
|
||||
};
|
||||
|
||||
const result = await (executor as any).runGraphCustomNode(node, { id: "FN-001" }, {}, OVERRIDE);
|
||||
|
||||
expect(result.outcome).toBe("success");
|
||||
// Column agent fetched (not the node's own agent).
|
||||
expect(agentStore.getAgent).toHaveBeenCalledWith("agent-col");
|
||||
// Column agent's model wins over the node's own pair.
|
||||
expect(captured.step.modelProvider).toBe("anthropic");
|
||||
expect(captured.step.modelId).toBe("claude-col");
|
||||
// Column agent's persona (soul + instructionsText) prepended to the prompt.
|
||||
expect(captured.step.prompt).toContain("I am the senior reviewer.");
|
||||
expect(captured.step.prompt).toContain("Always be thorough.");
|
||||
expect(captured.step.prompt).toContain("Review the diff.");
|
||||
// Audit log records substitution + mode.
|
||||
expect(
|
||||
loggedLines(store).some((l) => l.includes("running as column agent 'agent-col' (override)")),
|
||||
).toBe(true);
|
||||
});
|
||||
|
||||
it("defer column: node with own cfg.agentId keeps it; bare node adopts the column agent", async () => {
|
||||
// (a) own agentId present → defer yields own settings, column agent untouched.
|
||||
{
|
||||
const store = createMockStore();
|
||||
store.getTask.mockResolvedValue({ id: "FN-001", worktree: "/tmp/wt", log: [] } as any);
|
||||
const nodeOwnAgent = makeAgent({ id: "node-own-agent", soul: "node persona", instructionsText: "", runtimeConfig: { executorProvider: "openai", executorModelId: "gpt-node" } });
|
||||
const { executor, agentStore } = makeExecutor(store, nodeOwnAgent);
|
||||
const captured = spyStep(executor);
|
||||
|
||||
const node = {
|
||||
id: "review",
|
||||
kind: "prompt",
|
||||
column: "review",
|
||||
config: { executor: "agent", agentId: "node-own-agent", prompt: "Do it." },
|
||||
};
|
||||
await (executor as any).runGraphCustomNode(node, { id: "FN-001" }, {}, DEFER);
|
||||
|
||||
// Own agent fetched, NOT the column agent.
|
||||
expect(agentStore.getAgent).toHaveBeenCalledWith("node-own-agent");
|
||||
expect(agentStore.getAgent).not.toHaveBeenCalledWith("agent-col");
|
||||
expect(captured.step.modelProvider).toBe("openai");
|
||||
expect(captured.step.modelId).toBe("gpt-node");
|
||||
expect(
|
||||
loggedLines(store).some((l) => l.includes("running as column agent")),
|
||||
).toBe(false);
|
||||
}
|
||||
|
||||
// (b) bare node (no own agent/model) → defer adopts the column agent.
|
||||
{
|
||||
const store = createMockStore();
|
||||
store.getTask.mockResolvedValue({ id: "FN-001", worktree: "/tmp/wt", log: [] } as any);
|
||||
const { executor, agentStore } = makeExecutor(store, makeAgent());
|
||||
const captured = spyStep(executor);
|
||||
|
||||
const node = { id: "review", kind: "prompt", column: "review", config: { prompt: "Plain." } };
|
||||
await (executor as any).runGraphCustomNode(node, { id: "FN-001" }, {}, DEFER);
|
||||
|
||||
expect(agentStore.getAgent).toHaveBeenCalledWith("agent-col");
|
||||
expect(captured.step.modelProvider).toBe("anthropic");
|
||||
expect(captured.step.modelId).toBe("claude-col");
|
||||
expect(captured.step.prompt).toContain("I am the senior reviewer.");
|
||||
expect(
|
||||
loggedLines(store).some((l) => l.includes("running as column agent 'agent-col' (defer)")),
|
||||
).toBe(true);
|
||||
}
|
||||
});
|
||||
|
||||
it("missing column agent in registry → logged, node falls back, step still executes", async () => {
|
||||
const store = createMockStore();
|
||||
store.getTask.mockResolvedValue({ id: "FN-001", worktree: "/tmp/wt", log: [] } as any);
|
||||
// agentStore returns null for the column agent.
|
||||
const { executor } = makeExecutor(store, null);
|
||||
const captured = spyStep(executor);
|
||||
|
||||
const node = { id: "review", kind: "prompt", column: "review", config: { prompt: "Plain." } };
|
||||
const result = await (executor as any).runGraphCustomNode(node, { id: "FN-001" }, {}, OVERRIDE);
|
||||
|
||||
expect(result.outcome).toBe("success");
|
||||
// No column-agent model adopted (agent missing) → step has no model pair.
|
||||
expect(captured.step.modelProvider).toBeUndefined();
|
||||
expect(captured.step.modelId).toBeUndefined();
|
||||
expect(
|
||||
loggedLines(store).some((l) => l.includes("column agent 'agent-col' not found")),
|
||||
).toBe(true);
|
||||
});
|
||||
|
||||
it("node with no declared column → untouched resolution even when a binding is passed as undefined", async () => {
|
||||
const store = createMockStore();
|
||||
store.getTask.mockResolvedValue({ id: "FN-001", worktree: "/tmp/wt", log: [] } as any);
|
||||
const { executor, agentStore } = makeExecutor(store, makeAgent());
|
||||
const captured = spyStep(executor);
|
||||
|
||||
// No declared column → the seam wiring resolves no binding (undefined).
|
||||
const node = {
|
||||
id: "review",
|
||||
kind: "prompt",
|
||||
config: { executor: "model", modelProvider: "openai", modelId: "gpt-node", prompt: "Plain." },
|
||||
};
|
||||
await (executor as any).runGraphCustomNode(node, { id: "FN-001" }, {}, undefined);
|
||||
|
||||
// Column agent never fetched; node's own model preserved.
|
||||
expect(agentStore.getAgent).not.toHaveBeenCalled();
|
||||
expect(captured.step.modelProvider).toBe("openai");
|
||||
expect(captured.step.modelId).toBe("gpt-node");
|
||||
expect(loggedLines(store).some((l) => l.includes("column agent"))).toBe(false);
|
||||
});
|
||||
|
||||
it("CLI-executor node (raw command) in override column → mechanics unchanged, audit notes the skip", async () => {
|
||||
const store = createMockStore();
|
||||
store.getTask.mockResolvedValue({ id: "FN-001", worktree: "/tmp/wt", log: [] } as any);
|
||||
store.isWorkflowCliCommandApproved = vi.fn().mockResolvedValue(true);
|
||||
const { executor, agentStore } = makeExecutor(store, makeAgent());
|
||||
// Raw CLI runs runRawCliCommand, not a session — stub it.
|
||||
const rawSpy = vi.spyOn(executor as any, "runRawCliCommand").mockResolvedValue({ success: true });
|
||||
|
||||
const node = {
|
||||
id: "lint",
|
||||
kind: "script",
|
||||
column: "review",
|
||||
config: { executor: "cli", cliCommand: "npm run lint", cliSkipApproval: true, prompt: "" },
|
||||
};
|
||||
const result = await (executor as any).runGraphCustomNode(node, { id: "FN-001" }, {}, OVERRIDE);
|
||||
|
||||
expect(result.outcome).toBe("success");
|
||||
// Raw CLI mechanics unchanged: command still ran.
|
||||
expect(rawSpy).toHaveBeenCalled();
|
||||
// Column agent NOT fetched/adopted for raw CLI execution.
|
||||
expect(agentStore.getAgent).not.toHaveBeenCalled();
|
||||
// Audit explains the skip.
|
||||
expect(
|
||||
loggedLines(store).some(
|
||||
(l) =>
|
||||
l.includes("column agent 'agent-col' (override) not applied") &&
|
||||
l.includes("raw CLI execution runs no session"),
|
||||
),
|
||||
).toBe(true);
|
||||
});
|
||||
});
|
||||
@@ -9,8 +9,8 @@ import { delimiter, isAbsolute, join, relative, resolve as resolvePath } from "n
|
||||
import { existsSync, realpathSync } from "node:fs";
|
||||
import { readFile, rm, writeFile } from "node:fs/promises";
|
||||
import type { TaskStore, Task, TaskDetail, TaskTokenUsage, StepStatus, Settings, WorkflowStep, MissionStore, Slice, AgentState, AgentCapability, RunMutationContext, AgentHeartbeatConfig, Agent, AgentMemoryInclusionMode, ProjectSettings, MergeResult, WorkflowIrNode } from "@fusion/core";
|
||||
import { RetryStormError, TaskDeletedError, serializeRetryStormError, isExperimentalFeatureEnabled, resolveWorkflowIrForTask } from "@fusion/core";
|
||||
import type { TaskStep, WorkflowIr, WorkflowFieldDefinition } from "@fusion/core";
|
||||
import { RetryStormError, TaskDeletedError, serializeRetryStormError, isExperimentalFeatureEnabled, resolveWorkflowIrForTask, resolveColumnAgentBinding, resolveEffectiveAgent } from "@fusion/core";
|
||||
import type { TaskStep, WorkflowIr, WorkflowFieldDefinition, WorkflowColumnAgent } from "@fusion/core";
|
||||
import {
|
||||
buildWorkflowObservationFromTask,
|
||||
buildWorkflowObservation,
|
||||
@@ -3320,11 +3320,27 @@ export class TaskExecutor {
|
||||
// Definition load failure — leave undefined; deps/runner use fallbacks.
|
||||
}
|
||||
|
||||
// Column-agent binding (plan U3): the IR is NOT in scope inside
|
||||
// runGraphCustomNode, so resolve it here (the seam wiring) where the
|
||||
// selection is known, and thread a per-node binding lookup into the custom
|
||||
// node callback. Resolve the IR ONCE per run (never an uncached per-node
|
||||
// fetch — mirrors the hold-release.ts irCache posture); best-effort, so a
|
||||
// resolution failure simply yields no bindings (R8 graceful degradation).
|
||||
let columnAgentIr: WorkflowIr | undefined;
|
||||
try {
|
||||
columnAgentIr = await resolveWorkflowIrForTask(this.store, task.id);
|
||||
} catch {
|
||||
columnAgentIr = undefined;
|
||||
}
|
||||
const resolveBindingForNode = (nodeId: string): WorkflowColumnAgent | undefined =>
|
||||
columnAgentIr ? resolveColumnAgentBinding(columnAgentIr, nodeId) : undefined;
|
||||
|
||||
const runner = new WorkflowGraphTaskRunner({
|
||||
store: this.store,
|
||||
runId: resolvedRunId,
|
||||
seams: this.createGraphSeams(settings),
|
||||
runCustomNode: (node, nodeTask) => this.runGraphCustomNode(node, nodeTask, settings),
|
||||
runCustomNode: (node, nodeTask) =>
|
||||
this.runGraphCustomNode(node, nodeTask, settings, resolveBindingForNode(node.id)),
|
||||
onEvent: (event) => executorLog.log(`[workflow-graph] ${event.type} ${event.taskId}: ${event.detail}`),
|
||||
// Wire SQLite-backed per-branch persistence in production (#1407): the
|
||||
// executor writes each branch's currentNodeId/status to
|
||||
@@ -4495,11 +4511,77 @@ export class TaskExecutor {
|
||||
}
|
||||
}
|
||||
|
||||
/** Run a custom (non-seam) graph node on the proven WorkflowStep machinery. */
|
||||
/** Build the persona prefix for an agent from its TYPED identity fields (KTD-6).
|
||||
* Reads `soul` and `instructionsText` — the fields the `Agent` type actually
|
||||
* exposes (`packages/core/src/types.ts`) — and joins them. The custom-node
|
||||
* `"agent"` branch historically read a non-existent `customInstructions`
|
||||
* field (silently undefined); this is the single consistent source used by
|
||||
* both the node-agent and column-agent paths. */
|
||||
private buildAgentPersona(agent: Agent): string | undefined {
|
||||
const parts = [agent.soul, agent.instructionsText]
|
||||
.map((p) => (typeof p === "string" ? p.trim() : ""))
|
||||
.filter((p) => p.length > 0);
|
||||
return parts.length > 0 ? parts.join("\n\n") : undefined;
|
||||
}
|
||||
|
||||
/** Fetch the column agent and surface its model + persona for adoption by a
|
||||
* custom node (plan U3). Best-effort, mirroring the node-agent posture at the
|
||||
* `"agent"` branch: on null/throw, log and return undefined so the caller
|
||||
* falls back to the node's own/default resolution (R8). Emits a logEntry
|
||||
* naming the substitution and mode so the audit trail explains who ran. */
|
||||
private async adoptColumnAgentForNode(
|
||||
node: WorkflowIrNode,
|
||||
live: TaskDetail,
|
||||
columnAgentId: string,
|
||||
mode: WorkflowColumnAgent["mode"] | undefined,
|
||||
): Promise<{ modelProvider?: string; modelId?: string; persona?: string } | undefined> {
|
||||
try {
|
||||
const agent = await this.options.agentStore?.getAgent(columnAgentId);
|
||||
if (!agent) {
|
||||
await this.store.logEntry(
|
||||
live.id,
|
||||
`Workflow node '${node.id}': column agent '${columnAgentId}' not found — falling back to node/default resolution`,
|
||||
undefined,
|
||||
this.getRunContextFor(live.id),
|
||||
);
|
||||
return undefined;
|
||||
}
|
||||
const rc = (agent.runtimeConfig ?? {}) as { executorProvider?: string; executorModelId?: string };
|
||||
await this.store.logEntry(
|
||||
live.id,
|
||||
`Workflow node '${node.id}': running as column agent '${columnAgentId}' (${mode})`,
|
||||
undefined,
|
||||
this.getRunContextFor(live.id),
|
||||
);
|
||||
return {
|
||||
modelProvider: rc.executorProvider,
|
||||
modelId: rc.executorModelId,
|
||||
persona: this.buildAgentPersona(agent),
|
||||
};
|
||||
} catch {
|
||||
// Agent lookup is best-effort; fall back to node/default resolution (R8).
|
||||
await this.store.logEntry(
|
||||
live.id,
|
||||
`Workflow node '${node.id}': column agent '${columnAgentId}' lookup failed — falling back to node/default resolution`,
|
||||
undefined,
|
||||
this.getRunContextFor(live.id),
|
||||
);
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
/** Run a custom (non-seam) graph node on the proven WorkflowStep machinery.
|
||||
*
|
||||
* `columnBinding` (plan U3) is the agent binding governing this node's
|
||||
* declared column, resolved by the seam wiring in maybeExecuteWorkflowGraph
|
||||
* (the IR is not in scope here). When present, the core resolver decides
|
||||
* whether the column agent supersedes (override) or defers to the node's own
|
||||
* `cfg.agentId`/model pair — never a reimplemented precedence. */
|
||||
private async runGraphCustomNode(
|
||||
node: WorkflowIrNode,
|
||||
nodeTask: TaskDetail,
|
||||
settings: Settings,
|
||||
columnBinding?: WorkflowColumnAgent,
|
||||
): Promise<WorkflowNodeResult> {
|
||||
const cfg = node.config ?? {};
|
||||
const live = await this.store.getTask(nodeTask.id);
|
||||
@@ -4536,6 +4618,51 @@ export class TaskExecutor {
|
||||
let modelProvider = typeof cfg.modelProvider === "string" && cfg.modelProvider.trim() ? cfg.modelProvider : undefined;
|
||||
let modelId = typeof cfg.modelId === "string" && cfg.modelId.trim() ? cfg.modelId : undefined;
|
||||
|
||||
// ── Column-agent binding (plan U3, KTD-2/KTD-3) ──────────────────────────
|
||||
// When the node's declared column names an agent, the CORE resolver decides
|
||||
// whether the column agent supersedes (override) or defers to the node's own
|
||||
// settings — we never reimplement precedence. The node's own `cfg.agentId`
|
||||
// and complete model pair feed the resolver as "own settings" (KTD-5).
|
||||
const ownModelComplete = Boolean(modelProvider && modelId);
|
||||
const effective = resolveEffectiveAgent({
|
||||
binding: columnBinding,
|
||||
ownAgentId: typeof cfg.agentId === "string" && cfg.agentId.trim() ? cfg.agentId.trim() : undefined,
|
||||
ownModelProvider: ownModelComplete ? modelProvider : undefined,
|
||||
ownModelId: ownModelComplete ? modelId : undefined,
|
||||
});
|
||||
// The effective executor identity: a column agent supersedes the node's own
|
||||
// `executor: "agent"` adoption wholesale (identity + model + persona). When
|
||||
// the resolver yields the column agent, we run the column-agent adoption
|
||||
// path below INSTEAD of the node's own agent branch.
|
||||
const columnAgentId = effective.source === "column-agent" ? effective.agentId : undefined;
|
||||
const columnAgentMode = columnBinding?.mode;
|
||||
|
||||
if (columnAgentId) {
|
||||
// CLI executor with a raw command runs no session — the column agent
|
||||
// cannot contribute a model/persona to raw process execution, so it is a
|
||||
// no-op here. Log the skip so the audit trail explains why the column
|
||||
// agent did not apply (plan U3). Skill / model / script-via-session nodes
|
||||
// DO adopt the column agent below.
|
||||
if (executorKind === "cli" && rawCliCommand) {
|
||||
await this.store.logEntry(
|
||||
live.id,
|
||||
`Workflow node '${node.id}': column agent '${columnAgentId}' (${columnAgentMode}) not applied — raw CLI execution runs no session`,
|
||||
undefined,
|
||||
this.getRunContextFor(live.id),
|
||||
);
|
||||
} else {
|
||||
const adopted = await this.adoptColumnAgentForNode(node, live, columnAgentId, columnAgentMode);
|
||||
if (adopted) {
|
||||
modelProvider = adopted.modelProvider ?? modelProvider;
|
||||
modelId = adopted.modelId ?? modelId;
|
||||
if (adopted.persona) prompt = `${adopted.persona}\n\n${prompt}`;
|
||||
}
|
||||
// Whether or not the agent resolved, the column agent SUPERSEDES the
|
||||
// node's own `executor: "agent"` adoption — skip that branch so we never
|
||||
// blend the column agent's model with the node agent's persona.
|
||||
}
|
||||
}
|
||||
|
||||
// Executor kinds for prompt nodes:
|
||||
// - "model" (default): run the prompt on the configured/override model.
|
||||
// - "agent": run as a named agent — adopt its model and persona prompt.
|
||||
@@ -4543,14 +4670,18 @@ export class TaskExecutor {
|
||||
// - "cli": run a named project script with the prompt passed via env
|
||||
// (FUSION_NODE_PROMPT). Named scripts only — raw commands are
|
||||
// never accepted from node config.
|
||||
if (executorKind === "agent" && typeof cfg.agentId === "string" && cfg.agentId.trim()) {
|
||||
if (!columnAgentId && executorKind === "agent" && typeof cfg.agentId === "string" && cfg.agentId.trim()) {
|
||||
try {
|
||||
const agent = await this.options.agentStore?.getAgent(cfg.agentId);
|
||||
if (agent) {
|
||||
const rc = (agent.runtimeConfig ?? {}) as { executorProvider?: string; executorModelId?: string };
|
||||
modelProvider = rc.executorProvider ?? modelProvider;
|
||||
modelId = rc.executorModelId ?? modelId;
|
||||
const persona = (agent as { customInstructions?: string }).customInstructions;
|
||||
// KTD-6: read the TYPED persona fields (soul / instructionsText), not
|
||||
// the non-existent `customInstructions` (which was silently undefined,
|
||||
// so node-agent persona injection never actually fired). Same fields
|
||||
// the column-agent path uses — one consistent persona source.
|
||||
const persona = this.buildAgentPersona(agent);
|
||||
if (persona) prompt = `${persona}\n\n${prompt}`;
|
||||
} else {
|
||||
await this.store.logEntry(live.id, `Workflow node '${node.id}': agent '${cfg.agentId}' not found — using default model`, undefined, this.getRunContextFor(live.id));
|
||||
|
||||
Reference in New Issue
Block a user