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

feat(KB-229): add multi-step scheduled tasks

Browse Source 
- Add Step types to core data model (command, validation, approval steps)
- Update AutomationStore CRUD operations for step management
- Refactor CronRunner for sequential step execution with per-step config
- Add Dashboard API routes for step CRUD and step-aware schedule updates
- Create ScheduleStepsEditor component for visual step configuration
- Add StepTypeBadge component for step type visualization
- Update ScheduleCard and ScheduleForm to display and manage steps
- Add comprehensive test coverage for stores, runner, and components
- Add changeset for patch release
This commit is contained in:
gsxdsm
2026-03-31 05:14:30 -07:00
parent 62f87947f6
commit 7569af8a93
16 changed files with 2127 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files

269
packages/engine/src/cron-runner.test.ts
View File

@@ -1,7 +1,8 @@
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
import { CronRunner } from "./cron-runner.js";
import type { TaskStore, AutomationStore, ScheduledTask, AutomationRunResult, Settings } from "@kb/core";
import type { TaskStore, AutomationStore, ScheduledTask, AutomationRunResult, AutomationStep, Settings } from "@kb/core";
import { DEFAULT_SETTINGS } from "@kb/core";
import { randomUUID } from "node:crypto";
function createMockSchedule(overrides: Partial<ScheduledTask> = {}): ScheduledTask {
return {
@@ -357,4 +358,270 @@ describe("CronRunner", () => {
expect(runner["pollIntervalMs"]).toBe(60000);
});
});
// ── Multi-step execution ──────────────────────────────────────────
describe("multi-step execution", () => {
function makeStep(overrides: Partial<AutomationStep> = {}): AutomationStep {
return {
id: randomUUID(),
type: "command",
name: "Test step",
command: "echo step-output",
...overrides,
};
}
it("legacy single-command mode still works when no steps are defined", async () => {
const store = createMockStore();
const schedule = createMockSchedule({ command: "echo legacy-mode" });
const automationStore = createMockAutomationStore([schedule]);
runner = new CronRunner(store, automationStore);
const result = await runner.executeSchedule(schedule);
expect(result.success).toBe(true);
expect(result.output).toContain("legacy-mode");
expect(result.stepResults).toBeUndefined();
});
it("executes multiple command steps sequentially", async () => {
const store = createMockStore();
const schedule = createMockSchedule({
command: "",
steps: [
makeStep({ name: "Step 1", command: "echo first" }),
makeStep({ name: "Step 2", command: "echo second" }),
],
});
const automationStore = createMockAutomationStore([schedule]);
runner = new CronRunner(store, automationStore);
const result = await runner.executeSchedule(schedule);
expect(result.success).toBe(true);
expect(result.stepResults).toHaveLength(2);
expect(result.stepResults![0].success).toBe(true);
expect(result.stepResults![0].output).toContain("first");
expect(result.stepResults![1].success).toBe(true);
expect(result.stepResults![1].output).toContain("second");
});
it("stops on step failure when continueOnFailure is false", async () => {
const store = createMockStore();
const schedule = createMockSchedule({
command: "",
steps: [
makeStep({ name: "Failing step", command: "exit 1" }),
makeStep({ name: "Should not run", command: "echo should-not-run" }),
],
});
const automationStore = createMockAutomationStore([schedule]);
runner = new CronRunner(store, automationStore);
const result = await runner.executeSchedule(schedule);
expect(result.success).toBe(false);
expect(result.stepResults).toHaveLength(1); // Only the first step ran
expect(result.stepResults![0].success).toBe(false);
expect(result.error).toContain("1 step(s) failed");
expect(result.error).toContain("execution stopped");
});
it("continues after failure when continueOnFailure is true", async () => {
const store = createMockStore();
const schedule = createMockSchedule({
command: "",
steps: [
makeStep({ name: "Failing step", command: "exit 1", continueOnFailure: true }),
makeStep({ name: "Should still run", command: "echo continued" }),
],
});
const automationStore = createMockAutomationStore([schedule]);
runner = new CronRunner(store, automationStore);
const result = await runner.executeSchedule(schedule);
expect(result.success).toBe(false); // Overall failure
expect(result.stepResults).toHaveLength(2); // Both steps ran
expect(result.stepResults![0].success).toBe(false);
expect(result.stepResults![1].success).toBe(true);
expect(result.stepResults![1].output).toContain("continued");
});
it("uses per-step timeout override", async () => {
const store = createMockStore();
const schedule = createMockSchedule({
command: "",
timeoutMs: 30000, // schedule-level timeout (large)
steps: [
makeStep({ name: "Slow step", command: "sleep 60", timeoutMs: 100 }), // step-level timeout (tiny)
],
});
const automationStore = createMockAutomationStore([schedule]);
runner = new CronRunner(store, automationStore);
const result = await runner.executeSchedule(schedule);
expect(result.success).toBe(false);
expect(result.stepResults).toHaveLength(1);
expect(result.stepResults![0].error).toContain("timed out");
}, 10000);
it("falls back to schedule-level timeout when step has no timeout", async () => {
const store = createMockStore();
const schedule = createMockSchedule({
command: "",
timeoutMs: 100, // tiny schedule-level timeout
steps: [
makeStep({ name: "Slow step", command: "sleep 60" }), // no step-level timeout
],
});
const automationStore = createMockAutomationStore([schedule]);
runner = new CronRunner(store, automationStore);
const result = await runner.executeSchedule(schedule);
expect(result.success).toBe(false);
expect(result.stepResults![0].error).toContain("timed out");
}, 10000);
it("handles AI prompt step execution (mocked)", async () => {
const store = createMockStore();
const schedule = createMockSchedule({
command: "",
steps: [
makeStep({
type: "ai-prompt",
name: "AI Analysis",
prompt: "Analyze the codebase",
modelProvider: "anthropic",
modelId: "claude-sonnet-4-5",
command: undefined,
}),
],
});
const automationStore = createMockAutomationStore([schedule]);
runner = new CronRunner(store, automationStore);
const result = await runner.executeSchedule(schedule);
expect(result.success).toBe(true);
expect(result.stepResults).toHaveLength(1);
expect(result.stepResults![0].success).toBe(true);
expect(result.stepResults![0].output).toContain("anthropic/claude-sonnet-4-5");
expect(result.stepResults![0].output).toContain("Analyze the codebase");
});
it("fails AI prompt step when no prompt is provided", async () => {
const store = createMockStore();
const schedule = createMockSchedule({
command: "",
steps: [
makeStep({
type: "ai-prompt",
name: "Empty prompt",
prompt: "",
command: undefined,
}),
],
});
const automationStore = createMockAutomationStore([schedule]);
runner = new CronRunner(store, automationStore);
const result = await runner.executeSchedule(schedule);
expect(result.success).toBe(false);
expect(result.stepResults![0].error).toContain("no prompt specified");
});
it("fails command step when no command is provided", async () => {
const store = createMockStore();
const schedule = createMockSchedule({
command: "",
steps: [
makeStep({ name: "Empty command", command: "" }),
],
});
const automationStore = createMockAutomationStore([schedule]);
runner = new CronRunner(store, automationStore);
const result = await runner.executeSchedule(schedule);
expect(result.success).toBe(false);
expect(result.stepResults![0].error).toContain("no command specified");
});
it("aggregates output from all steps with headers", async () => {
const store = createMockStore();
const schedule = createMockSchedule({
command: "",
steps: [
makeStep({ name: "Alpha", command: "echo alpha-output" }),
makeStep({ name: "Beta", command: "echo beta-output" }),
],
});
const automationStore = createMockAutomationStore([schedule]);
runner = new CronRunner(store, automationStore);
const result = await runner.executeSchedule(schedule);
expect(result.output).toContain("Step 1: Alpha");
expect(result.output).toContain("alpha-output");
expect(result.output).toContain("Step 2: Beta");
expect(result.output).toContain("beta-output");
});
it("records run result with stepResults to automation store", async () => {
const store = createMockStore();
const schedule = createMockSchedule({
command: "",
steps: [makeStep({ name: "S1", command: "echo ok" })],
});
const automationStore = createMockAutomationStore([schedule]);
runner = new CronRunner(store, automationStore);
await runner.executeSchedule(schedule);
expect(automationStore.recordRun).toHaveBeenCalledWith(
schedule.id,
expect.objectContaining({
success: true,
stepResults: expect.arrayContaining([
expect.objectContaining({
stepName: "S1",
success: true,
}),
]),
}),
);
});
it("mixed step types execute correctly", async () => {
const store = createMockStore();
const schedule = createMockSchedule({
command: "",
steps: [
makeStep({ name: "Build", command: "echo build-done" }),
makeStep({
type: "ai-prompt",
name: "Summarize",
prompt: "Summarize results",
command: undefined,
}),
makeStep({ name: "Deploy", command: "echo deployed" }),
],
});
const automationStore = createMockAutomationStore([schedule]);
runner = new CronRunner(store, automationStore);
const result = await runner.executeSchedule(schedule);
expect(result.success).toBe(true);
expect(result.stepResults).toHaveLength(3);
expect(result.stepResults![0].stepName).toBe("Build");
expect(result.stepResults![1].stepName).toBe("Summarize");
expect(result.stepResults![2].stepName).toBe("Deploy");
});
});
});

