Write plugins that collect hardware data without owning pass/fail logic.

Plugin lifecycle

Every plugin is called in the same fixed order once per job cycle:

init(config, ctx: WorkerContext)
    └── for each step:
        run_step(action, inputs, ctx: StepContext)
cleanup(ctx: WorkerContext | None)

Corvus always calls cleanup(), even when the job fails or is aborted. During worker shutdown, cleanup() may receive None.

Context objects

WorkerContext

Available in init() and cleanup(). Contains job-level information.

StepContext

Available in run_step(). Extends WorkerContext with step-specific fields.

StationContext

Describes the physical test station. Read-only.

TriggerContext

serial = ctx.trigger.data.get("serial", "")

RoutingContext

Snapshot of the routing selection active at trigger time.

entry_id is stable — it never changes even if the description is updated. Use it for fixture mapping or routing-specific configuration.

# Branch behavior by part number
if ctx.routing and ctx.routing.part_number == "AMP-100":
    self._gain_correction = 1.05

# Gate a step by revision
if ctx.routing and ctx.routing.revision == "Rev B":
    self._apply_offset_cal()

Plugin skeleton

from plugin_base import BasePlugin
from context import StepContext, WorkerContext


class MyPlugin(BasePlugin):

    plugin_id = "my_plugin"

    def init(self, config: dict, ctx: WorkerContext) -> None:
        self.logger.info(
            "init  job=%s  part=%s  serial=%s",
            ctx.job_id,
            ctx.routing.part_number if ctx.routing else "N/A",
            ctx.trigger.data.get("serial", "N/A"),
        )
        # Open hardware connections, apply config

    def run_step(self, action: str, inputs: dict, ctx: StepContext) -> Any:
        self._assert_action(action, ["measure_voltage"])

        if action == "measure_voltage":
            value = self._hw.read_voltage(inputs.get("channel", 0))
            # Return raw data only — never decide pass/fail here
            return {"voltage": value, "unit": "V"}

    def cleanup(self, ctx: WorkerContext | None) -> None:
        # Release hardware, close connections
        # Must not raise
        self.logger.info("cleanup complete")

Output contract

run_step() returns raw measurement data. Corvus passes it to the validation engine — the plugin never decides pass or fail.

Every run_step return value is stored in the report under steps[n].raw_data. The result, validation, and reason fields are always computed by Corvus — the plugin does not control them.

Validation types and required return shapes

Each validation type reads a specific key from raw_data. The key name is configured in the sequence file.

Boolean

# sequence config: { "type": "boolean", "key": "state", "expected": true }
return {"state": True}

Numeric

Operators: >, <, >=, <=, ==, !=, range.

# sequence config: { "type": "numeric", "key": "voltage", "operator": "range", "min": 3.1, "max": 3.5 }
return {"voltage": 3.29, "unit": "V"}  # extra keys are ignored

String

Modes: exact or regex.

# sequence config: { "type": "string", "key": "status", "mode": "exact", "expected": "OK" }
return {"status": "OK", "code": 0}  # extra keys are ignored

Array

Two parallel arrays for waveform or frequency-response validation. Modes: strict, interpolate, key_points.

return {
    "channel":       "left",
    "frequencies":   [20, 100, 1000, 10000, 20000],
    "amplitudes_db": [-0.2, -0.1, 0.0, -0.3, -0.5],
}

No validation

If the step's validation field is absent or empty, the step auto-passes. Raw data is still recorded for traceability.

Error handling

Corvus catches all exceptions from run_step, records the step as ERROR, and includes the original traceback in the worker log.

init() return value is ignored. Raise from init() to fail job startup.

cleanup() return value is ignored. Must not raise — catch and log errors internally.

Strict rules