Events

Every status change in a running workflow fires an event: a thread starts, an action completes, a device faults. This page is for anyone hooking something up to that stream, whether that is a live view of a run, an audit trail, a dashboard, or a step inside a workflow that waits for something else to happen first. If you are not integrating anything, you can skip it.

What you would use events for

• Watch a run from outside the process. The daemon serves the stream over HTTP: poll GET /events, or subscribe to GET /events/stream over Server-Sent Events. See Daemon REST surfaces.
• Feed an external system. Register a sink on the runtime and every event from every execution arrives at your code: a JSONL audit file, a WebSocket fan-out, a downstream orchestrator. See Sinks.
• Make one step wait for another. Inside workflow code, an action emits a named event and another action, thread or branch waits on it. See Custom events from actions.

Two layers

There are two layers. Which one you want depends on where your code sits: inside a workflow, subscribing to events from this workflow's executions, or outside it, taking every event from every execution.

Layer	Type	Where to subscribe	Scope
EventBus (per-workflow)	event_name + ExecutionContext	wf.on(event_name, handler) inside @orca.workflow	Events from this workflow's executions only.
System-level IEventSink	RuntimeEvent (Pydantic, JSON-serializable)	runtime.register_sink(sink)	Every event from every execution. The daemon's /events/stream SSE feeds off this layer.

A _SystemEventForwarder promotes each per-workflow EventBus event into a RuntimeEvent and pushes it to every registered sink. You do not have to wire that yourself. A few events skip the workflow bus and go straight onto the system bus, because they are not about any one execution: incidents and device faults.

Event name shape

Events are dot-separated. Two forms:

<entity_type>.<entity_id>.<status>      # specific (most common)
<entity_type>.<status>                  # generalized (subscribe-friendly)

When the engine emits THREAD.{thread_id}.STARTED it also fans out to any handler subscribed to the generalized THREAD.STARTED. A sink that wants every thread start across the system subscribes to the generalized form once. A workflow that wants to react to one specific thread subscribes to the 3-part form.

Entity types and their statuses

