Declarative pipeline definition + durable runner (#97) - #106
Merged
Conversation
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) <noreply@anthropic.com>
Closed
5 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Closes #97. Part of #94. Builds on #85 (ToolDef) + #86 (connectorClient).
A pipeline is JSON data:
{name, params?, steps:[{tool, inputs?, bind?, forEach?}], sink?}.inputstake literals or refs —{$ref:'bind.dotted.path'}(prior step) /{$param:'name'}(run param);forEachfans a step over an array (concurrency deferred to #96). Definitions live per-instance inconfig.pipelines[name](no migration).PipelineRunWorkflow(WorkflowEntrypoint, mirrors job-apply.ts): linear walk, each step its ownstep.dofor durability past the 30s DO limit; results journaled bybind. Connector tokens re-minted inside each step (never captured across steps) so resume/replay is safe. Every step dispatches via the singlerunRegistryToolpath — auth/grant/consent enforced identically to direct tool calls. Optionalsinkupserts final output into an instance collection (clean seam for #96 dedupe/upsert).Start paths:
run_pipelineLLM tool (in BASE allowlist → 'sweep Sydney' works) + owner-scoped, auditedPOST /v1/instances/:id/pipelines/:name/run. Trigger(#92) seam:startPipelineRun(...trigger)present, wiring deferred.Verify:
tscclean · 818 api tests pass (0 regressions) · biome clean.🤖 Generated with Claude Code