Lineage delivery

The SDK sends the workspace and new events when a @phase function exits and when the @workflow function exits, after success or failure. A @step exit does not send anything.

Delivery at a boundary has a 30-second deadline. If it fails, the SDK logs the failure and the workflow continues. Events stay in memory, and the next boundary sends them again.

In a long async phase, the SDK also sends events while the phase runs, each time 5,000 events are waiting. To send them earlier, or from code that is not async, call Audit.flush():

from unitelabs.sdk import Audit, get_logger

logger = get_logger(__name__)
result = await Audit.flush()

if not result.ok:
    logger.warning("Lineage delivery incomplete: %d events pending: %s", result.pending, result.error)

To fail the run when events can't be sent, use strict mode. It raises LineageDeliveryError after the retries:

await Audit.flush(strict=True, max_attempts=6, backoff=5)

Sent events leave memory

After a flush sends events, the SDK removes them from ctx.state["events"]. This keeps the memory of a long run small. ctx.state["events"] holds only the events that were not sent yet. The Platform holds the sent ones. To read all events of a run, see Read lineage events.

To keep the sent events in ctx.state["events"] too, use with Audit.keep_events():, or set UNITELABS_LINEAGE_KEEP_EVENTS=1 for the whole process. The memory then grows with each event.

Next steps

• Data lineage: what lineage is and how the Platform shows it.
• Read lineage events through the API or live event stream.
• Emit lineage events for devices the library doesn't drive.