Build .NET workflows

A pipeline is a stored definition of stages. A stage names a handler and optionally carries an input and execution options. Your worker registers that handler; the server schedules attempts and records their results.

The examples below use the options and deadline.Token from the quickstart. In a service, pass your request or application cancellation token instead. Add these imports:

using System.Text.Json;
using PipelogiqSDK.Abstractions;
using PipelogiqSDK.Api;
using PipelogiqSDK.Builders;
using PipelogiqSDK.Contracts;
using PipelogiqSDK.StageHelper;

Define a sequence

using var pipeline = PipelineBuilder.Create("process-order", options)
    .WithAction("validate", "ValidateOrderHandler", new { OrderId = "order-42" })
    .WithAction("reserve", "ReserveStockHandler", new { OrderId = "order-42" })
    .WithAction("notify", "NotifyOrderHandler");
var created = await pipeline.SendAsync(deadline.Token);

Register implementations of all three named handlers before running this recipe. Stages without explicit dependencies follow the server's sequential scheduling rules. The final example omits input and should use an IStageHandler implementation.

WithAction<THandler>("name", input) uses typeof(THandler).Name as the handler name. The string overload is useful when the producer does not reference your worker assembly. Use stage names that are unique within the pipeline; dependency names refer to stages, not handler classes.

Branch and join with explicit dependencies

using var pipeline = PipelineBuilder.Create("review-order", options)
    .WithAction("load", "LoadOrderHandler", new { OrderId = "order-42" })
    .WithAction("stock", "CheckStockHandler", options: new StageOptions
    {
        DependsOn = new List<string> { "load" }
    })
    .WithAction("credit", "CheckCreditHandler", options: new StageOptions
    {
        DependsOn = new List<string> { "load" }
    })
    .WithAction("decision", "DecideOrderHandler", options: new StageOptions
    {
        DependsOn = new List<string> { "stock", "credit" }
    });
var created = await pipeline.SendAsync(deadline.Token);

After load, the stock and credit checks become eligible independently. decision waits for both. They may run on separate workers; actual concurrency depends on workers and server policies. The SDK serializes dependencies, while the server implements scheduling. Prefer explicit DependsOn edges for a reviewable branch/join definition.

Read and update context

Stage input describes that stage's request. Pipeline context carries shared execution data across stages. A stage's output is not automatically deserialized into the next stage's typed input; write a context value and read it in the next handler when you need that handoff.

On the producer:

using var pipeline = PipelineBuilder.Create("invoice-import", options)
    .AddLabel("workflow", "invoice-import")
    .AddContextItem("invoiceId", "inv-42")
    .AddSensitiveContextItem("accessToken", "example-token")
    .WithAction("validate", "ValidateInvoiceHandler");
var created = await pipeline.SendAsync(deadline.Token);

Inside a handler:

internal sealed class ValidateInvoiceHandler : IStageHandler
{
    public Task<IStageResult> ExecuteAsync(IStageContext? context = null)
    {
        var invoiceId = context.TryGetValue<string>("invoiceId");
        if (string.IsNullOrWhiteSpace(invoiceId))
            return Task.FromResult<IStageResult>(
                StageResult.MissingRequiredData("An invoice ID is required."));

        context.AddItem("invoiceStatus", "validated");
        context.RemoveItem("temporaryValue");
        context.LogInfo($"Validated invoice {invoiceId}.");
        return Task.FromResult<IStageResult>(StageResult.Success("Invoice validated."));
    }
}

TryGetValue<T> returns the type's default when a value is absent or cannot be converted. Validate mandatory values explicitly. AddItem changes the working context; RemoveItem emits a deletion patch. Nested changes are also detected. Only changed keys are sent back, so an unchanged old value does not overwrite another branch's update. Concurrent writes to the same key still use last-applied-result semantics; use your database for transactional business state.

Labels classify runs and logs. Context holds execution values. Labels are strings and are not delivered as a separate handler payload. AddLabel retains the legacy pipelineKeywords wire field; AddKeyword is an obsolete alias. Do not put credentials in labels.

Sensitive context values are available to handlers and masked in public status views. Use context.AddSensitiveItem(...) to replace a secret in a real runtime context. Masking is not database encryption or a secret manager. Do not copy a sensitive value into a new ordinary key.

Classify failures and configure retries

using var pipeline = PipelineBuilder.Create("call-partner", options)
    .WithAction("send", "SendToPartnerHandler", new { OrderId = "order-42" },
        new StageOptions
        {
            TimeOut = 30,
            MaxRetries = 4,
            RetryInterval = 2,
            Backoff = "exponential",
            MaxRetryInterval = 60,
            Jitter = true,
            RetryOnErrorCodes = new List<string>
            {
                StageErrorCodes.RateLimitExceeded,
                StageErrorCodes.TransportUnavailable
            }
        });
var created = await pipeline.SendAsync(deadline.Token);

