Plugin System

The plugin layer is a first-class framework component. It allows domain-specific customization of stage behavior — prompt sections, output schema enum values, threat model shortcuts, reproduction strategies, and patch policies — without forking or replacing stage implementations.

Location

src/plugins/

`StagePlugin` protocol

Any plugin must implement this protocol:

class StagePlugin(Protocol):
    id: str
    version: str
    description: str
    capabilities: tuple[PluginCapability, ...]

    def can_handle(self, input: PluginSelectionInput) -> bool: ...

    def before_stage(
        self, context: RunContext, stage: StageName, input: StageInput
    ) -> StageInput: ...

    def build_prompt_sections(
        self, context: RunContext, stage: StageName, input: StageInput
    ) -> tuple[PromptSection, ...]: ...

    def output_schema_overrides(
        self, context: RunContext, stage: StageName
    ) -> tuple[SchemaEnumOverride, ...]: ...

    def build_static_threat_model(
        self, context: RunContext, input: StageInput,
        source_snapshot: Mapping[str, Any], plan: Mapping[str, Any]
    ) -> Mapping[str, Any] | None: ...

    def after_stage(
        self, context: RunContext, stage: StageName,
        input: StageInput, output: StageOutput
    ) -> StageOutput: ...

    def map_stage_error(
        self, context: RunContext, stage: StageName, error: Exception
    ) -> StageError | None: ...

    def repro_policy(
        self, context: RunContext, bug: Mapping[str, Any]
    ) -> Mapping[str, Any]: ...

    def patch_policy(
        self, context: RunContext, bug: Mapping[str, Any]
    ) -> Mapping[str, Any]: ...

`PluginCapability`

Each plugin declares what it can handle:

@dataclass(frozen=True)
class PluginCapability:
    name: str
    languages: tuple[str, ...] = ()
    file_patterns: tuple[str, ...] = ()
    metadata: Mapping[str, Any] = field(default_factory=dict)

Capabilities are used by PluginSelectionService to deterministically match plugins to repositories.

`PluginSelectionInput`

@dataclass(frozen=True)
class PluginSelectionInput:
    repo_id: str
    run_id: str
    source_snapshot: ArtifactRef
    plan_artifact: ArtifactRef | None
    policy: RunPolicy
    metadata: Mapping[str, Any] = field(default_factory=dict)

`PluginContext`

The result of plugin selection. Carried on RunContext during the entire run:

@dataclass(frozen=True)
class PluginContext:
    plugin_id: str
    plugin_version: str
    plugin: StagePlugin
    selection_reason: str
    metadata: Mapping[str, Any] = field(default_factory=dict)

Stages access the selected plugin through context.plugins.

`PluginRegistry`

class PluginRegistry:
    def __init__(self, plugins: tuple[StagePlugin, ...]) -> None: ...
    def all(self) -> tuple[StagePlugin, ...]: ...
    def get(self, plugin_id: str) -> StagePlugin | None: ...

The production registry is production_plugin_registry():

PluginRegistry((LinuxPlugin(), WebApplicationPlugin(), DefaultPlugin()))

Priority order matters for deterministic matching. DefaultPlugin must always be last.

`PluginSelectionService` — three-tier selection

PluginSelectionService combines three strategies, evaluated in order:

class PluginSelectionService:
    def __init__(self, registry: PluginRegistry) -> None: ...
    def select(self, input: PluginSelectionInput, context: RunContext) -> PluginContext: ...

Tier 1 — Explicit override

If plugin_id or plugin appears in ScanRequest.metadata or RunPolicy.metadata, that plugin is selected directly without calling can_handle. Use this for tests or when backend has already determined the appropriate plugin.

Tier 2 — Deterministic `can_handle` match

Each non-default plugin is called with can_handle(input). The first plugin that returns True wins. This is synchronous and does not require an agent call.

Tier 3 — LLM agent fallback

When no deterministic match is found and non-default plugins exist, an agent task with purpose="plugin_selection" is run. The agent inspects the source snapshot summary and ranked plugin descriptors and returns a selected_plugin_id with a confidence score. Low-confidence or failed agent results fall back to DefaultPlugin.

Selection events

All three paths emit events through the context event bus:

