root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(FN-3767): add task runtime env injection into executor commands via pl

Browse Source 
Adds executor runtime environment support to the plugin system, enabling plugins to contribute environment variables that get injected into executor commands. The implementation includes new plugin types (`ExecutorRuntimeEnvPlugin`), aggregation in the plugin runner, thread-through into executor com

Fusion-Task-Id: FN-3767
This commit is contained in:
Fusion
2026-05-10 19:28:57 -07:00
committed by gsxdsm
parent a0c7c3315a
commit a04b3205b4
15 changed files with 624 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
packages/engine/src/__tests__/executor-core.test.ts
View File

@@ -1226,6 +1226,7 @@ describe("Executor verification gate (FN-3345)", () => {
undefined,
expect.anything(),
"executor",
expect.any(Object),
);
// Task should move to in-review
expect(store.moveTask).toHaveBeenCalledWith("FN-3345", "in-review");
@@ -1402,6 +1403,7 @@ describe("Executor verification gate (FN-3345)", () => {
undefined,
expect.anything(),
"executor",
expect.any(Object),
);
// Third call should be build
expect(mockedVerification).toHaveBeenNthCalledWith(
@@ -1414,6 +1416,7 @@ describe("Executor verification gate (FN-3345)", () => {
undefined,
expect.anything(),
"executor",
expect.any(Object),
);
// Task should move to in-review
expect(store.moveTask).toHaveBeenCalledWith("FN-3345", "in-review");

144
packages/engine/src/__tests__/executor-runtime-env.test.ts Normal file
View File

@@ -0,0 +1,144 @@
import { describe, it, expect, vi, beforeEach } from "vitest";
import type { FusionPlugin, PluginLoader, PluginStore, TaskStore } from "@fusion/core";
import { PluginRunner } from "../plugin-runner.js";
import { createLogger } from "../logger.js";
vi.mock("../logger.js", () => ({
createLogger: vi.fn(() => ({ log: vi.fn(), warn: vi.fn(), error: vi.fn() })),
executorLog: { log: vi.fn(), warn: vi.fn(), error: vi.fn() },
}));
describe("PluginRunner.collectExecutorRuntimeEnv", () => {
const createMockPlugin = (id: string, executorRuntimeEnv?: FusionPlugin["executorRuntimeEnv"]): FusionPlugin => ({
manifest: { id, name: id, version: "1.0.0" },
state: "started",
hooks: {},
executorRuntimeEnv,
});
const createRunner = (plugins: FusionPlugin[]) => {
const pluginLoader = {
getLoadedPlugins: vi.fn().mockReturnValue(plugins),
getPlugin: vi.fn(),
getPluginTools: vi.fn().mockReturnValue([]),
getPluginRoutes: vi.fn().mockReturnValue([]),
getPluginUiSlots: vi.fn().mockReturnValue([]),
getPluginUiContributions: vi.fn().mockReturnValue([]),
getPluginRuntimes: vi.fn().mockReturnValue([]),
getCliProviderContributions: vi.fn().mockReturnValue([]),
getPluginSkills: vi.fn().mockReturnValue([]),
getPluginWorkflowSteps: vi.fn().mockReturnValue([]),
getPluginWorkflowStepTemplates: vi.fn().mockReturnValue([]),
getPluginPromptContributions: vi.fn().mockReturnValue([]),
getPluginSetupInfo: vi.fn().mockReturnValue([]),
on: vi.fn(),
off: vi.fn(),
invokeHook: vi.fn(),
loadAllPlugins: vi.fn(),
stopAllPlugins: vi.fn(),
getPluginSchemaInitHooks: vi.fn().mockReturnValue([]),
checkPluginSetup: vi.fn(),
installPluginSetup: vi.fn(),
uninstallPluginSetup: vi.fn(),
loadPlugin: vi.fn(),
stopPlugin: vi.fn(),
reloadPlugin: vi.fn(),
} as unknown as PluginLoader;
const pluginStore = {
getPlugin: vi.fn(async (pluginId: string) => ({ id: pluginId, settings: {} })),
on: vi.fn(),
off: vi.fn(),
} as unknown as PluginStore;
const taskStore = {
on: vi.fn(),
off: vi.fn(),
getDatabase: vi.fn(),
} as unknown as TaskStore;
return new PluginRunner({ pluginLoader, pluginStore, taskStore, rootDir: "/repo" });
};
beforeEach(() => {
vi.clearAllMocks();
});
it("returns empty env/path when no plugins contribute", async () => {
const runner = createRunner([]);
const result = await runner.collectExecutorRuntimeEnv({ taskId: "FN-1", worktreePath: "/tmp/wt", rootDir: "/repo" });
expect(result).toEqual({ env: {}, pathPrepend: [], perPluginErrors: [] });
});
it("collects env/path from one plugin", async () => {
const runner = createRunner([
createMockPlugin("plugin-a", () => ({ pathPrepend: ["/opt/plugin-a/bin"], env: { A_TOKEN: "a" } })),
]);
const result = await runner.collectExecutorRuntimeEnv({ taskId: "FN-1", worktreePath: "/tmp/wt", rootDir: "/repo" });
expect(result.env).toEqual({ A_TOKEN: "a" });
expect(result.pathPrepend).toEqual(["/opt/plugin-a/bin"]);
expect(result.perPluginErrors).toEqual([]);
});
it("merges plugins with later env overriding and later path entries first", async () => {
const runner = createRunner([
createMockPlugin("plugin-a", () => ({ pathPrepend: ["/opt/a/bin"], env: { SHARED: "a", A_ONLY: "1" } })),
createMockPlugin("plugin-b", () => ({ pathPrepend: ["/opt/b/bin"], env: { SHARED: "b", B_ONLY: "1" } })),
]);
const result = await runner.collectExecutorRuntimeEnv({ taskId: "FN-1", worktreePath: "/tmp/wt", rootDir: "/repo" });
expect(result.env).toEqual({ SHARED: "b", A_ONLY: "1", B_ONLY: "1" });
expect(result.pathPrepend).toEqual(["/opt/b/bin", "/opt/a/bin"]);
});
it("records per-plugin errors when plugin throws", async () => {
const runner = createRunner([
createMockPlugin("plugin-ok", () => ({ env: { OK: "1" } })),
createMockPlugin("plugin-bad", () => {
throw new Error("boom");
}),
]);
const result = await runner.collectExecutorRuntimeEnv({ taskId: "FN-1", worktreePath: "/tmp/wt", rootDir: "/repo" });
expect(result.env).toEqual({ OK: "1" });
expect(result.perPluginErrors).toHaveLength(1);
expect(result.perPluginErrors[0]?.pluginId).toBe("plugin-bad");
expect(result.perPluginErrors[0]?.error.message).toContain("boom");
});
it("records schema errors for invalid contribution shapes", async () => {
const runner = createRunner([
createMockPlugin("plugin-path", () => ({ pathPrepend: ["relative/bin"] })),
createMockPlugin("plugin-path-env", () => ({ env: { PATH: "forbidden" } })),
createMockPlugin("plugin-value", () => ({ env: { GOOD: "ok", BAD: 42 as unknown as string } })),
]);
const result = await runner.collectExecutorRuntimeEnv({ taskId: "FN-1", worktreePath: "/tmp/wt", rootDir: "/repo" });
expect(result.env).toEqual({});
expect(result.pathPrepend).toEqual([]);
expect(result.perPluginErrors).toHaveLength(3);
expect(result.perPluginErrors.map((item) => item.pluginId)).toEqual([
"plugin-path",
"plugin-path-env",
"plugin-value",
]);
});
it("logs warnings when env keys are overridden", async () => {
const runner = createRunner([
createMockPlugin("plugin-a", () => ({ env: { SHARED: "a" } })),
createMockPlugin("plugin-b", () => ({ env: { SHARED: "b" } })),
]);
await runner.collectExecutorRuntimeEnv({ taskId: "FN-1", worktreePath: "/tmp/wt", rootDir: "/repo" });
const logger = vi.mocked(createLogger).mock.results.at(-1)?.value as { warn: ReturnType<typeof vi.fn> };
expect(logger.warn).toHaveBeenCalledWith(expect.stringContaining("key override: SHARED"));
});
});

55
packages/engine/src/executor.ts
View File

@@ -2,7 +2,7 @@ import { exec } from "node:child_process";
import { promisify } from "node:util";
const execAsync = promisify(exec);
import { isAbsolute, join, relative, resolve as resolvePath } from "node:path";
import { delimiter, isAbsolute, join, relative, resolve as resolvePath } from "node:path";
import { existsSync } from "node:fs";
import { readFile, writeFile } from "node:fs/promises";
import type { TaskStore, Task, TaskDetail, TaskTokenUsage, StepStatus, Settings, WorkflowStep, MissionStore, Slice, AgentState, AgentCapability, RunMutationContext, AgentHeartbeatConfig, Agent } from "@fusion/core";
@@ -172,6 +172,7 @@ async function runConfiguredCommand(
command: string,
cwd: string,
timeoutMs: number,
extraEnv?: NodeJS.ProcessEnv,
): Promise<RunCommandResult> {
try {
const { stdout, stderr } = await execAsync(command, {
@@ -179,6 +180,7 @@ async function runConfiguredCommand(
timeout: timeoutMs,
maxBuffer: 10 * 1024 * 1024,
encoding: "utf-8",
...(extraEnv !== undefined && { env: extraEnv }),
});
return {
@@ -1914,7 +1916,7 @@ export class TaskExecutor {
if (await this.shouldDeferCompletionForGlobalPause(task.id, "before workflow steps during completed-task recovery")) {
return false;
}
const workflowResult = await this.runWorkflowSteps(task, task.worktree, settings);
const workflowResult = await this.runWorkflowSteps(task, task.worktree, settings, undefined);
if (workflowResult === "deferred-paused") {
if (this.pausedAborted.has(task.id)) {
this.pausedAborted.delete(task.id);
@@ -2296,6 +2298,7 @@ export class TaskExecutor {
let stuckRequeue: boolean | null = null;
let taskDone = false;
let reviewAddressingActivated = false;
let taskEnv: NodeJS.ProcessEnv | undefined;
try {
await this.transitionReviewAddressing(task.id, ["queued"], "in-progress");
@@ -2415,7 +2418,7 @@ export class TaskExecutor {
if (settings.worktreeInitCommand) {
const initStartedAt = Date.now();
try {
const initResult = await runConfiguredCommand(settings.worktreeInitCommand, worktreePath, 300_000);
const initResult = await runConfiguredCommand(settings.worktreeInitCommand, worktreePath, 300_000, taskEnv);
if (initResult.spawnError || initResult.timedOut || initResult.exitCode !== 0) {
throw new Error(configuredCommandErrorMessage(initResult));
}
@@ -2442,7 +2445,7 @@ export class TaskExecutor {
if (scriptCommand) {
const setupStartedAt = Date.now();
try {
const setupResult = await runConfiguredCommand(scriptCommand, worktreePath, 120_000);
const setupResult = await runConfiguredCommand(scriptCommand, worktreePath, 120_000, taskEnv);
if (setupResult.spawnError || setupResult.timedOut || setupResult.exitCode !== 0) {
throw new Error(configuredCommandErrorMessage(setupResult));
}
@@ -2496,6 +2499,26 @@ export class TaskExecutor {
this.activeWorktrees.set(task.id, worktreePath);
executorLog.log(`${task.id}: worktree ready at ${worktreePath}`);
const runtimeEnvContribution = await this.options.pluginRunner?.collectExecutorRuntimeEnv({
taskId: task.id,
worktreePath,
rootDir: this.rootDir,
branch: task.branch ?? undefined,
});
const pathPrepend = runtimeEnvContribution?.pathPrepend ?? [];
const injectedEnv = runtimeEnvContribution?.env ?? {};
// We intentionally do NOT mutate process.env globally. Agent session subprocesses
// currently inherit the engine process env only; piping taskEnv into AgentRuntimeOptions
// is tracked as follow-up work.
taskEnv = {
...process.env,
...injectedEnv,
PATH: [...pathPrepend, process.env.PATH ?? ""].filter(Boolean).join(delimiter),
};
executorLog.log(
`${task.id}: executor runtime env injected (${pathPrepend.length} PATH entries, ${Object.keys(injectedEnv).length} env keys)`,
);
this.options.onStart?.(task, worktreePath);
const detail = await this.store.getTask(task.id);
@@ -2559,6 +2582,7 @@ export class TaskExecutor {
// Pass agentStore and messageStore for delegation and messaging tools
agentStore: this.options.agentStore,
messageStore: this.options.messageStore,
taskEnv,
onStepStart: (stepIndex) => {
this.options.stuckTaskDetector?.recordProgress(task.id);
try {
@@ -2650,7 +2674,7 @@ export class TaskExecutor {
// and when no verification commands are configured.
if (executionMode !== "fast") {
if (settings.testCommand?.trim() || settings.buildCommand?.trim()) {
const verificationResult = await this.runExecutorDeterministicVerification(task, worktreePath, settings);
const verificationResult = await this.runExecutorDeterministicVerification(task, worktreePath, settings, taskEnv);
if (!verificationResult.allPassed) {
const failedType = verificationResult.failedCommand === "testCommand" ? "test" : "build";
@@ -2695,6 +2719,7 @@ export class TaskExecutor {
settings,
attempt,
maxFixRetries,
taskEnv,
);
if (fixed) {
fixSucceeded = true;
@@ -2734,7 +2759,7 @@ export class TaskExecutor {
// Run workflow steps before moving to in-review — skip in fast mode
if (executionMode !== "fast") {
const workflowResult = await this.runWorkflowSteps(task, worktreePath, settings);
const workflowResult = await this.runWorkflowSteps(task, worktreePath, settings, taskEnv);
if (workflowResult === "deferred-paused") {
if (await this.parkTaskAfterWorkflowStepPause(task.id)) {
this.pausedAborted.delete(task.id);
@@ -3348,7 +3373,7 @@ export class TaskExecutor {
// Run workflow steps before moving to in-review — skip in fast mode
if (executionMode !== "fast") {
const workflowResult = await this.runWorkflowSteps(task, worktreePath, settings);
const workflowResult = await this.runWorkflowSteps(task, worktreePath, settings, taskEnv);
if (workflowResult === "deferred-paused") {
if (await this.parkTaskAfterWorkflowStepPause(task.id)) {
this.pausedAborted.delete(task.id);
@@ -3537,7 +3562,7 @@ export class TaskExecutor {
// Run workflow steps before moving to in-review — skip in fast mode
if (executionMode !== "fast") {
const workflowResult = await this.runWorkflowSteps(task, worktreePath, settings);
const workflowResult = await this.runWorkflowSteps(task, worktreePath, settings, taskEnv);
if (workflowResult === "deferred-paused") {
if (await this.parkTaskAfterWorkflowStepPause(task.id)) {
this.pausedAborted.delete(task.id);
@@ -4754,6 +4779,7 @@ ${feedback}
task: Task,
worktreePath: string,
settings: Settings,
extraEnv?: NodeJS.ProcessEnv,
): Promise<VerificationResult> {
const testCommand = settings.testCommand?.trim();
const buildCommand = settings.buildCommand?.trim();
@@ -4779,7 +4805,7 @@ ${feedback}
// Run test command first if configured
if (testCommand) {
const testResult = await runVerificationCommand(
this.store, worktreePath, task.id, testCommand, "test", undefined, executorLog, "executor",
this.store, worktreePath, task.id, testCommand, "test", undefined, executorLog, "executor", extraEnv,
);
result.testResult = testResult;
@@ -4794,7 +4820,7 @@ ${feedback}
// Run build command second if configured
if (buildCommand) {
const buildResult = await runVerificationCommand(
this.store, worktreePath, task.id, buildCommand, "build", undefined, executorLog, "executor",
this.store, worktreePath, task.id, buildCommand, "build", undefined, executorLog, "executor", extraEnv,
);
result.buildResult = buildResult;
@@ -4833,6 +4859,7 @@ ${feedback}
settings: Settings,
retryNumber: number,
maxRetries: number,
extraEnv?: NodeJS.ProcessEnv,
): Promise<boolean> {
try {
executorLog.log(`${task.id}: spawning executor verification fix agent (attempt ${retryNumber}/${maxRetries})`);
@@ -4961,7 +4988,7 @@ ${failureContext.output.slice(0, VERIFICATION_LOG_MAX_CHARS)}
undefined,
"executor",
);
const reRunResult = await this.runExecutorDeterministicVerification(task, worktreePath, settings);
const reRunResult = await this.runExecutorDeterministicVerification(task, worktreePath, settings, extraEnv);
return reRunResult.allPassed;
} finally {
@@ -5254,6 +5281,7 @@ ${failureFeedback}
task: Task,
worktreePath: string,
settings: Settings,
taskEnv?: NodeJS.ProcessEnv,
): Promise<WorkflowStepResult | "deferred-paused"> {
// Check if task has enabled workflow steps
const currentTask = await this.store.getTask(task.id);
@@ -5340,7 +5368,7 @@ ${failureFeedback}
try {
const result: WorkflowStepOutcome = stepMode === "script"
? await this.executeScriptWorkflowStep(task, ws, worktreePath, settings)
? await this.executeScriptWorkflowStep(task, ws, worktreePath, settings, taskEnv)
: await this.executeWorkflowStep(task, ws, worktreePath, settings);
if (await this.shouldDeferWorkflowStepCompletion(task.id, `workflow step '${ws.name}'`)) {
return "deferred-paused";
@@ -5459,6 +5487,7 @@ ${failureFeedback}
workflowStep: WorkflowStep,
worktreePath: string,
settings: Settings,
extraEnv?: NodeJS.ProcessEnv,
): Promise<{ success: boolean; output?: string; error?: string }> {
const scriptName = workflowStep.scriptName!.trim();
const scriptCommand = settings.scripts?.[scriptName];
@@ -5474,7 +5503,7 @@ ${failureFeedback}
await this.store.logEntry(task.id, `Workflow step '${workflowStep.name}' executing script '${scriptName}': ${scriptCommand}`);
try {
const scriptResult = await runConfiguredCommand(scriptCommand, worktreePath, 120_000);
const scriptResult = await runConfiguredCommand(scriptCommand, worktreePath, 120_000, extraEnv);
if (scriptResult.spawnError || scriptResult.timedOut || scriptResult.exitCode !== 0) {
return { success: false, error: configuredCommandErrorMessage(scriptResult) };
}

74
packages/engine/src/plugin-runner.ts
View File

@@ -27,9 +27,11 @@ import type {
PluginSetupManifest,
PluginSetupHooks,
PluginSetupCheckResult,
ExecutorRuntimeTaskContext,
} from "@fusion/core";
import type { ToolDefinition } from "@mariozechner/pi-coding-agent";
import { Type } from "@mariozechner/pi-ai";
import { isAbsolute } from "node:path";
import { createLogger, executorLog } from "./logger.js";
// Type for the task store's event data
@@ -440,6 +442,78 @@ export class PluginRunner {
);
}
async collectExecutorRuntimeEnv(taskCtx: ExecutorRuntimeTaskContext): Promise<{
env: Record<string, string>;
pathPrepend: string[];
perPluginErrors: Array<{ pluginId: string; error: Error }>;
}> {
const loadedPlugins = this.options.pluginLoader.getLoadedPlugins();
const pluginResults: Array<{ pluginId: string; env: Record<string, string>; pathPrepend: string[] }> = [];
const perPluginErrors: Array<{ pluginId: string; error: Error }> = [];
for (const plugin of loadedPlugins) {
if (!plugin.executorRuntimeEnv) {
continue;
}
const pluginId = plugin.manifest.id;
try {
const settings = await this.getPluginSettings(pluginId);
const context: PluginContext = {
pluginId,
taskStore: this.options.taskStore,
settings,
logger: this.createPluginLogger(pluginId),
emitEvent: (event: string, data: unknown) => {
this.log.log(`[plugin:${pluginId}] Event: ${event}`, data);
},
};
const contribution = await plugin.executorRuntimeEnv(taskCtx, context);
const env = contribution.env ?? {};
const pathPrepend = contribution.pathPrepend ?? [];
if (!Array.isArray(pathPrepend) || pathPrepend.some((entry) => typeof entry !== "string" || !isAbsolute(entry))) {
throw new Error("executorRuntimeEnv.pathPrepend must be an array of absolute path strings");
}
for (const [key, value] of Object.entries(env)) {
if (key === "PATH") {
throw new Error("executorRuntimeEnv.env must not contain PATH; use pathPrepend instead");
}
if (typeof value !== "string") {
throw new Error(`executorRuntimeEnv.env.${key} must be a string`);
}
}
pluginResults.push({ pluginId, env, pathPrepend });
} catch (error) {
const normalizedError = error instanceof Error ? error : new Error(String(error));
perPluginErrors.push({ pluginId, error: normalizedError });
this.log.warn(`executorRuntimeEnv failed for plugin ${pluginId}: ${normalizedError.message}`);
}
}
const mergedEnv: Record<string, string> = {};
const mergedPathPrepend: string[] = [];
for (const result of pluginResults) {
for (const [key, value] of Object.entries(result.env)) {
if (Object.prototype.hasOwnProperty.call(mergedEnv, key)) {
this.log.warn(`executorRuntimeEnv key override: ${key} overwritten by plugin ${result.pluginId}`);
}
mergedEnv[key] = value;
}
mergedPathPrepend.unshift(...result.pathPrepend);
}
return {
env: mergedEnv,
pathPrepend: mergedPathPrepend,
perPluginErrors,
};
}
getPromptContributionsForSurface(surface: PluginPromptSurface): Array<{
pluginId: string;
contribution: PluginPromptContribution;

4
packages/engine/src/step-session-executor.ts
View File

@@ -115,6 +115,8 @@ export interface StepSessionExecutorOptions {
actionGateContext?: AgentActionGateContext;
/** Optional permanent-agent action gating context. */
permanentAgentGating?: PermanentAgentGatingContext;
/** Task-scoped environment injected into non-git subprocesses. */
taskEnv?: NodeJS.ProcessEnv;
}
// ── File Scope Extraction ─────────────────────────────────────────────
@@ -1283,7 +1285,7 @@ Follow instructions precisely and avoid unrelated changes.`,
// Remove any partial directory left behind so the invariant holds:
// "if .worktrees/<slug> exists on disk, it is a fully registered git worktree."
try {
await execAsync(`rm -rf "${worktreePath}"`, { cwd: rootDir });
await execAsync(`rm -rf "${worktreePath}"`, { cwd: rootDir, env: this.options.taskEnv });
} catch {
// best-effort cleanup; log but don't mask the original error
stepExecLog.log(`Warning: failed to remove partial worktree directory after creation failure: ${worktreePath}`);