Pipelines, Part 2: Drafts, Audit Trails, and the Interfaces Around Them.

In Part 1, we described the plan: a pipeline contract, a type system, three client interfaces, and a set of rules for keeping 18 specification documents consistent. That plan was written before any application code existed.

Three months and roughly four thousand commits later, the plan has met reality. Much of it held. Some of it changed, and the most important change came from a question the plan hadn’t fully answered:

When an AI agent edits a pipeline, who decides that the edit goes live?

This article describes what we built. It covers the pipeline as it exists today, how a run executes, how versions and their status are stored, how agent changes are tracked, and how browsers, programs, agents, dashboards and external applications all reach the same pipeline.

Implementation snapshot: based on the source reviewed on 8 November 2026. Examples use illustrative names. This describes the implementation, not a benchmark, and it doesn’t establish which revision is running in a particular deployment.

What a pipeline is now #

The core definition from Part 1 survived: a versioned, declarative graph of nodes, with parameters, pinned templates, named datasources, and explicit outputs. It grew in four directions.

More node types. The original three were DQL, DML, and DDL. Three more joined them:

Names became folder paths. A pipeline name has between 2 and 10 lowercase segments, such as finance/revenue/daily_by_region. A flat name is refused. Names act as stable identifiers: child pipelines, dashboards and execution history refer to them, so we don’t rename them. When an agent tries to create a new top-level folder, the server refuses with new_root_requires_confirmation and lists the existing roots. The agent must reuse one, or ask a person and resend with confirm_new_root: true.

The result node became explicit. The plan detected the result node from the graph’s shape. A review of the specifications in early August replaced that with an explicit rule: a DQL node without an output block returns its result to the caller, and a pipeline may have at most one caller node. A pipeline that only writes data elsewhere may have none.

Pipelines carry release checks. An optional checks list holds up to 20 assertions, each with a query and an expected value, range, or row set. Only the server runs them and records what it observed. There is deliberately no tool that lets a client submit an observed value.

Validation reports everything at once #

Part 1 promised that the executor would never receive an invalid pipeline. The validator keeps that promise in eleven rule groups: structure, graph, node types, references, parameters, checks, settings, composition, calculators, scripts, and lake delivery.

Validation is exhaustive rather than fail-fast. A save with seventeen problems returns all seventeen in one response, and a test holds the validator to that. An agent can fix the whole document in one pass instead of finding problems one at a time.

Two details show the kind of care this needs.

Templates are rendered at save time. Static analysis can’t find every variable a template uses, because a reference can be conditional, come from an imported macro, or be built by an expression. So the validator builds sample values from the declared parameters and renders every referenced template. An undeclared variable becomes template_parameter_undeclared when the pipeline is saved, not when it runs.

Cycle detection is iterative. The first version used recursive depth-first search. A chain a couple of thousand nodes long overflowed the stack, so a request body of about 2 KB could crash validation. The detector is now iterative, and when it finds a cycle it reports the path, such as a -> b -> c -> a.

How a run executes #

Execution follows depends_on and nothing else. The executor runs each node as a Kotlin coroutine. Each node first waits for its dependencies and only then takes one of a limited number of parallel slots. The order matters: if a node took a slot first and then waited, waiting nodes could hold every slot and stall a long chain permanently. A named test guards that ordering.

The same care shows up in the failure paths:

• Failure and cancellation are different outcomes. One failing node cancels its siblings, and the execution ends FAILED. A user or system cancellation ends ABORTED and records why: client_disconnect, cancelled, or shutdown. A node that never started is recorded as ABORTED, never as FAILED.
• Timeouts are layered. A whole execution has a time budget. Each node has a wall-clock deadline, and each statement has its own timeout, resolved from the node, the pipeline, the datasource, or the dialect. An override outside the permitted range is refused when the pipeline is saved, not silently clamped. An author who asks for four hours should learn immediately that they can’t have it.
• Admission refuses rather than queues. Each user and each instance has a limit on concurrent executions. A request over the limit gets pipeline.execution.concurrency_limit immediately instead of waiting in an invisible queue.
• Crashed instances are cleaned up. Instances write a heartbeat. A sweep marks a RUNNING execution ABORTED, with the reason pipeline.execution.instance_lost, once its instance has been silent for three heartbeat intervals.