Event kind	When
`plugin_selection_started`	Selection begins
`plugin_selection_completed`	Plugin selected
`plugin_selection_fallback`	LLM fallback used

The selected PluginContext is stored on RunContext and the plugin id/version is included in CoreResult.data.

Built-in plugins

`DefaultPlugin`

@dataclass(frozen=True)
class DefaultPlugin:
    id: str = "default"
    version: str = "1"
    description: str = "..."

can_handle always returns True. All hook methods return empty/no-op results. Used when no domain-specific plugin matches.

`LinuxPlugin`

@dataclass(frozen=True)
class LinuxPlugin:
    id: str = "linux"
    version: str = "1"

can_handle returns True for repositories with Linux kernel source indicators (kernel header patterns, Kconfig, Makefile with kernel build targets, etc.).

Hooks:

build_prompt_sections(AUDIT) → Linux-specific vulnerability focus: memory safety, privilege escalation, syscall interfaces, race conditions.
output_schema_overrides(AUDIT) → Replaces bug_class enum with kernel bug classes (use_after_free, double_free, race_condition, integer_overflow, privilege_escalation, oob_read, oob_write, etc.).
repro_policy(...) → Returns kernel-specific reproduction guidance.
patch_policy(...) → Returns kernel-appropriate patch constraints.

`WebApplicationPlugin`

@dataclass(frozen=True)
class WebApplicationPlugin:
    id: str = "web_application"
    version: str = "1"

can_handle returns True for repositories with web framework indicators (e.g. package.json, requirements.txt with Django/Flask/Express, route handler file patterns, HTML/template files).

Hooks:

build_static_threat_model(...) → Returns a pre-built web application threat model (OWASP Top 10 components, browser/attacker profiles, session/cookie trust boundaries). This skips the threat-model agent call entirely.
build_prompt_sections(AUDIT) → Web security audit guidance from audit_prompt.txt (injection sinks, auth boundaries, CSRF/SSRF patterns, deserialization risks).
output_schema_overrides(AUDIT) → Replaces bug_class, attacker_profile, required_state, and required_services with web-specific values.

How `BaseStage` invokes plugin hooks

BaseStage.run(...) orchestrates plugin hook calls around execute(...):

before_stage(context, stage, input)
  → possibly augmented StageInput

output_schema_overrides(context, stage)
  → apply_schema_enum_overrides(base_schema, overrides)
  → effective output schema passed to agent

execute(context, effective_input)
  → calls build_prompt_sections(), builds_static_threat_model() if applicable

after_stage(context, stage, input, output)
  → possibly post-processed StageOutput

map_stage_error(context, stage, error)
  → possibly translated StageError

Writing a custom plugin

@dataclass(frozen=True)
class MyPlugin:
    id: str = "my_plugin"
    version: str = "1"
    description: str = "Custom domain plugin."
    capabilities: tuple[PluginCapability, ...] = (
        PluginCapability(
            name="my_framework",
            languages=("python",),
            file_patterns=("my_framework.conf",),
        ),
    )

    def can_handle(self, input: PluginSelectionInput) -> bool:
        # inspect source snapshot or plan artifact
        return True  # or False

    def build_prompt_sections(self, context, stage, input):
        if stage == StageName.AUDIT:
            return (PromptSection(title="My Domain", content="..."),)
        return ()

    def output_schema_overrides(self, context, stage):
        if stage == StageName.AUDIT:
            return (SchemaEnumOverride(
                stage=stage,
                field="bug_class",
                values=("my_class_a", "my_class_b"),
                mode="replace",
            ),)
        return ()

    def build_static_threat_model(self, context, input, source_snapshot, plan):
        return None  # use agent-driven threat model

    def before_stage(self, context, stage, input): return input
    def after_stage(self, context, stage, input, output): return output
    def map_stage_error(self, context, stage, error): return None
    def repro_policy(self, context, bug): return {}
    def patch_policy(self, context, bug): return {}

Register it in the runtime:

from plugins.registry import PluginRegistry
from plugins.builtin import DefaultPlugin

registry = PluginRegistry((MyPlugin(), DefaultPlugin()))

Plugin System

Location

StagePlugin protocol

PluginCapability

PluginSelectionInput

PluginContext

PluginRegistry

PluginSelectionService — three-tier selection