292
packages/engine/src/cron-runner.ts
View File

@@ -2,7 +2,7 @@ import { exec } from "node:child_process";
import { promisify } from "node:util";
import type { TaskStore } from "@kb/core";
import type { AutomationStore } from "@kb/core";
import type { ScheduledTask, AutomationRunResult } from "@kb/core";
import type { ScheduledTask, AutomationRunResult, AutomationStep, AutomationStepResult } from "@kb/core";
import { createLogger } from "./logger.js";
const execAsync = promisify(exec);
@@ -118,53 +118,26 @@ export class CronRunner {
}
/**
* Execute a single schedule's command.
* - Tracks in-flight state to prevent concurrent runs.
* - Enforces timeout and output buffer limits.
* - Records the run result in the automation store.
* Execute a single schedule.
*
* - **Legacy mode**: When `steps` is undefined/empty, execute `command` directly.
* - **Step mode**: When `steps` is present, execute steps sequentially.
*
* Tracks in-flight state to prevent concurrent runs.
* Records the run result in the automation store.
*/
async executeSchedule(schedule: ScheduledTask): Promise<AutomationRunResult> {
this.inFlight.add(schedule.id);
const startedAt = new Date().toISOString();
log.log(`Executing ${schedule.name} (${schedule.id}): ${schedule.command}`);
let result: AutomationRunResult;
try {
const timeoutMs = schedule.timeoutMs ?? DEFAULT_TIMEOUT_MS;
const { stdout, stderr } = await execAsync(schedule.command, {
timeout: timeoutMs,
maxBuffer: MAX_BUFFER,
shell: "/bin/sh",
});
const output = truncateOutput(stdout, stderr);
result = {
success: true,
output,
startedAt,
completedAt: new Date().toISOString(),
};
log.log(`✓ ${schedule.name} completed (${result.output.length} bytes output)`);
} catch (err: any) {
const stdout = err.stdout ?? "";
const stderr = err.stderr ?? "";
const output = truncateOutput(stdout, stderr);
const errorMessage = err.killed
? `Command timed out after ${(schedule.timeoutMs ?? DEFAULT_TIMEOUT_MS) / 1000}s`
: err.message ?? String(err);
result = {
success: false,
output,
error: errorMessage,
startedAt,
completedAt: new Date().toISOString(),
};
log.warn(`✗ ${schedule.name} failed: ${errorMessage}`);
if (schedule.steps && schedule.steps.length > 0) {
result = await this.executeSteps(schedule, startedAt);
} else {
result = await this.executeLegacyCommand(schedule, startedAt);
}
} finally {
this.inFlight.delete(schedule.id);
}
@@ -178,6 +151,245 @@ export class CronRunner {
return result;
}
/**
* Execute a legacy single-command schedule.
*/
private async executeLegacyCommand(
schedule: ScheduledTask,
startedAt: string,
): Promise<AutomationRunResult> {
log.log(`Executing ${schedule.name} (${schedule.id}): ${schedule.command}`);
try {
const timeoutMs = schedule.timeoutMs ?? DEFAULT_TIMEOUT_MS;
const { stdout, stderr } = await execAsync(schedule.command, {
timeout: timeoutMs,
maxBuffer: MAX_BUFFER,
shell: "/bin/sh",
});
const output = truncateOutput(stdout, stderr);
log.log(`✓ ${schedule.name} completed (${output.length} bytes output)`);
return {
success: true,
output,
startedAt,
completedAt: new Date().toISOString(),
};
} catch (err: any) {
const stdout = err.stdout ?? "";
const stderr = err.stderr ?? "";
const output = truncateOutput(stdout, stderr);
const errorMessage = err.killed
? `Command timed out after ${(schedule.timeoutMs ?? DEFAULT_TIMEOUT_MS) / 1000}s`
: err.message ?? String(err);
log.warn(`✗ ${schedule.name} failed: ${errorMessage}`);
return {
success: false,
output,
error: errorMessage,
startedAt,
completedAt: new Date().toISOString(),
};
}
}
/**
* Execute multiple steps sequentially.
* Aggregates per-step results into an overall AutomationRunResult.
*/
private async executeSteps(
schedule: ScheduledTask,
startedAt: string,
): Promise<AutomationRunResult> {
const steps = schedule.steps!;
log.log(`Executing ${schedule.name} (${schedule.id}): ${steps.length} steps`);
const stepResults: AutomationStepResult[] = [];
let overallSuccess = true;
let stoppedEarly = false;
for (let i = 0; i < steps.length; i++) {
const step = steps[i];
log.log(` Step ${i + 1}/${steps.length}: ${step.name} (${step.type})`);
const stepResult = await this.executeStep(schedule, step, i);
stepResults.push(stepResult);
if (!stepResult.success) {
overallSuccess = false;
if (!step.continueOnFailure) {
log.warn(` Step "${step.name}" failed — stopping execution`);
stoppedEarly = true;
break;
}
log.warn(` Step "${step.name}" failed — continuing (continueOnFailure=true)`);
} else {
log.log(` ✓ Step "${step.name}" completed`);
}
}
// Aggregate output from all steps
const outputParts: string[] = [];
for (const sr of stepResults) {
outputParts.push(`=== Step ${sr.stepIndex + 1}: ${sr.stepName} (${sr.success ? "success" : "FAILED"}) ===`);
if (sr.output) outputParts.push(sr.output);
if (sr.error) outputParts.push(`Error: ${sr.error}`);
}
const output = truncateOutput(outputParts.join("\n"), "");
// Build error summary
const failedSteps = stepResults.filter((sr) => !sr.success);
const error = failedSteps.length > 0
? `${failedSteps.length} step(s) failed: ${failedSteps.map((s) => s.stepName).join(", ")}${stoppedEarly ? " (execution stopped)" : ""}`
: undefined;
const status = overallSuccess ? "✓" : "✗";
log.log(`${status} ${schedule.name}: ${stepResults.length}/${steps.length} steps executed, ${failedSteps.length} failed`);
return {
success: overallSuccess,
output,
error,
startedAt,
completedAt: new Date().toISOString(),
stepResults,
};
}
/**
* Execute a single automation step.
*/
async executeStep(
schedule: ScheduledTask,
step: AutomationStep,
stepIndex: number,
): Promise<AutomationStepResult> {
const stepStartedAt = new Date().toISOString();
const timeoutMs = step.timeoutMs ?? schedule.timeoutMs ?? DEFAULT_TIMEOUT_MS;
if (step.type === "command") {
return this.executeCommandStep(step, stepIndex, timeoutMs, stepStartedAt);
} else if (step.type === "ai-prompt") {
return this.executeAiPromptStep(step, stepIndex, timeoutMs, stepStartedAt);
}
// Unknown step type
return {
stepId: step.id,
stepName: step.name,
stepIndex,
success: false,
output: "",
error: `Unknown step type: "${(step as any).type}"`,
startedAt: stepStartedAt,
completedAt: new Date().toISOString(),
};
}
/**
* Execute a command step using shell execution.
*/
private async executeCommandStep(
step: AutomationStep,
stepIndex: number,
timeoutMs: number,
startedAt: string,
): Promise<AutomationStepResult> {
if (!step.command?.trim()) {
return {
stepId: step.id,
stepName: step.name,
stepIndex,
success: false,
output: "",
error: "Command step has no command specified",
startedAt,
completedAt: new Date().toISOString(),
};
}
try {
const { stdout, stderr } = await execAsync(step.command, {
timeout: timeoutMs,
maxBuffer: MAX_BUFFER,
shell: "/bin/sh",
});
return {
stepId: step.id,
stepName: step.name,
stepIndex,
success: true,
output: truncateOutput(stdout, stderr),
startedAt,
completedAt: new Date().toISOString(),
};
} catch (err: any) {
const stdout = err.stdout ?? "";
const stderr = err.stderr ?? "";
const errorMessage = err.killed
? `Command timed out after ${timeoutMs / 1000}s`
: err.message ?? String(err);
return {
stepId: step.id,
stepName: step.name,
stepIndex,
success: false,
output: truncateOutput(stdout, stderr),
error: errorMessage,
startedAt,
completedAt: new Date().toISOString(),
};
}
}
/**
* Execute an AI prompt step.
* In a full implementation, this would create an agent session and run the prompt.
* For now, we log the prompt and model selection and return a placeholder result.
*/
private async executeAiPromptStep(
step: AutomationStep,
stepIndex: number,
_timeoutMs: number,
startedAt: string,
): Promise<AutomationStepResult> {
if (!step.prompt?.trim()) {
return {
stepId: step.id,
stepName: step.name,
stepIndex,
success: false,
output: "",
error: "AI prompt step has no prompt specified",
startedAt,
completedAt: new Date().toISOString(),
};
}
const model = step.modelProvider && step.modelId
? `${step.modelProvider}/${step.modelId}`
: "default";
log.log(` AI prompt step "${step.name}" using model: ${model}`);
log.log(` Prompt: ${step.prompt.slice(0, 100)}${step.prompt.length > 100 ? "…" : ""}`);
// TODO: Integrate with actual agent session for AI prompt execution
return {
stepId: step.id,
stepName: step.name,
stepIndex,
success: true,
output: `[AI prompt step — model: ${model}]\nPrompt: ${step.prompt}\n\n(AI execution not yet implemented — prompt recorded for future integration)`,
startedAt,
completedAt: new Date().toISOString(),
};
}
}
/** Combine and truncate stdout/stderr to stay within storage limits. */