ZERG Plugin System

ZERG Plugin System

Extend ZERG with custom quality gates, lifecycle hooks, and worker launchers.

Overview

The ZERG plugin system provides three extension points:

1. Quality Gate Plugins — Custom validation after merges (lint, security scans, benchmarks)
2. Lifecycle Hook Plugins — React to events (task starts/completes, level finishes, merges)
3. Launcher Plugins — Custom worker execution environments (Kubernetes, SSH clusters, cloud VMs)

All plugins are additive only — they cannot mutate orchestrator state, only observe and react. Plugin failures never crash the orchestrator; they are logged and execution continues.

Quick Start

Simple: YAML Hooks

Add shell commands triggered by lifecycle events in .zerg/config.yaml:

plugins:
  enabled: true
  hooks:
    - event: task_completed
      command: echo "Task completed at $(date)"
      timeout: 60
    - event: level_complete
      command: ./scripts/notify-slack.sh "Level {level} done"
      timeout: 120

Simple: YAML Quality Gates

Add custom quality gate commands:

plugins:
  quality_gates:
    - name: security-scan
      command: bandit -r src/ --severity medium
      required: false
      timeout: 300
    - name: complexity-check
      command: radon cc src/ --min B
      required: true
      timeout: 120

Advanced: Python Entry Points

For complex logic, authentication, or API calls, create Python plugin classes:

# my_zerg_plugins/gates.py
from zerg.plugins import QualityGatePlugin, GateContext
from zerg.types import GateRunResult
from zerg.constants import GateResult

class SonarQubeGate(QualityGatePlugin):
    @property
    def name(self) -> str:
        return "sonarqube"

    def run(self, ctx: GateContext) -> GateRunResult:
        # ctx.feature, ctx.level, ctx.cwd, ctx.config available
        return GateRunResult(
            gate_name=self.name,
            result=GateResult.PASS,
            command="sonarqube-api",
            exit_code=0,
            stdout="Quality gate passed",
        )

Register in pyproject.toml:

[project.entry-points."zerg.plugins"]
sonarqube = "my_zerg_plugins.gates:SonarQubeGate"

Install and run:

pip install -e .
/zerg:rush  # Plugins auto-discovered via entry points

Plugin Types

QualityGatePlugin

Validates code quality after each level merge. Runs sequentially after built-in gates.

from zerg.plugins import QualityGatePlugin, GateContext
from zerg.types import GateRunResult
from zerg.constants import GateResult

class MyCustomGate(QualityGatePlugin):
    @property
    def name(self) -> str:
        return "my-custom-gate"

    def run(self, ctx: GateContext) -> GateRunResult:
        # ctx.feature: str — feature name
        # ctx.level: int — level just merged
        # ctx.cwd: Path — working directory
        # ctx.config: ZergConfig — full config

        return GateRunResult(
            gate_name=self.name,
            result=GateResult.PASS,  # PASS | FAIL | SKIP | TIMEOUT | ERROR
            command="my-custom-check",
            exit_code=0,
            stdout="All checks passed",
            stderr="",
        )

Execution flow: Built-in gates (lint, build, test) → Plugin gates (registered order) → Merge completes

LifecycleHookPlugin

Observes ZERG lifecycle events without blocking execution.

from zerg.plugins import LifecycleHookPlugin, LifecycleEvent

class MyNotificationHook(LifecycleHookPlugin):
    @property
    def name(self) -> str:
        return "slack-notifier"

    def on_event(self, event: LifecycleEvent) -> None:
        # event.event_type: str — from PluginHookEvent enum
        # event.data: dict — event-specific payload
        # event.timestamp: datetime — when event occurred

        if event.event_type == "task_completed":
            task_id = event.data["task_id"]
            # Send notification...

Available events:

LauncherPlugin

Provides custom worker execution environments beyond subprocess and container.

from zerg.plugins import LauncherPlugin
from zerg.launcher import WorkerLauncher

class K8sLauncherPlugin(LauncherPlugin):
    @property
    def name(self) -> str:
        return "kubernetes"

    def create_launcher(self, config) -> WorkerLauncher:
        return K8sWorkerLauncher(config)

The returned WorkerLauncher subclass must implement launch(), wait(), and cleanup().

Configuration

YAML Hooks

plugins:
  hooks:
    - event: task_completed
      command: echo "Task completed at $(date)"
      timeout: 60

Variable substitution: {level}, {feature}, {task_id}, {worker_id} are replaced from event data.

YAML Quality Gates

plugins:
  quality_gates:
    - name: security-scan
      command: bandit -r src/ --severity medium
      required: false
      timeout: 300

Python Entry Points

Register plugin classes in your package's pyproject.toml:

[project.entry-points."zerg.plugins"]
sonarqube = "my_zerg_plugins.gates:SonarQubeGate"
k8s-launcher = "my_zerg_plugins.launchers:K8sLauncherPlugin"
slack-hooks = "my_zerg_plugins.hooks:SlackNotificationHook"

