Orchestration Persistence

tramai-orchestration keeps workflow persistence storage-agnostic. The workflow runtime owns step ordering, checkpoint timing, and resume rules — your application chooses how state is serialized and where checkpoints live.

What This Covers

• WorkflowStateCodec — encode/decode typed workflow state
• WorkflowCheckpointStore — store revisioned checkpoints
• WorkflowPersistence — wire codec and store together
• Reference stores: InMemory, File, Markdown, JDBC
• Collision-free on-disk layouts and legacy migration
• Safe persistence failure boundaries
• Multi-node ownership with WorkflowLeaseStore
• File-backed and JDBC lease guidance

When to Use Persistence

Add persistence when:

• workflows need to survive process restarts
• workflows run longer than the JVM's expected lifetime
• you need checkpoint-based resume capability
• multiple workers share the same workflow queue

Core Pieces

WorkflowStateCodec

The codec is the right place to choose JSON, YAML, protobuf-over-base64, or a domain-specific format.

data class ReviewState(
    val requestId: String,
    val draft: String? = null,
    val approved: Boolean = false,
)

object ReviewStateCodec : WorkflowStateCodec<ReviewState> {
    override fun encode(state: ReviewState): String = listOf(
        state.requestId, state.draft.orEmpty(), state.approved.toString(),
    ).joinToString("|")

    override fun decode(payload: String): ReviewState {
        val parts = payload.split("|", limit = 3)
        return ReviewState(
            requestId = parts[0],
            draft = parts.getOrNull(1).orEmpty().ifBlank { null },
            approved = parts.getOrNull(2)?.toBooleanStrictOrNull() ?: false,
        )
    }
}

For richer state, JSON is usually the most practical choice.

WorkflowCheckpointStore

Implement this SPI when you need a custom storage backend:

class MyCheckpointStore(
    private val client: CheckpointClient,
) : WorkflowCheckpointStore {
    override suspend fun load(
        workflowName: String,
        workflowId: String,
    ): WorkflowCheckpoint? = client.read(workflowName, workflowId)

    override suspend fun save(
        checkpoint: WorkflowCheckpoint,
        expectedRevision: Long?,
    ): WorkflowCheckpoint {
        return client.write(checkpoint, expectedRevision)
    }

    override suspend fun delete(
        workflowName: String,
        workflowId: String,
        expectedRevision: Long?,
    ) { client.delete(workflowName, workflowId, expectedRevision) }
}

Revision contract:

• first save: expectedRevision = null
• later saves: use the current revision
• stale writers: fail with WorkflowCheckpointConflictException

This gives database, object-store, and filesystem implementations the same optimistic-concurrency model.

Wiring a Workflow

val persistence = WorkflowPersistence(
    checkpointStore = MyCheckpointStore(client),
    stateCodec = ReviewStateCodec,
)

// Start
val result = workflow.run(
    initialState = ReviewState(requestId = "invoice-123", approved = true),
    persistence = persistence,
)

// Resume
val result = workflow.resume(
    context = WorkflowContext(workflowId = "invoice-123"),
    persistence = persistence,
)

Resume Compatibility Contract

Resume is intentionally strict, and this is the part of persistence most likely to surprise you in production. A checkpoint may resume only when all of the following hold:

• the checkpoint exists for the same workflow name and workflowId
• it was written with the runtime's required definition metadata
• the current workflow declares the same explicit definitionVersion
• the current definition digest matches the persisted digest

The persisted digest is structural: workflow name, top-level and nested step topology, step names and branch keys, and stop-policy compatibility inputs such as max step executions and max parallel branches.

Practical consequences:

• changing workflow structure without bumping definitionVersion still fails loudly on resume
• bumping definitionVersion intentionally also fails loudly against older checkpoints
• checkpoints created before this metadata contract are rejected rather than resumed heuristically

Treat definitionVersion as an operator-controlled compatibility boundary, not a comment.

Reference Stores

The repository includes these reference implementations:

These are conveniences, not the architectural boundary. You can replace them for Postgres, S3, GCS, Redis, or a domain-specific store.

Checkpoint File Layout

FileWorkflowCheckpointStore and MarkdownWorkflowCheckpointStore persist one file per checkpoint under the store root. Since 0.6.0 the path is derived with CollisionFreeWorkflowCheckpointPathStrategy: each identity segment (workflowName, workflowId) is URL-safe Base64 without padding, so every distinct logical key maps to a distinct file:

<root>/<base64url(workflowName)>/<base64url(workflowId)>/checkpoint.properties
<root>/<base64url(workflowName)>/<base64url(workflowId)>/checkpoint.md

This matters if you store arbitrary identifiers. The strategy it replaces sanitized segments to [A-Za-z0-9_-], which lossily collapsed keys such as order/a and order?a onto the same path. Records written by the older strategy remain readable: a load tries the canonical path first, then the legacy sanitized path, and only accepts a legacy record whose contents identify the requested workflow — the first legitimate update migrates the record and deletes the legacy file, so there is never a second authoritative copy.

Lease File Layout

FileWorkflowLeaseStore writes one lease.properties per workflow. Since 0.6.0 the path uses CollisionFreeWorkflowLeasePathStrategy: segments already inside the legacy-safe domain keep their existing path unchanged (ordinary UUID workflow ids do not move), and any other segment is encoded as ~ plus URL-safe Base64. Because ~ is outside the legacy-safe domain, a canonical path can never collide with a legacy path — two distinct workflows cannot address the same lease file, and storage layout cannot redefine ownership.

A live legacy lease is deliberately not migrated: the filesystem lock namespace must not change under an active lease. After release or expiry, the next claim uses the canonical path.

