Plugins

PACE has a plugin system that lets you extend pipeline behaviour without modifying the core framework. Plugins are Python packages discovered via setuptools entry points.

Plugin types

Base class	Entry point group	Purpose
HookBase	pace.hooks	Run Python code before or after any pipeline event
WebhookInBase	pace.webhook_in	Accept inbound HTTP requests on the PACE webhook server (port 9876)
WebhookOutBase	pace.webhook_out	Send outbound HTTP requests when pipeline events fire
PluginBase	pace.plugins	General-purpose plugin combining hooks and web handlers

Additional entry point groups: pace.llm_providers, pace.platforms, pace.reporters.

---

Hook events

Hooks are invoked at these points in the pipeline:

Event constant	Fires
BEFORE_DAY	Before any agent runs on day N
AFTER_DAY	After all agents complete on day N
BEFORE_FORGE	Before FORGE starts its tool-calling loop
AFTER_FORGE	After FORGE calls complete_handoff
BEFORE_GATE	Before GATE evaluates acceptance criteria
AFTER_GATE	After GATE produces its report
BEFORE_SENTINEL	Before SENTINEL runs its security review
AFTER_SENTINEL	After SENTINEL produces its report
BEFORE_CONDUIT	Before CONDUIT runs its DevOps review
AFTER_CONDUIT	After CONDUIT produces its report

---

Writing a hook plugin

Create a Python package with a class that extends HookBase:

my_pace_plugin/hook.py

from plugins.base import HookBase, AFTER_GATE

class SlackGateHook(HookBase):
    """Post a message to Slack whenever GATE produces a report."""

    events = [AFTER_GATE]

    def run(self, event: str, context: dict) -> None:
        gate_report = context.get("gate_report", {})
        decision = gate_report.get("gate_decision", "unknown")
        story = context.get("story_title", "unknown story")

        import requests
        requests.post(
            self.config.get("webhook_url"),
            json={"text": f"GATE {decision} — {story}"},
            timeout=5,
        )

Register it in pyproject.toml:

[project.entry-points."pace.hooks"]
slack_gate = "my_pace_plugin.hook:SlackGateHook"

Install the package in PACE’s virtualenv:

pip install -e path/to/my_pace_plugin

---

Writing an inbound webhook plugin

Inbound webhook plugins receive HTTP POST requests on the PACE webhook server (default port 9876).

my_pace_plugin/webhook_in.py

from plugins.base import WebhookInBase

class GitHubWebhookIn(WebhookInBase):
    """Accept GitHub push events and trigger a pipeline run."""

    path = "/github/push"

    def handle(self, payload: dict, headers: dict) -> dict:
        ref = payload.get("ref", "")
        if ref == "refs/heads/main":
            self.trigger_pipeline(day=None)  # run next pending day
        return {"status": "accepted"}

Register it:

[project.entry-points."pace.webhook_in"]
github_push = "my_pace_plugin.webhook_in:GitHubWebhookIn"

---

Writing an outbound webhook plugin

Outbound webhook plugins fire when pipeline events occur and send HTTP requests to external systems.

my_pace_plugin/webhook_out.py

from plugins.base import WebhookOutBase, AFTER_DAY

class PagerDutyOut(WebhookOutBase):
    """Create a PagerDuty incident when a day is held."""

    events = [AFTER_DAY]

    def should_fire(self, event: str, context: dict) -> bool:
        return context.get("day_decision") == "HOLD"

    def build_payload(self, event: str, context: dict) -> dict:
        return {
            "routing_key": self.config["routing_key"],
            "event_action": "trigger",
            "payload": {
                "summary": f"PACE HOLD: {context.get('hold_reason')}",
                "severity": "error",
                "source": "pace-pipeline",
            },
        }

    def endpoint(self) -> str:
        return "https://events.pagerduty.com/v2/enqueue"

---

Configuring plugins

Enable the plugin directory in pace/pace.config.yaml:

plugins:
  dirs:
    - "plugins/"        # scanned for local plugins
  webhook_port: 9876    # port for inbound webhook server

Pass configuration values to a specific plugin:

plugins:
  config:
    slack_gate:
      webhook_url: "${SLACK_WEBHOOK_URL}"
    pagerduty_out:
      routing_key: "${PAGERDUTY_ROUTING_KEY}"

Plugin config is passed as self.config inside the plugin class.

---

Plugin discovery order

1. Installed packages with matching entry points (scanned at startup)
2. Local directories listed in plugins.dirs (each must contain a pace_plugin.py module or a package with __init__.py)

Plugins are loaded in discovery order. If two plugins register the same hook event, both run sequentially.