Entity	Status enum	Statuses
ACTION	ActionStatus	CREATED, RESOLVED, AWAITING_LOCATION_RESERVATION, AWAITING_CO_THREADS, EXECUTING_ACTION, AWAITING_MOVE_RESERVATION, PREPARING_TO_MOVE, PICKING, PLACING, COMPLETED, ERRORED, SKIPPED, ABORTED
METHOD	MethodStatus	CREATED, IN_PROGRESS, COMPLETED, SKIPPED, PARTIAL_COMPLETE
THREAD	LabwareThreadStatus	CREATED, AWAITING_MANUAL_PLACE, RESOLVING_ACTION_LOCATION, AWAITING_ACTION_RESERVATION, ACTION_LOCATION_RESOLVED, AWAITING_MOVE_RESERVATION, AWAITING_MOVE_TARGET_AVAILABILITY, MOVING, AWAITING_CO_THREADS, EXECUTING_ACTION, AWAITING_EVENT, AWAITING_MANUAL_REMOVE, PAUSED, COMPLETED, STOPPING, STOPPED, ABORTED, FAILED (crash-only: the thread's task died on an unhandled error)
WORKFLOW	WorkflowStatus	CREATED, IN_PROGRESS, COMPLETED, ERRORED
EXECUTION	runtime-level	COMPLETED, FAILED, ABORTED (terminal only)
SUBMISSION	SubmissionStatus	ACCEPTED, then the submission's own terminal status

Engine-emitted special events

Event	When
INCIDENT.<incident_id>.<category>	Any incident was recorded. Summary fields only; the full typed detail stays on the incidents surface, keyed by incident_id.
DEVICE.<device_name>.FAULTED / .FAULT_CLEARED	A device fault was latched or cleared. Carries a DeviceFaultContext.
DEVICE_OP.<command>.FAILED	One device call failed inside an action.
GROUP.<group_id>.COMPLETED	A multi-group submission's group finishes. COMPLETED is the only GROUP status emitted.
SLOT.<slot_key>.<status>	Receiver-slot lifecycle: OVERFLOW, REJECTED, AWAITING_DECISION, ORPHANED, ABANDONED.
TIP_RACK.EMPTY / TIP_RACK.LOW	The co-labware coordinator detected tip exhaustion.
OPERATOR.INSTRUCTION	An action called ctx.manual_step(...).
CUSTOM.<name>.EMITTED	An action, method or thread called ctx.emit(...). See below.

Incidents and device faults are the two an integration usually cares about first. Neither used to reach the stream at all, so a UI polling the incident store was the only way to see a fault.

The RuntimeEvent shape

System-level sinks, and the SSE stream, deliver RuntimeEvent objects:

class RuntimeEvent(BaseModel):
    model_config = ConfigDict(frozen=True, extra="forbid")

    event_name: str          # the full name
    execution_id: str
    timestamp: float
    entity_type: str         # parsed from event_name
    entity_id: str           # parsed from event_name (empty for a 2-part name)
    status: str              # parsed from event_name
    context: ExecutionContext  # type-specific payload

Events that belong to no execution (an incident with no execution, a device fault outside a run) carry the system id in execution_id.

event.context carries the per-event payload. The concrete type depends on the entity:

• LocationActionExecutionContext / MoveActionExecutionContext for action events
• MethodExecutionContext for method events
• ThreadExecutionContext for thread events, and ManualInterventionContext when the thread is awaiting a manual place or remove
• ExecutionLifecycleContext, SubmissionExecutionContext, GroupLifecycleContext for the lifecycle families
• IncidentContext for INCIDENT.*, DeviceFaultContext for DEVICE.*
• CustomEventContext for CUSTOM.*
• OperatorInstructionContext for OPERATOR.INSTRUCTION

to_dict() (equivalently model_dump(mode="json")) gives the JSON wire shape, and that is what the SSE stream emits.

A terminal thread event keeps the why

ThreadExecutionContext carries pause_reason and last_error. The engine fills last_error on PAUSED, FAILED and ABORTED. That matters because the execution-record sink upserts on every thread transition. Without the error on the terminal event, the terminal upsert would overwrite the stored cause with None, and the record would say a thread aborted without saying what for.

Subscribing

From inside a workflow

@orca.workflow(name="my_assay")
def my_assay(wf: WorkflowContext):
    wf.start(plate_thread)
    wf.on("ACTION.COMPLETED", AuditLogger())

wf.on(...) accepts:

• A callable: def handler(event_name: str, context: ExecutionContext) -> None
• A SystemBoundEventHandler subclass with a handle(event_name, context) method, useful when the handler needs the bound System for cross-entity lookups

Handler exceptions are caught and logged to the orca logger. One handler failing does not stop other handlers or the runtime. A persistent handler exception fires a PLUGIN_HANDLER_EXCEPTION or EVENT_HANDLER_EXCEPTION incident.

From outside the workflow (sinks)

For runtime-level integration:

from orca.runtime.interfaces import IEventSink
from orca.events.runtime_event import RuntimeEvent

class JsonlSink:
    def __init__(self, path: str):
        self._fp = open(path, "a")

    def on_event(self, event: RuntimeEvent) -> None:
        self._fp.write(event.model_dump_json() + "\n")
        self._fp.flush()

sink = JsonlSink("/var/log/orca/events.jsonl")
runtime.register_sink(sink)
...
runtime.unregister_sink(sink)

IEventSink is a one-method Protocol: on_event(event: RuntimeEvent) -> None. Register as many sinks as you like; they all receive every event. unregister_sink detaches one again, so a transient consumer (a disconnecting UI, a scoped test waiter) does not leak a listener for the runtime's lifetime.

Sinks are useful for:

• Audit trails: JSONL on disk, append-only.
• Live dashboards: fan out over WebSocket to UI clients.
• External orchestration: feed events into a downstream system.
• Persistence and indexing: a hosted deployment's sink writes JSONL and maintains a database index for searchability.

The runtime registers one sink of its own at start(): ExecutionTrackingSink, which mirrors submission, thread and execution events into the execution-record store. It is registered whether or not you injected the record service, because the runtime owns event delivery either way.

Reading events back

events = runtime.get_events_since(timestamp)          # None means no filter
events = runtime.get_events_for_execution(execution_id)

Both return list[RuntimeEvent] from the runtime's in-process log. That log is unbounded: it grows for the life of the process and is only ever pruned when an execution is removed. A long-lived deployment should persist through a sink and read from its own store, not treat this log as the archive.

Backpressure and ordering

• on_event is called synchronously from the runtime tick. A sink that blocks the call blocks the runtime. Sinks that need expensive work (database writes, network) should hand off to a background queue or task.
• Within one sink, events arrive in the order the runtime emitted them. Across sinks, ordering is not synchronized.
• The daemon's SSE sink uses per-connection bounded queues (1000 events by default) and drops the oldest event when full, so a slow subscriber cannot back-pressure the runtime or affect other subscribers.

Daemon REST surfaces

Two endpoints in orca.daemon:

Endpoint	Use for
GET /events?since=<ts>&execution_id=<id>	Polling. Returns every event recorded at or after the timestamp; optionally filter to one execution.
GET /events/stream?execution_id=<id>	Server-Sent Events. Streams live; optional server-side filter.

The SSE stream uses one SseEventSink that persists across load and unload cycles, so your subscription stays alive when an operator unloads and loads a different topology.

SSE wire shape:

event: runtime_event
data: {"event_name": "THREAD.tid-2.STARTED", "execution_id": "...", ...}

The data payload is exactly RuntimeEvent.to_dict().

Custom events from actions

Inside an action, emit a custom event:

@orca.action(device=reader, inputs=[plate])
async def read_plate(ctx: ActionContext):
    result = await ctx.device().read("absorbance.pro")
    await ctx.emit("plate_reading", value="complete", data={"absorbance": result})

Other actions wait for it:

@orca.action(device=evaluator, inputs=[plate])
async def evaluate(ctx: ActionContext):
    value, data = await ctx.wait_for("plate_reading")
    ...

Labware threads can wait too. ThreadContext.wait_for uses consumed semantics so looping sees each new publish.

Workflows can branch:

yield orca.branch("plate_reading", {
    "complete": [evaluate_step],
    "failed": [retry_step],
})

A custom emit also reaches the runtime event surface. The same call that unblocks a wait_for fires CUSTOM.<name>.EMITTED with a CustomEventContext, so a registered sink and the SSE stream both see it without any extra wiring. Dots in your event name are replaced with underscores on the way out, because the bus grammar owns the dot: qc.result arrives as CUSTOM.qc_result.EMITTED, with the original name preserved in the context. Use . freely for namespacing in the name you wait on; just do not expect it in the wire event name.

Pitfalls

• on_event runs synchronously on the runtime tick. Do not block. Persistence sinks should queue or offload.
• The in-process event log has no size cap. Use a sink for anything long-lived.
• 2-part subscribers see ALL 3-part fans of the same status. Subscribing to ACTION.COMPLETED fires for every action across every thread. If you only want one entity's events, subscribe to the 3-part form.
• wf.on(...) is per-workflow, not global. It fires only for events emitted by executions of THIS workflow template. For cross-workflow integration use a sink.
• Handler exceptions are caught and logged; they do not bubble. If a handler is misbehaving, an EVENT_HANDLER_EXCEPTION incident fires. Acknowledging the incident does NOT disable the handler; fix the handler in code.
• SSE subscriber queues are bounded. A subscriber that cannot keep up loses events (oldest dropped). If you need lossless capture, use a sink with backpressure-aware queuing, not the SSE stream.
• ctx.emit is async. Awaiting it is required; a missed await silently fails to emit.
• OPERATOR.CONFIRM.<step_id> is not an event. It is an internal wait channel the manual-step confirmation resolves. Subscribing to it gets you nothing; watch OPERATOR.INSTRUCTION and the thread's status instead.

See also

• Stores: the IEventSink Protocol and where it sits among the other store interfaces.
• Context API: emit and wait_for semantics on each context.
• Recovery: the incident model behind INCIDENT.*.
• Runtime: wiring sinks at the runtime level.
• Glossary: every term on this page, in one place.