Safe Failure Boundaries

Built-in persistence stores expose fixed-text public failures. They do not copy filesystem paths, SQL, persisted payloads, or arbitrary storage exception messages into public exceptions, ordinary worker callbacks, or default logs. The original failure is available only through PersistenceFailureDiagnosticObserver, configured with the additive observer constructor overload:

val checkpointStore = FileWorkflowCheckpointStore(
    Path.of(".tramai/workflows"),
    DefaultWorkflowCheckpointPathStrategy("checkpoint.properties"),
    PersistenceFailureDiagnosticObserver { event ->
        // Route diagnostic-only event.failure to an access-controlled sink.
        diagnosticSink.record(event.resourceKind, event.operation, event.failureCode, event.failure)
    },
)

The observer is fail-open — an observer failure does not replace the persistence failure — and cancellation still wins: genuine coroutine cancellation propagates instead of becoming a persistence diagnostic. This is a constructor setting, not a property.

Checkpoint conflicts, lease conflicts, invalid checkpoints, invalid step-attempt records, and general read/write/delete/list failures all carry fixed public text; StaleWorkflowLeaseException remains the semantic fencing signal. Worker observers receive safe failures, and LoggingTramaiWorkerObserver logs exception class names only. Custom stores should preserve the same separation when they handle raw backend errors.

Multi-Node Ownership

Revision checks prevent stale writes but do not guarantee single active execution. For that, use WorkflowLeaseStore and WorkflowLeasePolicy.

val persistence = WorkflowPersistence(
    checkpointStore = MyCheckpointStore(client),
    stateCodec = ReviewStateCodec,
    leaseStore = MyLeaseStore(client),
    leasePolicy = WorkflowLeasePolicy(
        ownerId = "worker-7",
        leaseDurationMillis = 30_000,
    ),
)

Lease semantics

• run(...) claims the workflow before checkpointing starts
• resume(...) claims the workflow before resumed execution starts
• each successful checkpoint write renews the lease with the latest revision
• success or failure releases the lease

Lease Store Contract

class MyLeaseStore(
    private val client: LeaseClient,
) : WorkflowLeaseStore {
    override suspend fun currentLease(
        workflowName: String, workflowId: String,
    ): WorkflowLease? = client.current(workflowName, workflowId)

    override suspend fun claim(
        workflowName: String, workflowId: String,
        ownerId: String, checkpointRevision: Long?,
        leaseDurationMillis: Long,
    ): WorkflowLease = client.claim(workflowName, workflowId, ownerId, checkpointRevision, leaseDurationMillis)

    override suspend fun renew(
        lease: WorkflowLease, checkpointRevision: Long?,
        leaseDurationMillis: Long,
    ): WorkflowLease = client.renew(lease, checkpointRevision, leaseDurationMillis)

    override suspend fun release(lease: WorkflowLease) {
        client.release(lease)
    }
}

If another executor already owns the workflow, the store should fail with WorkflowLeaseConflictException.

File-Backed Leases

For local tooling or single-filesystem deployments:

val persistence = WorkflowPersistence(
    checkpointStore = FileWorkflowCheckpointStore(Path.of(".tramai/workflows")),
    stateCodec = ReviewStateCodec,
    leaseStore = FileWorkflowLeaseStore(Path.of(".tramai/workflows")),
    leasePolicy = WorkflowLeasePolicy(
        ownerId = "worker-local-1",
        leaseDurationMillis = 30_000,
    ),
)

Good for: one host or shared filesystem, inspectable artifacts, simple deployments. Not for: distributed deployments with weak filesystem semantics.

JDBC Lease Guidance

For multi-node backends centered on a relational database:

CREATE TABLE tramai_workflow_lease (
    workflow_name VARCHAR(255) NOT NULL,
    workflow_id VARCHAR(255) NOT NULL,
    lease_id VARCHAR(255) NOT NULL,
    owner_id VARCHAR(255) NOT NULL,
    checkpoint_revision BIGINT NULL,
    acquired_at_epoch_millis BIGINT NOT NULL,
    expires_at_epoch_millis BIGINT NOT NULL,
    PRIMARY KEY (workflow_name, workflow_id)
);

Key operational rule: lease ownership is time-bound. An active row with expires_at_epoch_millis > now blocks another owner. An expired row may be replaced.

Reference SQL patterns:

Claim — insert with PK. If PK exists and not expired, fail with conflict.

Renew — update where lease_id = ? AND owner_id = ? AND expires_at_epoch_millis > ?. Zero rows means conflict.

Release — delete where workflow_name, workflow_id, lease_id, owner_id.

Implementation notes:

• index (owner_id) for operator visibility
• use the same clock source for claim and renewal decisions
• keep lease duration larger than the expected time between checkpoint writes
• include checkpoint_revision for operator correlation
• treat zero-row renewals as conflicts, not silent no-ops

If you implement JDBC stores yourself, note that TramAI's own JDBC stores wrap raw driver operations so that SQLException becomes a sanitized IllegalStateException while CancellationException and domain errors propagate unwrapped. Do not catch java.sql.SQLException and rethrow it verbatim into application-visible failures.

Choosing a Backend

Limitations

• checkpoints happen only at top-level workflow step boundaries
• nested branch internals are not resumed mid-step
• in-flight parallel work is not resumed mid-branch
• partially emitted provider streams are not resumed token-by-token

Next Steps

• Orchestration — workflow DSL, step types, engine boundary
• Scheduling — cron execution and delay steps
• Workflow Server — expose persistent workflows over HTTP
• API and Binary Compatibility — the preview stability class