lace-app-sdk

Pipelines — typed workflow graphs

Versioned graphs of typed steps with checkpoints, retries, idempotency, and dead-letter handling. The Studio compiles the same contract the API executes.

What a pipeline is

A pipeline is a PipelineDefinition: an ordered graph of PipelineSteps, each with an input/output JSON Schema, a uses reference (which tool / LLM / sub-pipeline / control-flow primitive it invokes), and orchestration policy (retry, idempotency, timeout, checkpoint). Definitions are versioned — v1.0 keeps flat backoff; v1.1+ opts into exponential backoff and default transient-error classification.

Declaring a pipeline

Implement an AppPipelineProvider and expose it in your manifest:

pythonapp/pipelines.py

from lace_app_sdk.pipelines import AppPipelineProvider
from lace.pipeline.definition import PipelineDefinition, PipelineStep

class IntakePipelines(AppPipelineProvider):
    app_id = "acme.field_intake"

    def pipeline_definitions(self):
        return [
            PipelineDefinition(
                pipeline_id="field_intake.triage",
                version="1.0",
                steps=[
                    PipelineStep(step_id="normalize", uses="acme.normalize:v1"),
                    PipelineStep(step_id="classify", uses="llm.classify:v1",
                                     retry={"max_attempts": 3, "backoff": "exponential"}),
                    PipelineStep(step_id="route", uses="branch",
                                     config={"branches": {"urgent": "escalate", "default": "reply"}}),
                ],
            )
        ]

In app/manifest.py:

MANIFEST = LaceAppManifest(
    app_id="acme.field_intake",
    pipeline_providers=["app.pipelines:IntakePipelines"],
    ...
)

Control flow

Steps are not just linear — the pipeline compiler (src/lace/pipeline/compiler.py & control_flow.py) understands:

Primitive	What it does
`branch` / `switch`	Conditional fan-out to named branches
`foreach` / `parallel_foreach`	Iterate over a collection, sequentially or in parallel
`parallel`	Run steps concurrently and join
`reduce` / `repeat_until`	Accumulate / loop until a condition
`trycatch` / `fallback`	Handle step failures without aborting the graph
`delay` / `wait_until` / `wait_for_input`	Time and human-input gates
`return`	Early exit with a value

Runs that finish

Every step is checkpointed as it completes (src/lace/pipeline/runtime.py). A crash, timeout, or restart resumes from the last completed step — not from the start, and not by disappearing. Idempotency keys make a retried step safe to re-execute, and work that cannot succeed lands in a dead-letter queue where it can be inspected instead of silently dropped. Traces are emitted per step (src/lace/pipeline/trace.py & trace_report.py) and surfaced in the Studio.

Workflow Studio

Studio is a visual compiler over the same PipelineDefinition schema. You drag steps, bind inputs/outputs, set retry policies, and Studio emits YAML that the runtime executes verbatim. There is no "Studio-only" pipeline — the file is the contract.

Invoking a pipeline

From code: POST /v1/pipelines/{pipeline_id}/runs (or via an app tool that starts one).
From an agent: declare a PipelineCapability on an AgentDefinition so the model can dispatch to it.
On a schedule: declare an AppScheduleSpecDeclaration in the manifest with invoke_path pointing at a pipeline trigger route.

Next: routes & UI or workflows & pipelines overview.