Timeout and retry intervals use seconds. Retry eligibility, delays, and policy overrides are decided by the server. Pass context.GetCancellationToken() into HTTP/database operations so handler work cooperates with timeout and lease loss.

A malformed typed input becomes a validation failure before business code runs. Use RunNextIfFailed only when continuing after that failure is an intentional workflow rule. It does not turn the failure into success.

Make creation and external effects repeatable

// Persist this key with the business operation before the first submission.
var operationKey = "order-42-fulfillment-v1";
using var pipeline = PipelineBuilder.Create("fulfill-order", options)
    .WithIdempotencyKey(operationKey)
    .WithAction("fulfill", "FulfillOrderHandler", new { OrderId = "order-42" });
var created = await pipeline.SendAsync(deadline.Token);
Console.WriteLine($"{created.Id}; existing={created.WasExisting}");

using var client = new PipelogiqApiClient(options);
var found = await client.GetPipelineByIdempotencyKeyAsync(operationKey, deadline.Token);

A creation key is scoped to the authenticated application. Reuse it for retries of the same logical request; do not generate a new key after an uncertain network response. Use an opaque key in real applications rather than embedding confidential business data. Treat a changed request under an existing key as a conflict, not an update operation.

Pipeline idempotency prevents duplicate pipeline creation. It cannot make an external payment, reservation, or email exactly once. Keep a stable external idempotency key and business state in your database. After an unknown outcome, query the external system before issuing another command. The insurance workflow example demonstrates this state-machine approach.

Append work from a handler

Return appended stages with the result so the server can apply them together. The newly named handler must already be registered on a worker.

internal sealed class ExpandHandler : IStageHandler
{
    public Task<IStageResult> ExecuteAsync(IStageContext? context = null)
    {
        var followUp = new StageInfo
        {
            StageName = "follow-up",
            StageHandlerName = "HelloHandler",
            Input = JsonSerializer.Serialize(new { Name = "from an appended stage" }),
            Options = new StageOptions { MaxRetries = 0 }
        };
        return Task.FromResult<IStageResult>(
            StageResult.Success("Follow-up planned.", new[] { followUp }));
    }
}

The lower-level AppendAgentStagesAsync(pipelineId, request, ct) also exists for application-managed append operations despite its agent-oriented name. Inside a handler, result-based append avoids splitting completion and follow-up creation into separate requests.

Pause for external approval

A normal handler can return a waiting result:

internal sealed class ApprovalGateHandler : IStageHandler
{
    public Task<IStageResult> ExecuteAsync(IStageContext? context = null)
        => Task.FromResult<IStageResult>(new StageResultDto
        {
            IsSuccess = true,
            IsWaitingForApproval = true,
            Result = "Waiting for the requested review."
        });
}

Your application presents the request to an authorized reviewer, loads the pipeline, validates that the selected waiting stage belongs to it, and calls the API:

using var client = new PipelogiqApiClient(options);
var pipeline = await client.GetPipelineAsync(pipelineId, ct);
var gate = pipeline.Stages!.Single(stage =>
    stage.Id == stageId && stage.Status == StageStatuses.WaitingForApproval);
await client.ResumeStageApprovalAsync(gate.Id, approved: true, ct: ct);

Here pipelineId, stageId, and ct are supplied by your application's authenticated review flow. Rejection uses approved: false and a nonempty rejectionReason. The server resolves the waiting stage; it does not rerun the gate handler. Generic rejection follows the workflow's failure/continuation rules. Built-in agent rejection has an additional recovery path described in AI agents.

Inspect and cancel

GetPipelineAsync returns pipeline status and detailed stages, including output, attempts, retry count, next retry time, last error code, failure disposition, logs, and execution ID where available. Use IsTerminal with PipelineStatuses.IsTerminal as a fallback. A paused pipeline is not terminal.

using var client = new PipelogiqApiClient(options);
await client.CancelPipelineAsync(pipelineId, ct);
var cancelled = await client.GetPipelineAsync(pipelineId, ct);

Cancellation changes orchestration state and fences outdated results. It cannot roll back a side effect already committed by your handler. Model refunds, releases, and other compensation as explicit business operations.

Events and application logs

EventBuilder.Create("order-updated", "OrderUpdatedHandler", options) creates a single event stage. PipelineBuilder.AsEvent<OrderUpdatedHandler>("order-updated") inserts an event stage into a pipeline. These builders do not accept a typed event input; supply context and use an input-free handler. Server 0.5.0 schedules events with ordinary dependencies, retries, execution metadata, and fencing.

For a log outside a stage:

using var log = LogBuilder.Create(
    Microsoft.Extensions.Logging.LogLevel.Information,
    "Import service started.", options).AddLabel("service", "invoice-import");
await log.SendAsync(deadline.Token);

Application logs support labels, not pipeline context. Use the context logging extensions inside a handler. For time-triggered workflows, continue to schedules.