Orchestrator & Pipeline

The StageOrchestrator is the central controller that assembles stage sequences, wires dependencies, selects plugins, and returns typed results. The pipeline itself is expressed as a sequence of StageName values plus a SequenceTransition object that owns inter-stage data handoffs.

`StageOrchestrator`

class StageOrchestrator:
    def __init__(
        self,
        *,
        event_bus: EventBus,
        artifact_store: ArtifactStore,
        cancellation: CancellationManager,
        agent_runner: AgentRunner,
        plugin_resolver: PluginResolver,
        triage: SharedTriageService,
        policy: RunPolicy,
        stages: Mapping[StageName, BaseStage] | None = None,
    ) -> None: ...

    def plan_repo(self, request: PlanRequest) -> CoreResult: ...
    def run_scan(self, request: ScanRequest) -> CoreResult: ...
    def verify_and_patch(self, request: VerificationRequest) -> CoreResult: ...

build_stage_orchestrator(config) in runtime/config.py constructs the default instance from a RuntimeConfig.

What the orchestrator owns

Choose which stages run for each public API call.
Select the plugin context before scan execution (via PluginSelectionService).
Select the concrete threat-model and audit stage implementations from ScanExecution.threat_model_strategy and ScanExecution.audit_strategy.
Validate and reuse cached threat-model artifacts.
Pass audit resume artifacts and component shard assignments into scan state.
Create RunContext and share it across all stages in a sequence.
Maintain PipelineState across stages and apply transition logic.
Send raw findings to SharedTriageService after audit.
Stop immediately on cancellation or failure.
Normalize final output into CoreResult.

`PipelineState`

An immutable snapshot of shared state across stages in one sequence. Transition objects transform it between stages.

@dataclass(frozen=True)
class PipelineState:
    repo_id: str
    run_id: str
    source_snapshot: ArtifactRef | None = None
    plan_artifact: ArtifactRef | None = None
    selected_plugin: PluginContext | None = None
    artifacts: Mapping[str, ArtifactRef] = field(default_factory=dict)
    data: Mapping[str, Any] = field(default_factory=dict)
    raw_findings: tuple[Mapping[str, Any], ...] = ()
    deduplicated_bugs: tuple[Mapping[str, Any], ...] = ()
    verification: Mapping[str, Any] | None = None

`SequenceTransition` protocol

Each workflow uses a different transition implementation. The transition owns which stage inputs come from which state fields, and how each stage output updates the state.

class SequenceTransition(Protocol):
    def build_input(self, stage: StageName, state: PipelineState) -> StageInput: ...
    def apply_output(
        self, stage: StageName, state: PipelineState, output: StageOutput
    ) -> PipelineState: ...

The orchestrator's inner loop:

state = initial_state
for stage_name in sequence:
    stage = self.stages[stage_name]
    stage_input = transition.build_input(stage_name, state)
    stage_output = stage.run(context, stage_input)
    state = transition.apply_output(stage_name, state, stage_output)

Three workflows and their transitions

Plan workflow — `PlanTransition`

sequence: [PLAN]

PlanTransition maps:

Input: source_snapshot from state
Output: updates artifacts["plan"] in state

Scan workflow — `ScanTransition`

sequence: [PLAN, THREAT_MODEL, AUDIT]   # when plan_artifact not provided
sequence: [THREAT_MODEL, AUDIT]          # when plan_artifact provided
sequence: [AUDIT]                        # when threat model cache is valid

Plugin selection runs before this sequence starts. The orchestrator chooses the concrete threat-model and audit stage implementations from ScanExecution strategy fields. ScanTransition handles all three stage names (PLAN, THREAT_MODEL, AUDIT) in its build_input/apply_output methods.

ScanTransition maps:

THREAT_MODEL input: source_snapshot, plan_artifact, include_paths, bug_classes, depth, threat_model_strategy, selected_plugin metadata
THREAT_MODEL output: saves artifacts["threat_model"] and data["threat_model_summary"] in state
AUDIT input: source_snapshot, plan_artifact, artifacts["threat_model"], threat_model_summary, audit_strategy, audit_resume_policy, depth, variant_context, assigned_component_ids, audit_shard_id, selected_plugin metadata
AUDIT output: calls SharedTriageService.ingest(...), stores raw_findings, deduplicated_bugs, artifacts["raw_findings"], artifacts["audit_component_state"], and artifacts["deduplicated_bugs"]

Verify/patch workflow — `VerifyPatchTransition`

sequence: [VERIFY, PATCH]

VerifyPatchTransition maps:

VERIFY input: source_snapshot, plan_artifact, deduplicated_bugs, selected_plugin metadata
VERIFY output: saves verification mapping and artifacts["verification_evidence"] in state
PATCH input: source_snapshot, plan_artifact, artifacts["verification_evidence"], bugs, verification, selected_plugin metadata
PATCH output: saves artifacts["patch"] in state

Plugin selection

For run_scan, the orchestrator calls PluginSelectionService.select(...) to choose a PluginContext before building the RunContext. The selection uses a three-tier strategy:

Explicit override — if plugin_id or plugin appears in scan metadata or RunPolicy.metadata.
Deterministic can_handle match — each non-default plugin is asked whether it can handle the selection input; first match wins.
LLM agent fallback — when no deterministic match is found, an agent task inspects the source summary and plugin descriptors and returns a plugin id. Low-confidence results fall back to DefaultPlugin.

The selected plugin id/version is included in CoreResult.data so backends can pass it to subsequent verify_and_patch calls.

For verify_and_patch, the orchestrator does not re-resolve the plugin. Instead it restores the original plugin id/version from request metadata onto a default plugin object.

Stage sequence summary

API call	Stage sequence	Transition
`plan_repo`	`[PLAN]`	`PlanTransition`
`run_scan`	`[PLAN, THREAT_MODEL, AUDIT]` when no `plan_artifact`; `[THREAT_MODEL, AUDIT]` when `plan_artifact` provided; `[AUDIT]` when threat model cache valid	`ScanTransition`
`verify_and_patch`	`[VERIFY, PATCH]`	`VerifyPatchTransition`

Data flow diagram

flowchart TD
    subgraph plan_repo
        PR[PlanRequest] --> PS[PlanStage]
        PS --> PA[vega.plan artifact]
    end

    subgraph run_scan
        SR[ScanRequest] --> Plugin[PluginSelectionService]
        Plugin --> TM[ThreatModelStage]
        PA2[vega.plan artifact] --> TM
        TM --> TMA[vega.threat_model artifact]
        TMA --> AS[AuditStage]
        AS --> RF[vega.raw_findings + vega.audit_component_state]
        RF --> Triage[SharedTriageService]
        Triage --> Bugs[deduplicated_bugs]
    end

    subgraph verify_and_patch
        VR[VerificationRequest] --> VS[VerifyStage]
        Bugs2[deduplicated_bugs] --> VS
        VS --> VE[vega.verification_results]
        VE --> Patch[PatchStage]
        Patch --> PA3[vega.remediation_results]
    end

Strategy resolution

Threat model strategies

`ScanExecution.threat_model_strategy`	Concrete stage
`"codex_direct"` or `"direct"`	`CodexDirectThreatModelStage`
`"llm_bug_class_selection"` or `"bug_class"`	`BugClassThreatModelStage`
`"default"` or `"none"`	`DefaultThreatModelStage`

Audit strategies

`ScanExecution.audit_strategy`	Concrete stage
`"default"` or `"component"`	`DefaultAuditStage`
`"variant_analysis"` or `"variant"`	`VariantAnalysisAuditStage`