From d88bfc4c75594880b2eea4a609b8563e1754a7e8 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Thu, 4 Jun 2026 23:53:22 -0700 Subject: [PATCH] feat(engine): custom workflow nodes run as their column's agent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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). --- .../executor-column-agent-custom-node.test.ts | 217 ++++++++++++++++++ packages/engine/src/executor.ts | 143 +++++++++++- 2 files changed, 354 insertions(+), 6 deletions(-) create mode 100644 packages/engine/src/__tests__/executor-column-agent-custom-node.test.ts diff --git a/packages/engine/src/__tests__/executor-column-agent-custom-node.test.ts b/packages/engine/src/__tests__/executor-column-agent-custom-node.test.ts new file mode 100644 index 0000000000..c7a854c3d2 --- /dev/null +++ b/packages/engine/src/__tests__/executor-column-agent-custom-node.test.ts @@ -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 = {}) { + 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, 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): 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); + }); +}); diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 3a29414187..e1f3028606 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -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 { 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));