The caller’s result is stored in Redis before the source connection closes, with a time-to-live (300 seconds by default) and a 100 MB size cap. Before reporting success, the executor checks that the stored result can be read. If a successful run’s result had quietly disappeared, the client couldn’t tell it apart from a pipeline that legitimately returns nothing.

Versions, drafts, and status #

The plan said versions were immutable and that each edit created a new version. That worked for people. Once agents began writing pipelines, it wasn’t enough. If every agent edit became the next live version, a dashboard could start reading an agent’s work-in-progress without anyone deciding it should.

So we added a draft-and-release lifecycle. Each version row now has one of three statuses:

The pipeline itself has no status column. Its status comes from its versions and a current_version pointer, so a separate flag can never drift out of step with the rows it describes.

Writing a draft #

Every authored create lands as a draft, whether it comes from the editor, REST, or MCP. The only exception is a promotion import, which arrives already released from the deployment where it was reviewed. The choice is a required argument in code, with no default. A new create path that forgets to make it is a compile error, not a pipeline that went live without review.

The first edit after a release copies the released body into a new draft. Later edits overwrite that draft in place. An edit identical to the released body creates nothing.

Every write is protected by a body hash: a SHA-256 digest of the canonical JSON body, computed by the database rather than supplied by the client. A REST client sends the hash it last read in an If-Match header. An MCP client sends it as expected_hash. If someone else has changed the draft in the meantime, the write is refused with pipeline.version.conflict, and the refusal includes the current state.

The expected response is to re-read, rebase, and try again. The instructions we give agents say this directly, because a blind retry would overwrite a person’s edit.

Running a draft #

Without an explicit version, an execution runs the working version: the draft if there is one, otherwise the current release. That rule lives in one function. It was once three copies in three places that could drift apart. This lets an agent test its own draft without affecting anything that runs the released version.

Releasing #

Release is a separate, deliberate act. It runs in one database transaction under a workspace-level guard:

• Dependencies first. A release is refused if it would pin a template version that is still a draft, unless the person explicitly includes that template in the same release.
• Checks gate the release. The server runs the pipeline’s release checks fresh. A failing check blocks the release. A person can override it only by giving a written reason of at least ten characters, and that reason is audited.
• Audit in the same transaction. The audit row is written in the same transaction as the release. If the audit write fails, the release doesn’t happen.

Not just pipelines #

Templates, parameter sets, visualizations and dashboards follow the same lifecycle: the same three statuses, the same hash precondition, the same refusal codes with a family prefix, and audit event names generated from one pattern. A release can include artifacts from all five families when they depend on each other.

Keeping track of what agents change #

“Agents author, people release” is only a policy unless the system enforces it. We enforce it in layers, so that no single forgotten check undoes it.

What an agent cannot do #

• No tool releases anything. The MCP server has 61 tools. None of them releases, promotes, switches the current version, restores, or discards.
• An agent’s key cannot reach the routes that do. Datapipelines has four kinds of API key, and each works only on its own surface. An MCP key reaches /mcp and nothing else. An endpoint key reaches the published endpoints bound to it. A dashboard key reaches the runtime of the dashboards its bindings cover. A server key reaches the promotion routes between deployments. The restriction is applied centrally, so a newly added route can’t become reachable to a key because someone forgot a check.
• Lifecycle actions require a person. Discarding, restoring, switching and purging require an interactive sign-in session.
• Agents never handle credentials. There is no tool for creating a datasource. A credential passed through a tool call would pass through the agent’s context, its transcript, and whatever its client logs. People register datasources, and agents use them by name.

What the record shows #