Discovery happens automatically via PluginRegistry.load_entry_points("zerg.plugins") on orchestrator startup.

Security Model

Read-Only State Access

Plugins receive immutable views of orchestrator state:

• GateContext — read-only with feature, level, cwd, config
• LifecycleEvent — read-only with event_type, data, timestamp
• No access to Orchestrator._state, _workers, or _task_queue

Timeout Enforcement

• YAML hooks: 1-600 seconds (default: 60)
• YAML gates: 1-3600 seconds (default: 300)
• Python plugins: Must complete within timeout or killed

Exception Isolation

Each plugin invocation is individually wrapped in try/except. A failing plugin never crashes the orchestrator.

No Shell Injection

YAML commands are parsed with shlex.split and executed via subprocess.run(shell=False).

Examples

Slack Notifications

YAML:

plugins:
  hooks:
    - event: level_complete
      command: |
        curl -X POST https://hooks.slack.com/services/YOUR/WEBHOOK/URL \
          -H 'Content-Type: application/json' \
          -d '{"text":"Level {level} completed for feature {feature}"}'
      timeout: 30

Python (for OAuth, custom formatting):

from zerg.plugins import LifecycleHookPlugin, LifecycleEvent
from slack_sdk import WebClient

class SlackNotifier(LifecycleHookPlugin):
    def __init__(self):
        self.client = WebClient(token=os.getenv("SLACK_BOT_TOKEN"))

    @property
    def name(self) -> str:
        return "slack-notifier"

    def on_event(self, event: LifecycleEvent) -> None:
        if event.event_type == "level_complete":
            self.client.chat_postMessage(
                channel="#zerg-builds",
                text=f"Level {event.data['level']} completed",
            )

Security Gate (Trivy)

from zerg.plugins import QualityGatePlugin, GateContext
from zerg.types import GateRunResult
from zerg.constants import GateResult
import subprocess

class TrivyScanGate(QualityGatePlugin):
    @property
    def name(self) -> str:
        return "trivy-scan"

    def run(self, ctx: GateContext) -> GateRunResult:
        result = subprocess.run(
            ["trivy", "fs", "--severity", "HIGH,CRITICAL", str(ctx.cwd)],
            capture_output=True, text=True, timeout=300
        )
        return GateRunResult(
            gate_name=self.name,
            result=GateResult.PASS if result.returncode == 0 else GateResult.FAIL,
            command="trivy fs",
            exit_code=result.returncode,
            stdout=result.stdout,
            stderr=result.stderr,
        )

Kubernetes Launcher

from zerg.plugins import LauncherPlugin
from zerg.launcher import WorkerLauncher
from kubernetes import client, config

class K8sLauncher(WorkerLauncher):
    def __init__(self, zerg_config):
        super().__init__(zerg_config)
        config.load_kube_config()
        self.api = client.CoreV1Api()

    def launch(self, worker_id: str, task_id: str) -> dict:
        pod = client.V1Pod(
            metadata=client.V1ObjectMeta(name=f"zerg-{worker_id}"),
            spec=client.V1PodSpec(
                containers=[client.V1Container(
                    name="worker",
                    image="claude-code:latest",
                    env=[{"name": "TASK_ID", "value": task_id}]
                )]
            )
        )
        self.api.create_namespaced_pod(namespace="zerg", body=pod)
        return {"pod_name": f"zerg-{worker_id}"}

    def wait(self, launch_info: dict) -> int:
        # Poll pod status...
        pass

    def cleanup(self, launch_info: dict) -> None:
        self.api.delete_namespaced_pod(
            name=launch_info["pod_name"], namespace="zerg"
        )

class K8sLauncherPlugin(LauncherPlugin):
    @property
    def name(self) -> str:
        return "kubernetes"

    def create_launcher(self, config):
        return K8sLauncher(config)

Best Practices

Quality Gates

• Return GateResult.SKIP for gates that don't apply to the current level
• Use required: false for informational gates (warnings only)
• Set realistic timeouts — allow 3x typical execution time
• Capture full output in stdout/stderr for debugging

Lifecycle Hooks

• Keep hooks fast — they block event processing
• Use async operations for slow tasks (API calls, notifications)
• Never throw exceptions — return gracefully on errors

Launchers

• Implement proper cleanup — delete pods, VMs, temp resources
• Handle partial failures — launcher crash shouldn't orphan workers
• Support resume — launchers should be idempotent on retry

General

• Prefix all log messages with plugin name: logger.info(f"[{self.name}] ...")
• Version entry points for breaking changes: sonarqube-v2 = "pkg:GateV2"
• Test plugins with mock mode before production

Troubleshooting

Code References

• Plugin ABCs: zerg/plugins.py
• Plugin Registry: zerg/plugins.py
• Config Models: zerg/plugin_config.py
• Lifecycle Events: zerg/constants.py
• Integration Tests: tests/integration/test_plugin_lifecycle.py