From 03953343fbe93e70b9e07a00f97f82306d8e288a Mon Sep 17 00:00:00 2001 From: Serge Ivo Date: Sun, 2 Aug 2026 08:10:34 +1000 Subject: [PATCH] feat(pipelines): declarative pipeline definition + durable runner (#97) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A pipeline is JSON DATA — an ordered list of steps, each dispatching a registry tool (#85/#86) through the SINGLE runRegistryTool path so connector auth/grant/consent (#86/#90) are enforced identically to a direct tool call. Outputs thread between steps by `bind`; inputs reference prior outputs + run params via a $ref/$param convention. A source→transform→sink agent is now configuration, not code. - lib/pipeline.ts: schema { name, params?, steps:[{tool,inputs,bind?,forEach?}], sink? }, validatePipeline, resolveInputs ($ref/$param + recursion + forEach item), executePipelineStep (pure, unit-testable), loadPipeline (from instance config). - workflows/pipeline-run.ts: PipelineRunWorkflow (WorkflowEntrypoint) — walks steps, each in its own step.do for durability/resumability past the 30s DO limit (mirrors JobApplyWorkflow). Connector tokens re-minted inside each step, never captured across steps (resume-determinism caveat). Optional sink upserts the final output into an instance collection via the AgentDO records route. - lib/pipeline-run-start.ts: the single kick path (audit + create), shared by the tool + the API route; the clean hook for cron/webhook triggers (#92, not built here). - run_pipeline first-party ToolDef (registry) so an agent can start a declared pipeline on request ("sweep Sydney"); added to BASE so every agent may call it. - POST /v1/instances/:id/pipelines/:name/run (owner-scoped + audited) + GET /v1/instances/:id/pipelines. PIPELINE_RUN workflow binding (Env, wrangler, index export). Definitions stored per instance in agent_instances.config.pipelines[name] — the least-invasive store (reuses the row requireOwnedInstance already reads; no new migration; editable via the existing settings/MCP config round-trip). Tests: schema validation, input resolution + fan-out, the runner threads outputs between steps + dispatches via runRegistryTool (registry mocked), loadPipeline, startPipelineRun kicks the workflow, and the POST endpoint owner-gated + audited. The Workflow class itself isn't unit-tested (no Workflow harness); executePipelineStep + startPipelineRun cover its logic directly. api tsc clean; api suite 791 green (761 baseline + new, 0 regressions). Co-Authored-By: Claude Opus 4.8 (1M context) --- workers/api/src/agent-do-tools.ts | 2 + workers/api/src/index.ts | 2 + .../api/src/lib/pipeline-run-start.test.ts | 73 ++++++ workers/api/src/lib/pipeline-run-start.ts | 35 +++ workers/api/src/lib/pipeline.test.ts | 138 +++++++++++ workers/api/src/lib/pipeline.ts | 227 ++++++++++++++++++ workers/api/src/lib/tool-registry.test.ts | 7 +- workers/api/src/lib/tool-registry.ts | 32 ++- workers/api/src/routes/tools.test.ts | 64 ++++- workers/api/src/routes/tools.ts | 43 ++++ workers/api/src/types.ts | 2 + workers/api/src/workflows/pipeline-run.ts | 122 ++++++++++ workers/api/wrangler.toml | 6 + 13 files changed, 744 insertions(+), 9 deletions(-) create mode 100644 workers/api/src/lib/pipeline-run-start.test.ts create mode 100644 workers/api/src/lib/pipeline-run-start.ts create mode 100644 workers/api/src/lib/pipeline.test.ts create mode 100644 workers/api/src/lib/pipeline.ts create mode 100644 workers/api/src/workflows/pipeline-run.ts diff --git a/workers/api/src/agent-do-tools.ts b/workers/api/src/agent-do-tools.ts index ffd020fc..3182470e 100644 --- a/workers/api/src/agent-do-tools.ts +++ b/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. */ diff --git a/workers/api/src/index.ts b/workers/api/src/index.ts index 9c0b4f7e..4eb99718 100644 --- a/workers/api/src/index.ts +++ b/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"; diff --git a/workers/api/src/lib/pipeline-run-start.test.ts b/workers/api/src/lib/pipeline-run-start.test.ts new file mode 100644 index 00000000..857013e3 --- /dev/null +++ b/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(); + }); +}); diff --git a/workers/api/src/lib/pipeline-run-start.ts b/workers/api/src/lib/pipeline-run-start.ts new file mode 100644 index 00000000..f481e5ec --- /dev/null +++ b/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; + +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, + trigger: StartTrigger, +): Promise { + 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 }; +} diff --git a/workers/api/src/lib/pipeline.test.ts b/workers/api/src/lib/pipeline.test.ts new file mode 100644 index 00000000..09fcab07 --- /dev/null +++ b/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"); + }); +}); diff --git a/workers/api/src/lib/pipeline.ts b/workers/api/src/lib/pipeline.ts new file mode 100644 index 00000000..2d8a77e5 --- /dev/null +++ b/workers/api/src/lib/pipeline.ts @@ -0,0 +1,227 @@ +// Declarative data pipelines (issue #97, epic #94). A pipeline is DATA, not code: an +// ordered list of steps, each dispatching a registry tool (#85/#86) through the SINGLE +// runRegistryTool path so connector auth/grant/consent are enforced identically to a +// direct tool call. Outputs thread between steps by name (`bind`); inputs reference prior +// outputs + run params via a `$`-prefixed reference convention. This module holds the +// PURE schema + step-execution logic (validation, input resolution, one step) so it is +// unit-testable without the Workflow harness; the durable runner (workflows/pipeline-run.ts) +// wraps executePipelineStep in step.do for durability/resumability past the 30s DO limit. +import type { Env } from "../types.js"; +import { getRegistryTool, runRegistryTool } from "./tool-registry.js"; + +/** + * A reference into the run's data: `{ "$ref": "stepBind.field" }` reads a bound step's + * output (dotted path), `{ "$param": "name" }` reads a run parameter. Anything else is a + * literal (passed through verbatim, recursively for objects/arrays). + */ +export type PipelineInputValue = + | { $ref: string } + | { $param: string } + | { [key: string]: PipelineInputValue } + | PipelineInputValue[] + | string + | number + | boolean + | null; + +export interface PipelineStep { + /** Registry tool to dispatch (must exist in the tool registry). */ + tool: string; + /** The tool's input args — each value may be a literal or a $ref/$param reference. */ + inputs?: Record; + /** Name this step's output so later steps can `$ref` it. Defaults to the step index. */ + bind?: string; + /** + * Trivial fan-out seam (#96 owns the rich version): when set to a reference that + * resolves to an array, the step runs ONCE PER item — the item is exposed to `inputs` + * as `$param: "item"` — and this step's bound output is the array of per-item results. + * Sequential (no concurrency here); a concurrency cap lands with the #96 step library. + */ + forEach?: PipelineInputValue; +} + +export interface PipelineSink { + /** Target instance collection (#91) the final records are upserted into. */ + collection: string; + /** Unique key field for upsert/dedupe (optional; pass-through to the sink tool). */ + keyField?: string; +} + +/** A pipeline definition — pure data, storable per instance (see loadPipeline). */ +export interface PipelineDef { + name: string; + /** Declared run params (names → JSON type hint). Advisory; validation is name-based. */ + params?: Record; + steps: PipelineStep[]; + sink?: PipelineSink; +} + +/** A parsed tool result carried between steps. `output` is the tool's content parsed as + * JSON when possible, else the raw string. */ +export interface StepResult { + tool: string; + bind: string; + success: boolean; + content: string; + output: unknown; +} + +/** Validate a candidate pipeline definition. Returns an error string, or null if valid. + * Boundary validation only (definitions come from stored config / API callers). */ +export function validatePipeline(def: unknown): string | null { + if (!def || typeof def !== "object" || Array.isArray(def)) return "Pipeline must be an object"; + const p = def as Record; + if (typeof p.name !== "string" || !p.name.trim()) return "Pipeline.name is required"; + if (!Array.isArray(p.steps) || p.steps.length === 0) return "Pipeline.steps must be a non-empty array"; + const binds = new Set(); + for (let i = 0; i < p.steps.length; i++) { + const s = p.steps[i] as Record; + if (!s || typeof s !== "object") return `Step ${i} must be an object`; + if (typeof s.tool !== "string" || !s.tool.trim()) return `Step ${i}: "tool" is required`; + if (!getRegistryTool(s.tool as string)) return `Step ${i}: unknown tool "${s.tool}"`; + if (s.inputs !== undefined && (typeof s.inputs !== "object" || s.inputs === null || Array.isArray(s.inputs))) { + return `Step ${i}: "inputs" must be an object`; + } + if (s.bind !== undefined) { + if (typeof s.bind !== "string" || !s.bind.trim()) return `Step ${i}: "bind" must be a non-empty string`; + if (binds.has(s.bind as string)) return `Step ${i}: duplicate bind "${s.bind}"`; + binds.add(s.bind as string); + } + } + if (p.sink !== undefined) { + const sink = p.sink as Record; + if (!sink || typeof sink !== "object" || Array.isArray(sink)) return "sink must be an object"; + if (typeof sink.collection !== "string" || !sink.collection.trim()) return "sink.collection is required"; + } + return null; +} + +/** The default bind name for a step without an explicit one. */ +export function stepBind(step: PipelineStep, index: number): string { + return step.bind ?? `step${index}`; +} + +/** Read a dotted path (e.g. "geocode.lat") from the run's bound-output map + params. */ +function readPath(root: unknown, path: string): unknown { + let cur = root; + for (const key of path.split(".")) { + if (cur == null || typeof cur !== "object") return undefined; + cur = (cur as Record)[key]; + } + return cur; +} + +/** + * Resolve one input value against the run scope. `$ref` reads from bound step outputs + * (their parsed `output`), `$param` reads a run param, and a special `item` scope is used + * for forEach. Objects/arrays are resolved recursively so nested references work. + */ +export function resolveInputValue( + value: PipelineInputValue, + scope: { outputs: Record; params: Record; item?: unknown }, +): unknown { + if (value === null || typeof value !== "object") return value; + if (Array.isArray(value)) return value.map((v) => resolveInputValue(v, scope)); + const obj = value as Record; + if (typeof obj.$param === "string") { + // `$param: "item"` inside a forEach body reads the current fan-out item. + if (obj.$param === "item" && scope.item !== undefined) return scope.item; + return scope.params[obj.$param]; + } + if (typeof obj.$ref === "string") return readPath(scope.outputs, obj.$ref); + // A plain object literal — resolve each value recursively. + const out: Record = {}; + for (const [k, v] of Object.entries(obj)) out[k] = resolveInputValue(v as PipelineInputValue, scope); + return out; +} + +/** Resolve a step's whole `inputs` map against the run scope. */ +export function resolveInputs( + inputs: Record | undefined, + scope: { outputs: Record; params: Record; item?: unknown }, +): Record { + const resolved: Record = {}; + for (const [k, v] of Object.entries(inputs ?? {})) resolved[k] = resolveInputValue(v, scope); + return resolved; +} + +/** Parse a tool's string content as JSON when it looks like JSON, else return the string. */ +function parseOutput(content: string): unknown { + const t = content.trim(); + if (t && (t[0] === "{" || t[0] === "[")) { + try { + return JSON.parse(t); + } catch { + /* not JSON — fall through */ + } + } + return content; +} + +export interface PipelineRunCtx { + env: Env; + userId: string; + instanceId: string; +} + +/** + * Execute ONE pipeline step: resolve its inputs against prior outputs + params, dispatch + * via runRegistryTool (the single path enforcing connector auth/grant/consent), and return + * the result with its output parsed. On `forEach`, the step runs once per resolved array + * item (sequentially) and the output is the array of per-item parsed outputs. + * + * PURE w.r.t. the Workflow: it takes a plain ctx + the accumulated outputs/params, so the + * durable runner can call it inside step.do and unit tests can call it directly. + */ +export async function executePipelineStep( + ctx: PipelineRunCtx, + step: PipelineStep, + index: number, + outputs: Record, + params: Record, +): Promise { + const bind = stepBind(step, index); + const registryCtx = { env: ctx.env, userId: ctx.userId, instanceId: ctx.instanceId }; + + if (step.forEach !== undefined) { + const list = resolveInputValue(step.forEach, { outputs, params }); + if (!Array.isArray(list)) { + return { tool: step.tool, bind, success: false, content: `forEach did not resolve to an array for step ${index}`, output: null }; + } + const results: unknown[] = []; + let allOk = true; + for (const item of list) { + const input = resolveInputs(step.inputs, { outputs, params, item }); + const r = await runRegistryTool(step.tool, registryCtx, input); + if (!r.success) allOk = false; + results.push(parseOutput(r.content)); + } + return { tool: step.tool, bind, success: allOk, content: `${results.length} item(s)`, output: results }; + } + + const input = resolveInputs(step.inputs, { outputs, params }); + const r = await runRegistryTool(step.tool, registryCtx, input); + return { tool: step.tool, bind, success: r.success, content: r.content, output: parseOutput(r.content) }; +} + +/** + * Load a pipeline definition stored on the instance. Definitions live in the instance's + * existing `agent_instances.config` JSON under `config.pipelines[name]` — the least-invasive + * store (reuses the row already read by requireOwnedInstance; no new migration, and it's + * editable via the instance-settings UI/MCP that already round-trips `config`). Returns the + * def, or null if the instance/pipeline isn't found or the stored def is invalid. + */ +export async function loadPipeline(env: Env, instanceId: string, userId: string, name: string): Promise { + const row = await env.DB.prepare("SELECT config FROM agent_instances WHERE id = ?1 AND user_id = ?2").bind(instanceId, userId).first<{ config: string }>(); + if (!row) return null; + let cfg: Record; + try { + cfg = JSON.parse(row.config || "{}") as Record; + } catch { + return null; + } + const pipelines = cfg.pipelines as Record | undefined; + const def = pipelines?.[name]; + if (!def || validatePipeline(def) !== null) return null; + return def as PipelineDef; +} diff --git a/workers/api/src/lib/tool-registry.test.ts b/workers/api/src/lib/tool-registry.test.ts index e90a50e3..efb579db 100644 --- a/workers/api/src/lib/tool-registry.test.ts +++ b/workers/api/src/lib/tool-registry.test.ts @@ -35,13 +35,14 @@ describe("tool registry", () => { expect(def?.jsonSchema).toBe(getRegistryTool("github_workflow_runs")?.jsonSchema); }); - it("every registry tool declares a jsonSchema, a tier, and a connector", () => { + it("every registry tool declares a jsonSchema and a tier", () => { for (const t of registryTools()) { expect(t.jsonSchema.type).toBe("object"); expect(t.jsonSchema.properties).toEqual(expect.any(Object)); expect(["base", "standard", "runtime", "connector"]).toContain(t.tier); - // The current registry is connector-only; all entries name their connector. - expect(typeof t.connector).toBe("string"); + // Connector-tier tools name their connector; first-party tools (e.g. run_pipeline, + // issue #97) don't — the registry now carries both. + if (t.tier === "connector") expect(typeof t.connector).toBe("string"); } }); diff --git a/workers/api/src/lib/tool-registry.ts b/workers/api/src/lib/tool-registry.ts index 438ccbcd..0e1598b6 100644 --- a/workers/api/src/lib/tool-registry.ts +++ b/workers/api/src/lib/tool-registry.ts @@ -64,10 +64,36 @@ export type RegistryTool = ToolDef; /** * First-party registry tools that are NOT provided by a connector (base/standard/runtime - * tiers). Empty for now — kept so the REGISTRY can carry non-connector tools without - * changing its shape. + * tiers). `run_pipeline` (issue #97) lets an agent start a declarative pipeline the owner + * has declared on the instance ("sweep Sydney" → run the `leads` pipeline with city=Sydney). */ -const FIRST_PARTY_TOOLS: ToolDef[] = []; +const FIRST_PARTY_TOOLS: ToolDef[] = [ + { + name: "run_pipeline", + description: + "Run a declarative data pipeline that the owner has configured on this agent. Pass the pipeline `name` and any `params` (e.g. {city:\"Sydney\"}). The pipeline runs durably in the background (source → transform → sink); it does not return results inline — tell the user it's started.", + tier: "base", + jsonSchema: { + type: "object", + properties: { + name: { type: "string", description: "Name of a pipeline configured on this instance." }, + params: { type: "object", description: "Run parameters passed to the pipeline (JSON object)." }, + }, + required: ["name"], + }, + handler: async (ctx, input) => { + if (!ctx.instanceId || !ctx.userId) return { content: "run_pipeline needs an owned instance context.", success: false }; + const name = String(input.name ?? ""); + if (!name) return { content: "Pipeline name is required.", success: false }; + const params = (input.params && typeof input.params === "object" && !Array.isArray(input.params) ? input.params : {}) as Record; + // Deferred import avoids a cycle (pipeline.ts imports this module for the registry). + const { startPipelineRun } = await import("./pipeline-run-start.js"); + const started = await startPipelineRun(ctx.env, ctx.instanceId, ctx.userId, name, params, "chat"); + if (!started.ok) return { content: started.error, success: false }; + return { content: `Started pipeline "${name}" (run ${started.runId}). It runs in the background; check the trace/board for progress.`, success: true }; + }, + }, +]; // The tool REGISTRY, keyed by name: every connector's tools (flattened from the connector // registry, with connector/tier/scope stamped) plus first-party tools. Add a connector = diff --git a/workers/api/src/routes/tools.test.ts b/workers/api/src/routes/tools.test.ts index 7b4f993e..8ae294eb 100644 --- a/workers/api/src/routes/tools.test.ts +++ b/workers/api/src/routes/tools.test.ts @@ -6,13 +6,14 @@ import { toolRoutes } from "./tools.js"; const SECRET = "test-secret"; -function testApp(opts: { owned?: boolean } = { owned: true }) { +function testApp(opts: { owned?: boolean; config?: string; create?: (arg: unknown) => Promise<{ id: string }> } = { owned: true }) { const app = new Hono(); app.route("/v1/instances", toolRoutes); app.onError((err, c) => { if (err instanceof HttpError) return c.json({ error: err.message }, err.status as 400); throw err; }); + const config = opts.config ?? "{}"; const env = { SESSION_SIGNING_KEY: SECRET, // githubAppConfigured() → false (no GITHUB_APP_ID), so github tools return a @@ -23,18 +24,25 @@ function testApp(opts: { owned?: boolean } = { owned: true }) { bind() { return { first: async () => - sql.includes("FROM agent_instances") && opts.owned - ? { id: "i1", agent_id: "a1", user_id: "u1", status: "active", config: "{}", created_at: "", updated_at: "" } + sql.includes("FROM agent_instances") && (opts.owned ?? true) + ? { id: "i1", agent_id: "a1", user_id: "u1", status: "active", config, created_at: "", updated_at: "" } : null, + // logEvent (pipeline audit) does a .run() insert — must not throw. + run: async () => ({}), }; }, }; }, }, + // Durable pipeline runner (issue #97) — stubbed to capture .create() calls. + PIPELINE_RUN: { create: opts.create ?? (async () => ({ id: "wf-test" })) }, }; return { app, env }; } +// A valid stored pipeline whose step uses a real registry tool (so validatePipeline passes). +const STORED_PIPELINE = { pipelines: { sweep: { name: "sweep", steps: [{ tool: "github_workflow_runs", inputs: { repo: { $param: "repo" } }, bind: "runs" }], sink: { collection: "results" } } } }; + const tok = (uid: string) => signSession(uid, SECRET, { roles: ["user"] }); const req = (app: Hono, env: unknown, path: string, init: RequestInit, t: string) => app.request(path, { ...init, headers: { Authorization: `Bearer ${t}`, "Content-Type": "application/json", ...(init.headers || {}) } }, env); @@ -136,3 +144,53 @@ describe("POST /v1/instances/:id/tools/:name", () => { fetchSpy.mockRestore(); }); }); + +describe("GET /v1/instances/:id/pipelines (issue #97)", () => { + it("lists pipelines declared in the instance config", async () => { + const { app, env } = testApp({ config: JSON.stringify(STORED_PIPELINE) }); + const res = await req(app, env, "/v1/instances/i1/pipelines", {}, await tok("u1")); + expect(res.status).toBe(200); + const body = (await res.json()) as any; + expect(body.pipelines).toHaveLength(1); + expect(body.pipelines[0]).toMatchObject({ name: "sweep", steps: 1, sink: "results", valid: true }); + }); + + it("404s when the instance isn't owned", async () => { + const { app, env } = testApp({ owned: false }); + const res = await req(app, env, "/v1/instances/i1/pipelines", {}, await tok("u1")); + expect(res.status).toBe(404); + }); +}); + +describe("POST /v1/instances/:id/pipelines/:name/run (issue #97)", () => { + it("owner-gated: 404s when the instance isn't owned (never kicks the workflow)", async () => { + const create = vi.fn(async () => ({ id: "wf" })); + const { app, env } = testApp({ owned: false, config: JSON.stringify(STORED_PIPELINE), create }); + const res = await req(app, env, "/v1/instances/i1/pipelines/sweep/run", { method: "POST", body: "{}" }, await tok("u1")); + expect(res.status).toBe(404); + expect(create).not.toHaveBeenCalled(); + }); + + it("404s an unknown pipeline name", async () => { + const { app, env } = testApp({ config: JSON.stringify(STORED_PIPELINE) }); + const res = await req(app, env, "/v1/instances/i1/pipelines/nope/run", { method: "POST", body: "{}" }, await tok("u1")); + expect(res.status).toBe(404); + }); + + it("kicks the durable workflow with the def + params and returns run ids", async () => { + const create = vi.fn(async () => ({ id: "wf-99" })); + const { app, env } = testApp({ config: JSON.stringify(STORED_PIPELINE), create }); + const res = await req(app, env, "/v1/instances/i1/pipelines/sweep/run", { method: "POST", body: JSON.stringify({ params: { repo: "owner/name" } }) }, await tok("u1")); + expect(res.status).toBe(200); + const body = (await res.json()) as any; + expect(body.ok).toBe(true); + expect(body.workflowId).toBe("wf-99"); + expect(body.runId).toBeTruthy(); + expect(create).toHaveBeenCalledTimes(1); + const arg = create.mock.calls[0][0] as any; + expect(arg.params.pipeline.name).toBe("sweep"); + expect(arg.params.params).toEqual({ repo: "owner/name" }); + expect(arg.params.trigger).toBe("api"); + expect(arg.params.userId).toBe("u1"); + }); +}); diff --git a/workers/api/src/routes/tools.ts b/workers/api/src/routes/tools.ts index db5d1d75..9d0664aa 100644 --- a/workers/api/src/routes/tools.ts +++ b/workers/api/src/routes/tools.ts @@ -3,6 +3,8 @@ import { HttpError, requireUser } from "../lib/auth.js"; import { requireOwnedInstance } from "./instances-runtime.js"; import { getRegistryTool, registryTools, runRegistryTool, type JsonSchema } from "../lib/tool-registry.js"; import { listConsents, revokeConsent, setConsent } from "../lib/connector-consent.js"; +import { startPipelineRun } from "../lib/pipeline-run-start.js"; +import { validatePipeline, type PipelineDef } from "../lib/pipeline.js"; import type { Env } from "../types.js"; /** @@ -79,6 +81,47 @@ toolRoutes.post("/:id/tools/:name", async (c) => { return c.json(result); }); +/** + * GET /v1/instances/:id/pipelines — declarative pipelines (#97) configured on this + * instance. Definitions live in the instance's `config.pipelines` (data, not code). + */ +toolRoutes.get("/:id/pipelines", async (c) => { + const session = await requireUser(c); + const instance = await requireOwnedInstance(c.env, c.req.param("id"), session.uid); + let cfg: Record = {}; + try { + cfg = JSON.parse(instance.config || "{}") as Record; + } catch { + /* malformed config → no pipelines */ + } + const pipelines = (cfg.pipelines && typeof cfg.pipelines === "object" ? cfg.pipelines : {}) as Record; + const list = Object.entries(pipelines).map(([name, def]) => ({ + name, + steps: Array.isArray(def?.steps) ? def.steps.length : 0, + sink: def?.sink?.collection, + valid: validatePipeline(def) === null, + })); + return c.json({ pipelines: list }); +}); + +/** + * POST /v1/instances/:id/pipelines/:name/run { params } — start a durable pipeline run + * (#97). Owner-scoped (requireOwnedInstance) + audited (startPipelineRun logs + * pipeline.requested with the caller's uid). Kicks the PipelineRunWorkflow and returns the + * run + workflow ids; results land in the trace/collection, not inline. + */ +toolRoutes.post("/:id/pipelines/:name/run", async (c) => { + const session = await requireUser(c); + const instanceId = c.req.param("id"); + await requireOwnedInstance(c.env, instanceId, session.uid); + const name = c.req.param("name"); + const body = (await c.req.json().catch(() => ({}))) as { params?: Record }; + const params = body.params && typeof body.params === "object" && !Array.isArray(body.params) ? body.params : {}; + const started = await startPipelineRun(c.env, instanceId, session.uid, name, params, "api"); + if (!started.ok) throw new HttpError(404, started.error); + return c.json({ ok: true, runId: started.runId, workflowId: started.workflowId }); +}); + /** GET /v1/instances/:id/connectors/consent — write-consents granted on this instance. */ toolRoutes.get("/:id/connectors/consent", async (c) => { const session = await requireUser(c); diff --git a/workers/api/src/types.ts b/workers/api/src/types.ts index d75173ef..17d4d1fa 100644 --- a/workers/api/src/types.ts +++ b/workers/api/src/types.ts @@ -15,6 +15,8 @@ export interface Env { JOB_APPLY: Workflow; /** Remote LLM brain that drives a local coding CLI toward an objective (AgentCoder port). */ CODING_SESSION: Workflow; + /** Durable runner for declarative data pipelines (issue #97) — walks a pipeline's steps. */ + PIPELINE_RUN: Workflow; /** WebSocket relay DO — one per instance, bridges cloud→runner without tunnels. */ RELAY: DurableObjectNamespace; GITHUB_CLIENT_ID: string; diff --git a/workers/api/src/workflows/pipeline-run.ts b/workers/api/src/workflows/pipeline-run.ts new file mode 100644 index 00000000..ee31fa84 --- /dev/null +++ b/workers/api/src/workflows/pipeline-run.ts @@ -0,0 +1,122 @@ +import { WorkflowEntrypoint, type WorkflowEvent, type WorkflowStep } from "cloudflare:workers"; +import { executePipelineStep, stepBind, type PipelineDef, type StepResult } from "../lib/pipeline.js"; +import { logError } from "../lib/error-log.js"; +import { logEvent } from "../lib/events.js"; +import { isTransientInfraError } from "../lib/transient-error.js"; +import type { Env } from "../types.js"; + +export interface PipelineRunParams { + instanceId: string; + userId: string; + /** The pipeline definition to run (resolved by the caller from instance config). */ + pipeline: PipelineDef; + /** Run parameters (from chat args, the trigger payload, or the API body). */ + params?: Record; + /** Groups all trace events for this run. */ + runId: string; + /** How the run was started, for the audit trail. */ + trigger?: "chat" | "api" | "trigger"; +} + +export interface PipelineRunResult { + outcome: "completed" | "failed"; + steps: number; + sunk?: number; + detail?: string; +} + +/** + * The durable pipeline runner (issue #97). Walks a declarative pipeline's steps IN ORDER, + * each in its own `step.do` so the run is durable + resumable past the 30s DO limit — the + * same machinery as JobApplyWorkflow / CodingSessionWorkflow. Each step dispatches its + * registry tool through runRegistryTool (inside executePipelineStep), so connector + * auth/grant/consent (#86/#90) are enforced identically to a direct tool call. Outputs + * thread between steps by `bind`; the final bound output feeds the optional `sink`. + * + * Resume-determinism caveat (mirrors job-apply): connector tokens are re-minted INSIDE each + * step.do (runRegistryTool → connectorClient runs per step) and NEVER captured across steps, + * so a resume after an isolate reset re-authenticates rather than replaying a stale token. + * The step call order is deterministic (linear walk, stable `s{i}` names), so replay is + * stable. + */ +export class PipelineRunWorkflow extends WorkflowEntrypoint { + async run(event: WorkflowEvent, step: WorkflowStep): Promise { + const { instanceId, userId, pipeline, params = {}, runId, trigger = "api" } = event.payload; + const env = this.env; + try { + await step.do("trace-start", async () => { + await logEvent(env, { source: "pipeline", event: "pipeline.start", message: `Run "${pipeline.name}" (${trigger})`, userId, instanceId, traceId: runId, context: { pipeline: pipeline.name, steps: pipeline.steps.length, trigger } }).catch(() => undefined); + return null; + }); + + // Bound outputs accumulate across steps; NOT captured across step.do closures as a + // mutable token — the value written here is journaled by the Workflow and replayed + // deterministically on resume. Connector auth is re-minted per step (see caveat). + const outputs: Record = {}; + let lastOutput: unknown = null; + + for (let i = 0; i < pipeline.steps.length; i++) { + const s = pipeline.steps[i]; + const bind = stepBind(s, i); + // Each step is a durable unit. On failure the runner records it and stops (a + // clean seam: #96's step library can add per-step retry/continue policy here). + // step.do requires a Serializable return; a step's `output` is arbitrary tool JSON + // (typed `unknown`), which doesn't satisfy that constraint at compile time even + // though it's plain JSON at runtime. Route the callback through `unknown` and cast + // the journaled result back to StepResult — same escape hatch job-apply uses. + const result = (await step.do(`s${i}-${s.tool}`, async () => (await executePipelineStep({ env, userId, instanceId }, s, i, outputs, params)) as unknown as Record)) as unknown as StepResult; + outputs[bind] = result.output; + lastOutput = result.output; + await step.do(`s${i}-trace`, async () => { + await logEvent(env, { source: "pipeline", event: "pipeline.step", level: result.success ? "info" : "warn", message: `${s.tool} → ${bind}: ${result.content.slice(0, 160)}`, userId, instanceId, traceId: runId, context: { step: i, tool: s.tool, bind, success: result.success } }).catch(() => undefined); + return null; + }); + if (!result.success) { + await step.do("trace-fail", async () => { + await logEvent(env, { source: "pipeline", event: "pipeline.end", level: "warn", message: `Failed at step ${i} (${s.tool}): ${result.content.slice(0, 160)}`, userId, instanceId, traceId: runId, context: { failedStep: i, tool: s.tool } }).catch(() => undefined); + return null; + }); + return { outcome: "failed", steps: i + 1, detail: `step ${i} (${s.tool}) failed: ${result.content.slice(0, 200)}` }; + } + } + + // Optional sink: upsert the final step's output into an instance collection (#91) + // via the AgentDO's records route — the same DO-fetch pattern job-apply uses to + // reach the instance. Each record is its own durable step so a large sink resumes + // mid-write. Dedupe/upsert-by-key is #96's job; here we insert. + let sunk = 0; + if (pipeline.sink) { + const records = Array.isArray(lastOutput) ? lastOutput : lastOutput != null ? [lastOutput] : []; + const collection = pipeline.sink.collection; + for (let r = 0; r < records.length; r++) { + const rec = records[r]; + if (rec == null || typeof rec !== "object") continue; + await step.do(`sink-${r}`, async () => { + const stub = env.AGENT.get(env.AGENT.idFromName(instanceId)); + const res = await stub.fetch(new Request(`https://agent/collections/${encodeURIComponent(collection)}/records`, { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ data: rec }) })); + if (!res.ok) throw new Error(`sink insert failed (${res.status}): ${(await res.text()).slice(0, 160)}`); + return null; + }); + sunk++; + } + } + + await step.do("trace-end", async () => { + await logEvent(env, { source: "pipeline", event: "pipeline.end", message: `Completed "${pipeline.name}": ${pipeline.steps.length} step(s)${pipeline.sink ? `, ${sunk} → ${pipeline.sink.collection}` : ""}`, userId, instanceId, traceId: runId, context: { steps: pipeline.steps.length, sunk } }).catch(() => undefined); + return null; + }); + return { outcome: "completed", steps: pipeline.steps.length, sunk }; + } catch (err) { + const msg = err instanceof Error ? err.message : String(err); + // A DO/isolate reset from a deploy is TRANSIENT — re-throw so the Workflow retries + // + resumes from its last completed step (same as job-apply); don't manufacture a + // crash on every deploy. + if (isTransientInfraError(msg)) { + await logEvent(env, { source: "pipeline", event: "pipeline.interrupted", message: `pipeline interrupted by a deploy, resuming: ${msg}`.slice(0, 200), userId, instanceId, traceId: runId }).catch(() => undefined); + throw err; + } + await logError(env, { source: "pipeline-run", userId, status: 500, message: `pipeline "${pipeline.name}" crashed: ${msg}`, context: { instanceId, runId, pipeline: pipeline.name, stack: err instanceof Error ? String(err.stack || "").slice(0, 1500) : undefined } }); + return { outcome: "failed", steps: 0, detail: msg }; + } + } +} diff --git a/workers/api/wrangler.toml b/workers/api/wrangler.toml index 6c26d63b..e2c5f17e 100644 --- a/workers/api/wrangler.toml +++ b/workers/api/wrangler.toml @@ -85,6 +85,12 @@ name = "pags-coding-session" binding = "CODING_SESSION" class_name = "CodingSessionWorkflow" +# Durable runner for declarative data pipelines (issue #97). +[[workflows]] +name = "pags-pipeline-run" +binding = "PIPELINE_RUN" +class_name = "PipelineRunWorkflow" + [[migrations]] tag = "v1" new_classes = ["AgentDO"]