Declarative pipeline definition + durable runner (#97)

workers/api/src/agent-do-tools.ts

@@ -24,6 +24,8 @@ const BASE = [
 	"set_user_preference",
 	// Every agent owns a work board; let it reshape its own columns/view on request.
 	"configure_board",
+	// Start a declarative data pipeline the owner configured on this instance (#97).
+	"run_pipeline",
 ] as const;
 
 /** Read the vector knowledge base (RAG). Only agents that HAVE an index get these. */

workers/api/src/index.ts

@@ -50,6 +50,8 @@ export { AgentDO } from "./agent-do.js";
 export { JobApplyWorkflow } from "./workflows/job-apply.js";
 // Re-export the coding-orchestrator Workflow class for wrangler (AgentCoder port)
 export { CodingSessionWorkflow } from "./workflows/coding-session.js";
+// Re-export the declarative-pipeline Workflow class for wrangler (issue #97)
+export { PipelineRunWorkflow } from "./workflows/pipeline-run.js";
 // Re-export the WebSocket relay DO for wrangler
 export { RelayDO } from "./relay-do.js";
 

workers/api/src/lib/pipeline-run-start.test.ts

@@ -0,0 +1,73 @@
+import { describe, expect, it, vi } from "vitest";
+
+// A registry whose tools are all "known" so a stored def validates.
+vi.mock("./tool-registry.js", () => ({
+	getRegistryTool: (name: string) => ({ name }),
+	runRegistryTool: vi.fn(),
+}));
+
+import { startPipelineRun } from "./pipeline-run-start.js";
+import { loadPipeline } from "./pipeline.js";
+import type { Env } from "../types.js";
+
+const PIPE = { name: "leads", steps: [{ tool: "geocode", inputs: { city: { $param: "city" } } }], sink: { collection: "leads" } };
+
+/** Env whose agent_instances row carries a config with a pipelines map. */
+function envWithConfig(config: unknown, create?: (arg: unknown) => Promise<{ id: string }>): Env {
+	return {
+ DB: { prepare: () => ({ bind: () => ({ first: async () => (config === null ? null : { config: typeof config === "string" ? config : JSON.stringify(config) }) }) }) },
+ PIPELINE_RUN: { create: create ?? (async () => ({ id: "wf-1" })) },
+	} as unknown as Env;
+}
+
+describe("loadPipeline", () => {
+	it("reads a named pipeline from config.pipelines", async () => {
+ const def = await loadPipeline(envWithConfig({ pipelines: { leads: PIPE } }), "i1", "u1", "leads");
+ expect(def?.name).toBe("leads");
+	});
+
+	it("returns null when the instance row is missing", async () => {
+ expect(await loadPipeline(envWithConfig(null), "i1", "u1", "leads")).toBeNull();
+	});
+
+	it("returns null for an unknown pipeline name", async () => {
+ expect(await loadPipeline(envWithConfig({ pipelines: { other: PIPE } }), "i1", "u1", "leads")).toBeNull();
+	});
+
+	it("returns null for malformed config JSON", async () => {
+ expect(await loadPipeline(envWithConfig("{not json"), "i1", "u1", "leads")).toBeNull();
+	});
+
+	it("returns null for an invalid stored definition", async () => {
+ expect(await loadPipeline(envWithConfig({ pipelines: { leads: { name: "leads", steps: [] } } }), "i1", "u1", "leads")).toBeNull();
+	});
+});
+
+describe("startPipelineRun", () => {
+	it("kicks the PIPELINE_RUN workflow with the loaded def + params", async () => {
+ const create = vi.fn(async () => ({ id: "wf-42" }));
+ const env = envWithConfig({ pipelines: { leads: PIPE } }, create);
+ const res = await startPipelineRun(env, "i1", "u1", "leads", { city: "Sydney" }, "chat");
+ expect(res.ok).toBe(true);
+ if (res.ok) {
+ expect(res.workflowId).toBe("wf-42");
+ expect(res.runId).toBeTruthy();
+ }
+ expect(create).toHaveBeenCalledTimes(1);
+ const arg = create.mock.calls[0][0] as { params: { pipeline: { name: string }; params: unknown; trigger: string; instanceId: string; userId: string } };
+ expect(arg.params.pipeline.name).toBe("leads");
+ expect(arg.params.params).toEqual({ city: "Sydney" });
+ expect(arg.params.trigger).toBe("chat");
+ expect(arg.params.instanceId).toBe("i1");
+ expect(arg.params.userId).toBe("u1");
+	});
+
+	it("returns an error (does not kick) for an unknown pipeline", async () => {
+ const create = vi.fn(async () => ({ id: "wf" }));
+ const env = envWithConfig({ pipelines: {} }, create);
+ const res = await startPipelineRun(env, "i1", "u1", "nope", {}, "api");
+ expect(res.ok).toBe(false);
+ if (!res.ok) expect(res.error).toMatch(/No pipeline named "nope"/);
+ expect(create).not.toHaveBeenCalled();
+	});
+});

workers/api/src/lib/pipeline-run-start.ts

@@ -0,0 +1,35 @@
+// The SINGLE kick path for a pipeline run (issue #97), shared by the LLM-callable
+// `run_pipeline` tool and the POST /v1/instances/:id/pipelines/:name/run endpoint so both
+// audit + start a run identically. Cron/webhook triggers (#92) call this same helper —
+// that's the clean hook for trigger wiring, which is NOT built here.
+import { loadPipeline } from "./pipeline.js";
+import { logEvent } from "./events.js";
+import type { PipelineRunParams } from "../workflows/pipeline-run.js";
+import type { Env } from "../types.js";
+
+export type StartTrigger = NonNullable<PipelineRunParams["trigger"]>;
+
+export type StartResult = { ok: true; runId: string; workflowId: string } | { ok: false; error: string };
+
+/**
+ * Load the named pipeline from the instance's config, audit the start, and kick the durable
+ * PipelineRunWorkflow. Owner-scoping is the caller's responsibility (the API route calls
+ * requireOwnedInstance; the tool runs with the owner's own instanceId/userId). Returns an
+ * error string when the pipeline isn't found so callers surface a clean message.
+ */
+export async function startPipelineRun(
+	env: Env,
+	instanceId: string,
+	userId: string,
+	name: string,
+	params: Record<string, unknown>,
+	trigger: StartTrigger,
+): Promise<StartResult> {
+	const pipeline = await loadPipeline(env, instanceId, userId, name);
+	if (!pipeline) return { ok: false, error: `No pipeline named "${name}" is configured on this agent (or its definition is invalid).` };
+	const runId = crypto.randomUUID();
+	// Audit the start BEFORE the kick so a failed create still leaves a record of who asked.
+	await logEvent(env, { source: "pipeline", event: "pipeline.requested", message: `Start "${name}" via ${trigger}`, userId, instanceId, traceId: runId, context: { pipeline: name, trigger, params } }).catch(() => undefined);
+	const wf = await env.PIPELINE_RUN.create({ params: { instanceId, userId, pipeline, params, runId, trigger } satisfies PipelineRunParams });
+	return { ok: true, runId, workflowId: wf.id };
+}

workers/api/src/lib/pipeline.test.ts

@@ -0,0 +1,138 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+// Mock the tool registry so schema validation (getRegistryTool) + step dispatch
+// (runRegistryTool) are controllable without real connectors. A couple of fake tools:
+// - "geocode": returns a JSON object {lat,lng}
+// - "places": returns a JSON array of results
+// - "reachable": echoes its input so we can assert threaded/fan-out inputs
+const KNOWN = new Set(["geocode", "places", "reachable", "noop"]);
+const runRegistryTool = vi.fn();
+vi.mock("./tool-registry.js", () => ({
+	getRegistryTool: (name: string) => (KNOWN.has(name) ? { name } : undefined),
+	runRegistryTool: (...args: unknown[]) => runRegistryTool(...args),
+}));
+
+import { executePipelineStep, resolveInputs, resolveInputValue, validatePipeline, stepBind, type PipelineDef } from "./pipeline.js";
+import type { Env } from "../types.js";
+
+const env = {} as Env;
+const ctx = { env, userId: "u1", instanceId: "i1" };
+
+beforeEach(() => {
+	runRegistryTool.mockReset();
+});
+
+describe("validatePipeline", () => {
+	it("accepts a minimal valid pipeline", () => {
+ const def: PipelineDef = { name: "p", steps: [{ tool: "geocode", inputs: { city: "Sydney" } }] };
+ expect(validatePipeline(def)).toBeNull();
+	});
+
+	it("rejects a missing name", () => {
+ expect(validatePipeline({ steps: [{ tool: "geocode" }] })).toMatch(/name is required/);
+	});
+
+	it("rejects empty steps", () => {
+ expect(validatePipeline({ name: "p", steps: [] })).toMatch(/non-empty array/);
+	});
+
+	it("rejects an unknown tool", () => {
+ expect(validatePipeline({ name: "p", steps: [{ tool: "does_not_exist" }] })).toMatch(/unknown tool/);
+	});
+
+	it("rejects duplicate bind names", () => {
+ const def = { name: "p", steps: [{ tool: "geocode", bind: "g" }, { tool: "places", bind: "g" }] };
+ expect(validatePipeline(def)).toMatch(/duplicate bind/);
+	});
+
+	it("rejects a sink without a collection", () => {
+ const def = { name: "p", steps: [{ tool: "geocode" }], sink: {} };
+ expect(validatePipeline(def)).toMatch(/sink.collection is required/);
+	});
+});
+
+describe("resolveInputValue / resolveInputs", () => {
+	const scope = { outputs: { geo: { lat: -33.8, lng: 151.2 } }, params: { city: "Sydney", radius: 5000 } };
+
+	it("passes literals through", () => {
+ expect(resolveInputValue("hi", scope)).toBe("hi");
+ expect(resolveInputValue(42, scope)).toBe(42);
+	});
+
+	it("resolves $param", () => {
+ expect(resolveInputValue({ $param: "city" }, scope)).toBe("Sydney");
+	});
+
+	it("resolves $ref with a dotted path", () => {
+ expect(resolveInputValue({ $ref: "geo.lat" }, scope)).toBe(-33.8);
+	});
+
+	it("resolves nested objects + arrays recursively", () => {
+ const out = resolveInputs({ center: { lat: { $ref: "geo.lat" }, lng: { $ref: "geo.lng" } }, tags: [{ $param: "city" }, "fixed"] }, scope);
+ expect(out).toEqual({ center: { lat: -33.8, lng: 151.2 }, tags: ["Sydney", "fixed"] });
+	});
+
+	it("resolves $param:item to the fan-out item", () => {
+ expect(resolveInputValue({ $param: "item" }, { ...scope, item: { n: 1 } })).toEqual({ n: 1 });
+	});
+});
+
+describe("executePipelineStep — dispatches via runRegistryTool + threads outputs", () => {
+	it("dispatches the tool through runRegistryTool with resolved inputs", async () => {
+ runRegistryTool.mockResolvedValue({ name: "geocode", content: '{"lat":-33.8,"lng":151.2}', success: true });
+ const step = { tool: "geocode", inputs: { city: { $param: "city" } }, bind: "geo" };
+ const res = await executePipelineStep(ctx, step, 0, {}, { city: "Sydney" });
+ // dispatched through the single registry path with (name, ctx, resolvedInput)
+ expect(runRegistryTool).toHaveBeenCalledWith("geocode", { env, userId: "u1", instanceId: "i1" }, { city: "Sydney" });
+ // parsed JSON output threaded under the bind
+ expect(res.bind).toBe("geo");
+ expect(res.success).toBe(true);
+ expect(res.output).toEqual({ lat: -33.8, lng: 151.2 });
+	});
+
+	it("threads a prior step's output into the next step's inputs", async () => {
+ runRegistryTool.mockResolvedValue({ name: "places", content: "[]", success: true });
+ const outputs = { geo: { lat: -33.8, lng: 151.2 } };
+ const step = { tool: "places", inputs: { lat: { $ref: "geo.lat" }, lng: { $ref: "geo.lng" } } };
+ await executePipelineStep(ctx, step, 1, outputs, {});
+ expect(runRegistryTool).toHaveBeenCalledWith("places", expect.anything(), { lat: -33.8, lng: 151.2 });
+	});
+
+	it("keeps a non-JSON tool result as a raw string", async () => {
+ runRegistryTool.mockResolvedValue({ name: "noop", content: "done", success: true });
+ const res = await executePipelineStep(ctx, { tool: "noop" }, 0, {}, {});
+ expect(res.output).toBe("done");
+	});
+
+	it("propagates a failed tool call as an unsuccessful step", async () => {
+ runRegistryTool.mockResolvedValue({ name: "geocode", content: "boom", success: false });
+ const res = await executePipelineStep(ctx, { tool: "geocode" }, 0, {}, {});
+ expect(res.success).toBe(false);
+	});
+
+	it("forEach runs the step once per array item and collects results", async () => {
+ runRegistryTool
+ .mockResolvedValueOnce({ name: "reachable", content: '{"ok":true}', success: true })
+ .mockResolvedValueOnce({ name: "reachable", content: '{"ok":false}', success: true });
+ const outputs = { sites: ["a.com", "b.com"] };
+ const step = { tool: "reachable", forEach: { $ref: "sites" }, inputs: { url: { $param: "item" } }, bind: "checked" };
+ const res = await executePipelineStep(ctx, step, 0, outputs, {});
+ expect(runRegistryTool).toHaveBeenCalledTimes(2);
+ expect(runRegistryTool).toHaveBeenNthCalledWith(1, "reachable", expect.anything(), { url: "a.com" });
+ expect(runRegistryTool).toHaveBeenNthCalledWith(2, "reachable", expect.anything(), { url: "b.com" });
+ expect(res.output).toEqual([{ ok: true }, { ok: false }]);
+	});
+
+	it("forEach over a non-array is a step failure", async () => {
+ const res = await executePipelineStep(ctx, { tool: "reachable", forEach: { $param: "nope" } }, 0, {}, {});
+ expect(res.success).toBe(false);
+ expect(runRegistryTool).not.toHaveBeenCalled();
+	});
+});
+
+describe("stepBind", () => {
+	it("uses the explicit bind, else stepN", () => {
+ expect(stepBind({ tool: "x", bind: "y" }, 3)).toBe("y");
+ expect(stepBind({ tool: "x" }, 3)).toBe("step3");
+	});
+});