Each version row records its writer and surface. Version rows record who created and last updated them, and through which surface: session, api_key, or mcp. A database constraint limits the values. The Versions tab shows that surface beside each version. API keys act as their own service identities, so an agent’s change is attributed to the agent’s key, not to whoever created the key.

Every MCP call is audited. Every MCP request passes through one dispatcher, and that dispatcher records an mcp.tool.called audit event for each call. Calls to tools that change state also record mcp.tool.write. Each record includes:

• the tool and its target
• the outcome and any error code
• the correlation ID
• the elapsed time

The free-form SQL probe tool records a SHA-256 hash and the length of the SQL, never the statement itself. Because the audit happens at the single point of entry, a new tool can’t skip it.

Lifecycle events are their own records. Release, discard, restore, purge, and current-version switches write named events such as pipeline.version.released, template.version.discarded, and dashboard.current_switched.

Every execution records its origin. Executions record who started them and through which trigger: UI, REST, MCP, PIPELINE, ENDPOINT, SCHEDULE, or DASHBOARD. When a key started the run, the execution also records the key’s kind.

The audit log also enforces a rule: only the credential that started an execution through MCP, or someone with the right permission, may cancel it.

Where the record is thinner #

Our current tracking has limits, and they’re worth stating plainly.

• A draft records only its last writer, not every intermediate edit.
• MCP call auditing is best-effort. If the audit write fails, the tool call is not blocked. The failure is logged as a warning. Release auditing, by contrast, is strict.
• Audit records are retained for 365 days by default, and an operator can shorten that to no less than 30.

One application layer, many surfaces #

The plan described MCP as a thin adapter over REST. We changed that early. The MCP server doesn’t call the REST API. Both surfaces call a shared application layer below them, so idempotent launches, working-version resolution, permission checks, and audit are implemented once rather than copied.

Every REST handler and every MCP tool declares exactly one permission from a single catalog. Each entry in the MCP tool catalog also declares whether it changes state. That property is declared, never guessed from the tool’s name, because a tool that writes but is declared read-only is a security hole.

REST: predictable at the edges #

The REST API kept the envelopes from the plan. A success is {schema_version, correlation_id, data}. A failure carries code, message, user_message, details, and doc_url. Application errors pass through one exception handler. Authentication rejections, which happen before a request reaches any controller, are written by a dedicated writer that emits the same envelope. A client parses one shape everywhere.

Some of the rules are small but intentional:

• Custom headers use a DP- prefix: DP-Correlation-Id, DP-Result-TTL-Seconds, DP-Result-Page-Rows. Idempotency-Key keeps its standard name.
• A retried execute joins the original run. A retry carrying the same Idempotency-Key attaches to the existing execution instead of starting another one.
• Malformed bodies reveal nothing. A malformed execute body is rejected with 400 before the pipeline is looked up, so a bad request can’t be used to learn which pipeline IDs exist.
• Large numbers fail as the caller’s error. A number like 1e10000 is a 400 for the caller, not a 500.
• Failures put the root cause last. A failed execution carries an exception chain in error.exception.caused_by, ordered from outermost to root cause. Depth and stack frames are both capped.

SSE: the execution as a stream of facts #

POST /api/v1/pipelines/{id}/execute responds with an event stream. Nine event types describe an execution:

execution_started, node_started, node_progress, node_completed, node_failed, pipeline_completed, pipeline_failed, execution_aborted, and data_ready.

The ordering is a contract:

• Each node emits exactly one node_started, zero or more node_progress samples, and exactly one node_completed or node_failed.
• Each execution ends with exactly one terminal event.
• When the caller has a result, pipeline_completed is followed by data_ready, which carries the first page of rows and a URL for the rest.

A heartbeat comment every 15 seconds keeps idle connections open through proxies. If the client disconnects, the execution gets a 30-second grace period before it is cancelled. Because the stream starts with a POST, the browser reads it with fetch and a streaming body reader rather than EventSource.

The plan proposed resuming a stream from its last event ID. We replaced that with replay. Events go to a Redis log kept for one hour after completion, and to durable database rows kept for seven days by default. GET /api/v1/executions/{id}/events replays and then follows the log, removing duplicates. Expiry and outage are reported differently. An expired log is 410 result.expired. A storage fault is 503 result.storage_unavailable, because “gone” and “temporarily unreachable” call for different responses.

MCP: built for agents #

The MCP server uses the protocol’s stateless Streamable HTTP transport at /mcp, pinned to protocol version 2025-06-18. The catalog has grown from 15 planned tools to 61:

Several tools are shaped around the way an agent works:

• pipelines_execute_node is for debugging one node, not for running a pipeline. It renders one node’s SQL, runs it against that node’s datasource, and returns up to 50 rows. It creates no execution record. It fills missing parameters with sample values and lists which ones it filled.
• executions_get_result pages results. Large pages, CSV, and Arrow come back as a cursor URL rather than inline, so a big result doesn’t flood the agent’s context.
• pipelines_run_checks records what the server observed. The agent can ask for a check run, but it can’t supply the answer.

The server also teaches agents how to use it. When an agent connects, it receives instructions: author drafts, introspect before writing SQL, and rebase on conflict. A failed execution’s root cause is the last entry of error.exception.caused_by, and the agent should report it together with the correlation ID. The full manual is available three ways: as an MCP resource, through docs_get, and as plain Markdown over HTTP at /skill.md. The last is for clients that don’t support MCP resources.

The semantic layer article describes how agents record what they learn about data through semantics_record.

Dashboards, published endpoints, and schedules #

Pipelines are also consumed by things other than people and agents.

Dashboards pin released pipelines. A dashboard source names a pipeline and a version, and saving a dashboard that points at an unreleased version is refused with dashboard.validation.source_not_released. A dashboard therefore never displays an agent’s draft. Visualizations render with Plotly, as tables, or as KPI tiles, and dashboard refreshes stream over SSE. Cascading dashboard parameters covers how dashboard inputs are resolved on the server.

Published endpoints turn a pipeline into an API. endpoints_create publishes a released pipeline as a GET endpoint under /api/<category>/<version>/<path>. The pipeline must be free of side effects, including in any child pipelines it runs. A node that writes to a registered datasource is refused with endpoint.pipeline_not_readonly. The reason is that browsers, crawlers and proxies retry, prefetch and crawl GET requests freely, so serving one must never change data. Every other method gets 405 with Allow: GET. Only endpoint keys bound to the path can call it.

Schedules run pipelines on a cron expression in a named time zone. Scheduled runs appear in execution history with the SCHEDULE trigger, beside every other kind of run.

What held, and what changed #

The registries from Part 1 became tests. Specification-drift tests keep error codes, calculator kinds and the name grammar in agreement with their documents, and an audit script checks that every configuration key and error code has exactly one authoritative home.

Current limits #

• Results are short-lived. Stored results expire after at most an hour, and a large or long-lived result means running the pipeline again.
• Concurrency limits are per instance. There is no cross-instance limit, so several replicas admit several times the configured limit.
• List parameters aren’t available yet. The contract declares multi-value parameters, but saving one is refused.
• pipelines_execute blocks. An MCP client receives the result when the run ends, without intermediate progress.
• Reconnecting a stream replays it. A reconnecting client reads the stream from the start rather than resuming where it left off.
• Draft history is shallow. It records the last writer, not each intermediate edit.
• Distributed tracing isn’t shipped. OpenTelemetry remains a design target.

The thread through both parts #

Part 1 argued that agents need a contract precise enough to obey. Part 2 adds what that contract couldn’t provide alone: a boundary between proposing a change and accepting it.

Agents can do a great deal in Datapipelines. They can inspect datasources, write templates, assemble pipelines, run them, record what they learn, and publish endpoints for pipelines that are already released. What they can’t do is make their own work live. A person releases it, the system records who did what and through which surface, and the release leaves an audit record written in the same transaction.

That boundary is enforced by the system, not by agents remembering an instruction.